Merge branch 'master' into IGNITE-28221-Combining-of-the-message-factories

# Conflicts:
#	modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
#	modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java
#	modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryCustomEventMessage.java
#	modules/core/src/test/java/org/apache/ignite/internal/util/distributed/MessagesPluginProvider.java

Author: Vladsz83

.mvn/wrapper/MavenWrapperDownloader.java

@@ -15,5 +15,6 @@
 # limitations under the License.
 #
 
+wrapperVersion=3.3.4
+distributionType=only-script
 distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.6/apache-maven-3.9.6-bin.zip
-wrapperUrl=https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar

.mvn/wrapper/maven-wrapper.jar

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.benchmarks.jmh.sql;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.cache.CacheInterceptorAdapter;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Benchmark cache with interceptor queries.
+ */
+@Fork(1)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Warmup(iterations = 5, time = 5)
+@Measurement(iterations = 10, time = 5)
+@State(Scope.Benchmark)
+public class JmhCacheWithInterceptorBenchmark extends JmhSqlAbstractBenchmark {
+    /** Query engine. */
+    @Param({"CALCITE"})
+    protected String engine;
+
+    /** Keep binary mode. */
+    @Param({"true", "false"})
+    protected boolean keepBinary;
+
+    /** {@inheritDoc} */
+    @Override protected CacheConfiguration<Integer, Item> cacheConfiguration() {
+        return super.cacheConfiguration().setInterceptor(new CacheInterceptorAdapter<>());
+    }
+
+    /** Test update operation. */
+    @Benchmark
+    public void update() {
+        int key = ThreadLocalRandom.current().nextInt(KEYS_CNT);
+
+        executeSql("UPDATE CACHE.Item SET fld = fld + 1 WHERE fldIdx=?", key);
+    }
+
+    /**
+     * Run benchmarks.
+     *
+     * @param args Args.
+     * @throws Exception Exception.
+     */
+    public static void main(String[] args) throws Exception {
+        final Options options = new OptionsBuilder()
+            .include(JmhCacheWithInterceptorBenchmark.class.getSimpleName())
+            .build();
+
+        new Runner(options).run();
+    }
+}

.mvn/wrapper/maven-wrapper.properties

@@ -145,6 +145,9 @@ public class ExecutionContext<Row> extends AbstractQueryContext implements DataC
     /** */
     private Object[] correlations = new Object[16];
 
+    /** Entries holder per execution thread. */
+    private static final ThreadLocal<Collection<QueryTxEntry>> txEntriesHolder = new ThreadLocal<>();
+
     /**
      * @param qctx Parent base query context.
      * @param qryId Query ID.
@@ -182,7 +185,7 @@ public ExecutionContext(
         this.ioTracker = ioTracker;
         this.params = params;
         this.timeout = timeout;
-        this.qryTxEntries = qryTxEntries;
+        this.qryTxEntries = qryTxEntries == null ? txEntriesHolder.get() : qryTxEntries;
 
         startTs = U.currentTimeMillis();
 
@@ -421,12 +424,17 @@ public void execute(RunnableX task, Consumer<Throwable> onError) {
 
         executor.execute(qryId, fragmentId(), () -> {
             try {
+                txEntriesHolder.set(qryTxEntries);
+
                 if (!isCancelled())
                     task.run();
             }
             catch (Throwable e) {
                 onError.accept(e);
             }
+            finally {
+                txEntriesHolder.remove();
+            }
         });
     }
 

modules/benchmarks/src/main/java/org/apache/ignite/internal/benchmarks/jmh/sql/JmhCacheWithInterceptorBenchmark.java

@@ -672,6 +672,9 @@ private ListFieldsQueryCursor<?> mapAndExecutePlan(
                     try {
                         SessionContextImpl sesCtx = qry.context().unwrap(SessionContextImpl.class);
 
+                        QueryProperties props = qry.context().unwrap(QueryProperties.class);
+                        boolean keepBinaryMode = props == null || props.keepBinary();
+
                         QueryStartRequest req = new QueryStartRequest(
                             qry.id(),
                             qry.localQueryId(),
@@ -684,7 +687,8 @@ private ListFieldsQueryCursor<?> mapAndExecutePlan(
                             parametersMarshalled,
                             timeout,
                             ectx.getQryTxEntries(),
-                            sesCtx == null ? null : sesCtx.attributes()
+                            sesCtx == null ? null : sesCtx.attributes(),
+                            keepBinaryMode
                         );
 
                         messageService().send(nodeId, req);
@@ -881,8 +885,13 @@ private void onMessage(UUID nodeId, final QueryStartRequest msg) {
                 )
             );
 
+            boolean keepBinaryMode = msg.keepBinaryMode();
+            QueryProperties qryProps = new QueryProperties(null, keepBinaryMode, false);
+
             final BaseQueryContext qctx = createQueryContext(
-                msg.applicationAttributes() == null ? Contexts.empty() : Contexts.of(new SessionContextImpl(msg.applicationAttributes())),
+                msg.applicationAttributes() == null ?
+                    Contexts.of(qryProps) :
+                    Contexts.of(new SessionContextImpl(msg.applicationAttributes()), qryProps),
                 msg.schema());
 
             FragmentPlan fragmentPlan = fragmentPlanCache.computeIfAbsent(msg.root(), k -> prepareFragment(qctx, k));

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionContext.java

@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.query.calcite.exec;
+
+import java.util.Map;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyDefinition;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexKeyType;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexPlainRowImpl;
+import org.apache.ignite.internal.cache.query.index.sorted.IndexRow;
+import org.apache.ignite.internal.cache.query.index.sorted.InlineIndexRowHandler;
+import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndex;
+import org.apache.ignite.internal.cache.query.index.sorted.keys.IndexKey;
+import org.apache.ignite.internal.cache.query.index.sorted.keys.IndexKeyFactory;
+import org.apache.ignite.internal.processors.query.QueryUtils;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.RangeIterable;
+import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.schema.ColumnDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
+import org.jetbrains.annotations.Nullable;
+
+/** Extension for column {@value QueryUtils#KEY_FIELD_NAME} in case of composite primary key. */
+public class IndexWrappedKeyScan<Row> extends IndexScan<Row> {
+    /** */
+    public IndexWrappedKeyScan(
+        ExecutionContext<Row> ectx,
+        CacheTableDescriptor desc,
+        InlineIndex idx,
+        ImmutableIntList idxFieldMapping,
+        int[] parts,
+        RangeIterable<Row> ranges,
+        @Nullable ImmutableBitSet requiredColumns
+    ) {
+        super(ectx, desc, idx, idxFieldMapping, parts, ranges, requiredColumns);
+    }
+
+    /** */
+    @Override protected IndexRow row2indexRow(Row bound) {
+        if (bound == null)
+            return null;
+
+        RowHandler<Row> rowHnd = ectx.rowHandler();
+
+        Object key = rowHnd.get(QueryUtils.KEY_COL, bound);
+        assert key != null : String.format("idxName=%s, bound=%s", idx.name(), Commons.toString(rowHnd, bound));
+
+        if (key instanceof BinaryObject)
+            return binaryObject2indexRow((BinaryObject)key);
+
+        throw new IgniteException(String.format(
+            "Unsupported type for index boundary: [expected=%s, current=%s]",
+            BinaryObject.class.getName(), key.getClass().getName()
+        ));
+    }
+
+    /** */
+    private IndexRow binaryObject2indexRow(BinaryObject o) {
+        assert o.type().typeName().equals(idx.indexDefinition().typeDescriptor().keyTypeName()) : String.format(
+            "idx=%s, o=%s, oType=%s, idxKeyType=%s",
+            idx.name(), o, o.type().typeName(), idx.indexDefinition().typeDescriptor().keyTypeName()
+        );
+
+        InlineIndexRowHandler idxRowHnd = idx.segment(0).rowHandler();
+        IndexKey[] keys = new IndexKey[idx.indexDefinition().indexKeyDefinitions().size()];
+
+        int i = 0;
+        for (Map.Entry<String, IndexKeyDefinition> e : idx.indexDefinition().indexKeyDefinitions().entrySet()) {
+            String keyName = e.getKey();
+
+            ColumnDescriptor fieldDesc = desc.columnDescriptor(keyName);
+            assert fieldDesc != null : String.format("idx=%s, o=%s, keyName=%s", idx.name(), o, keyName);
+
+            Object field = o.field(keyName);
+            Object key = TypeUtils.fromInternal(ectx, field, fieldDesc.storageType());
+
+            keys[i++] = wrapIndexKey(key, e.getValue().indexKeyType());
+        }
+
+        return new IndexPlainRowImpl(keys, idxRowHnd);
+    }
+
+    /** */
+    private IndexKey wrapIndexKey(Object key, IndexKeyType keyType) {
+        return IndexKeyFactory.wrap(key, keyType, cctx.cacheObjectContext(), idx.indexDefinition().keyTypeSettings());
+    }
+}