IGNITE-28277 CacheInterceptor need to take into account cache keepBinary mode (#12911)

I make a fix only for calcite related part, honestly - i afraid to make
an equal changes on h2 related part and prefer to store it - as is. If
approach is ok - i fill follow up issue for related documentation
change.

Author: zstan

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

@@ -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();
+    }
+}

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.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/rel/HashJoinNode.java

@@ -394,7 +394,7 @@ private AbstractMatchingHashJoin(
         private void downstreamPush(Row left, Row right) throws Exception {
             requested--;
 
-            downstream().push(outRowFactory.apply(left, right));;
+            downstream().push(outRowFactory.apply(left, right));
         }
 
         /** {@inheritDoc} */

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

@@ -31,9 +31,10 @@
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.cache.context.SessionContextImpl;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheProxyImpl;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
 import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
 import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.QueryProperties;
 import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
 import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
 import org.apache.ignite.internal.processors.query.calcite.schema.ModifyTuple;
@@ -200,9 +201,15 @@ private void flushTuples(boolean force) throws IgniteCheckedException {
         this.tuples = new ArrayList<>(MODIFY_BATCH_SIZE);
 
         GridCacheContext<Object, Object> cctx = desc.cacheContext();
-        GridCacheProxyImpl<Object, Object> cache = cctx.cache().keepBinary();
+        IgniteInternalCache<Object, Object> cache = cctx.cache();
         GridNearTxLocal tx = Commons.queryTransaction(context(), cctx.shared());
 
+        QueryProperties props = context().unwrap(QueryProperties.class);
+        boolean keepBinaryMode = props == null || props.keepBinary();
+
+        if (keepBinaryMode)
+            cache = cache.keepBinary();
+
         if (tx == null)
             invokeOutsideTransaction(tuples, cache);
         else
@@ -217,7 +224,7 @@ private void flushTuples(boolean force) throws IgniteCheckedException {
      */
     private void invokeOutsideTransaction(
         List<ModifyTuple> tuples,
-        GridCacheProxyImpl<Object, Object> cache
+        IgniteInternalCache<Object, Object> cache
     ) throws IgniteCheckedException {
         SessionContextImpl sesCtx = context().unwrap(SessionContextImpl.class);
         Map<String, String> sesAttrs = sesCtx == null ? null : sesCtx.attributes();
@@ -251,7 +258,7 @@ private void invokeOutsideTransaction(
      */
     private void invokeInsideTransaction(
         List<ModifyTuple> tuples,
-        GridCacheProxyImpl<Object, Object> cache,
+        IgniteInternalCache<Object, Object> cache,
         GridNearTxLocal userTx
     ) throws IgniteCheckedException {
         userTx.resume();

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/QueryStartRequest.java

@@ -79,6 +79,10 @@ public class QueryStartRequest implements CalciteMarshalableMessage, ExecutionCo
     @Order(10)
     @Nullable Map<String, String> appAttrs;
 
+    /** */
+    @Order(11)
+    boolean keepBinaryMode;
+
     /** */
     @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
     public QueryStartRequest(
@@ -93,7 +97,8 @@ public QueryStartRequest(
         @Nullable byte[] paramsBytes,
         long timeout,
         @Nullable Collection<QueryTxEntry> qryTxEntries,
-        @Nullable Map<String, String> appAttrs
+        @Nullable Map<String, String> appAttrs,
+        boolean keepBinaryMode
     ) {
         this.qryId = qryId;
         this.originatingQryId = originatingQryId;
@@ -107,6 +112,7 @@ public QueryStartRequest(
         this.timeout = timeout;
         this.qryTxEntries = qryTxEntries;
         this.appAttrs = appAttrs;
+        this.keepBinaryMode = keepBinaryMode;
     }
 
     /** */
@@ -199,6 +205,11 @@ public long timeout() {
         return appAttrs;
     }
 
+    /** */
+    public boolean keepBinaryMode() {
+        return keepBinaryMode;
+    }
+
     /** {@inheritDoc} */
     @Override public void prepareMarshal(GridCacheSharedContext<?, ?> ctx) throws IgniteCheckedException {
         if (paramsBytes == null && params != null)

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CancelTest.java

@@ -72,7 +72,7 @@ public class CancelTest extends GridCommonAbstractTest {
             .setKeyFieldName("id")
             .setValueFieldName("val")
             .addQueryField("id", Integer.class.getName(), null)
-            .addQueryField("val", String.class.getName(), null);;
+            .addQueryField("val", String.class.getName(), null);
 
         return super.getConfiguration(igniteInstanceName)
             .setCacheConfiguration(