IGNITE-28352 Calcite. User defined sql function miss entries are written under the same tx lock

Author: zstan

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

@@ -361,6 +361,7 @@ private ExecutionContext<?> baseInboxContext(UUID nodeId, UUID qryId, long fragm
             NoOpIoTracker.INSTANCE,
             0,
             ImmutableMap.of(),
+            null,
             null);
     }
 }

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

@@ -145,11 +145,15 @@ public class ExecutionContext<Row> extends AbstractQueryContext implements DataC
     /** */
     private Object[] correlations = new Object[16];
 
+    /** Modified entries holder. */
+    @Nullable private final TxAwareModifiedEntriesHolder mofiedEntriesHolder;
+
     /**
      * @param qctx Parent base query context.
      * @param qryId Query ID.
      * @param fragmentDesc Partitions information.
      * @param params Parameters.
+     * @param mofiedEntriesHolder Modified entries holder.
      */
     @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
     public ExecutionContext(
@@ -166,7 +170,8 @@ public ExecutionContext(
         IoTracker ioTracker,
         long timeout,
         Map<String, Object> params,
-        @Nullable Collection<QueryTxEntry> qryTxEntries
+        @Nullable Collection<QueryTxEntry> qryTxEntries,
+        @Nullable TxAwareModifiedEntriesHolder mofiedEntriesHolder
     ) {
         super(qctx);
 
@@ -183,6 +188,7 @@ public ExecutionContext(
         this.params = params;
         this.timeout = timeout;
         this.qryTxEntries = qryTxEntries;
+        this.mofiedEntriesHolder = mofiedEntriesHolder;
 
         startTs = U.currentTimeMillis();
 
@@ -421,12 +427,19 @@ public void execute(RunnableX task, Consumer<Throwable> onError) {
 
         executor.execute(qryId, fragmentId(), () -> {
             try {
+                if (mofiedEntriesHolder != null)
+                    mofiedEntriesHolder.store(qryTxEntries);
+
                 if (!isCancelled())
                     task.run();
             }
             catch (Throwable e) {
                 onError.accept(e);
             }
+            finally {
+                if (mofiedEntriesHolder != null)
+                    mofiedEntriesHolder.detach();
+            }
         });
     }
 

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

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.query.calcite.exec;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
@@ -37,6 +38,7 @@
 import org.apache.ignite.cache.query.QueryCancelledException;
 import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
 import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.TransactionConfiguration;
 import org.apache.ignite.events.CacheQueryReadEvent;
 import org.apache.ignite.events.EventType;
 import org.apache.ignite.internal.GridKernalContext;
@@ -78,6 +80,7 @@
 import org.apache.ignite.internal.processors.query.calcite.message.MessageType;
 import org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
 import org.apache.ignite.internal.processors.query.calcite.message.QueryStartResponse;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryTxEntry;
 import org.apache.ignite.internal.processors.query.calcite.metadata.AffinityService;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentDescription;
 import org.apache.ignite.internal.processors.query.calcite.metadata.FragmentMapping;
@@ -205,6 +208,13 @@ public class ExecutionServiceImpl<Row> extends AbstractService implements Execut
     /** */
     private final Map<String, FragmentPlan> fragmentPlanCache = new GridBoundedConcurrentLinkedHashMap<>(1024);
 
+    /**
+     * Transaction modified entries holder.
+     *
+     * @see TransactionConfiguration#isTxAwareQueriesEnabled()
+     */
+    private TxAwareModifiedEntriesHolder mofiedEntriesHolder;
+
     /**
      * @param ctx Kernal.
      */
@@ -477,13 +487,15 @@ public void injectService(InjectResourcesService injectSvc) {
         eventManager().addDiscoveryEventListener(discoLsnr, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
 
         iteratorsHolder().init();
+        mofiedEntriesHolder = new TxAwareModifiedEntriesHolder(U.isTxAwareQueriesEnabled(ctx));
     }
 
     /** {@inheritDoc} */
     @Override public void tearDown() {
         eventManager().removeDiscoveryEventListener(discoLsnr, EventType.EVT_NODE_FAILED, EventType.EVT_NODE_LEFT);
 
         iteratorsHolder().tearDown();
+        mofiedEntriesHolder = null;
     }
 
     /** */
@@ -515,7 +527,8 @@ private FragmentPlan prepareFragment(BaseQueryContext ctx, String jsonFragment)
             case DML:
                 ListFieldsQueryCursor<?> cur = mapAndExecutePlan(
                     qry,
-                    (MultiStepPlan)plan
+                    (MultiStepPlan)plan,
+                    mofiedEntriesHolder
                 );
 
                 cur.iterator().hasNext();
@@ -525,7 +538,8 @@ private FragmentPlan prepareFragment(BaseQueryContext ctx, String jsonFragment)
             case QUERY:
                 return mapAndExecutePlan(
                     qry,
-                    (MultiStepPlan)plan
+                    (MultiStepPlan)plan,
+                    mofiedEntriesHolder
                 );
 
             case EXPLAIN:
@@ -577,7 +591,8 @@ private FieldsQueryCursor<List<?>> executeDdl(RootQuery<Row> qry, DdlPlan plan)
     /** */
     private ListFieldsQueryCursor<?> mapAndExecutePlan(
         RootQuery<Row> qry,
-        MultiStepPlan plan
+        MultiStepPlan plan,
+        TxAwareModifiedEntriesHolder mofiedEntriesHolder
     ) {
         qry.mapping();
 
@@ -626,6 +641,8 @@ private ListFieldsQueryCursor<?> mapAndExecutePlan(
         MemoryTracker qryMemoryTracker = qry.createMemoryTracker(memoryTracker, cfg.getQueryMemoryQuota());
 
         final GridNearTxLocal userTx = Commons.queryTransaction(qry.context(), ctx.cache().context());
+        final Collection<QueryTxEntry> writeEntrs = userTx == null ?
+            mofiedEntriesHolder.retrieve() : ExecutionContext.transactionChanges(userTx.writeEntries());
 
         ExecutionContext<Row> ectx = new ExecutionContext<>(
             qry.context(),
@@ -641,7 +658,8 @@ private ListFieldsQueryCursor<?> mapAndExecutePlan(
             createIoTracker(locNodeId, qry.localQueryId()),
             timeout,
             qryParams,
-            userTx == null ? null : ExecutionContext.transactionChanges(userTx.writeEntries()));
+            writeEntrs,
+            mofiedEntriesHolder);
 
         Node<Row> node = new LogicalRelImplementor<>(ectx, partitionService(), mailboxRegistry(),
             exchangeService(), failureProcessor()).go(fragment.root());
@@ -901,7 +919,8 @@ private void onMessage(UUID nodeId, final QueryStartRequest msg) {
                 createIoTracker(nodeId, msg.originatingQueryId()),
                 msg.timeout(),
                 Commons.parametersMap(msg.parameters()),
-                msg.queryTransactionEntries()
+                msg.queryTransactionEntries(),
+                null
             );
 
             executeFragment(qry, fragmentPlan, ectx);

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

@@ -0,0 +1,56 @@
+/*
+ * 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.Collection;
+import java.util.Collections;
+import org.apache.ignite.internal.processors.query.calcite.message.QueryTxEntry;
+import org.jetbrains.annotations.Nullable;
+
+/** Per thread modified entries holder. */
+public class TxAwareModifiedEntriesHolder {
+    /** Transaction modified entries holder. */
+    @Nullable private final ThreadLocal<Collection<QueryTxEntry>> holder;
+
+    /** */
+    public TxAwareModifiedEntriesHolder(boolean txAware) {
+        if (txAware)
+            holder = new ThreadLocal<>();
+        else
+            holder = null;
+    }
+
+    /** Store entries if applicable. */
+    public void store(Collection<QueryTxEntry> items) {
+        if (holder != null)
+            holder.set(items);
+    }
+
+    /** Retirieve entries if applicable. */
+    public Collection<QueryTxEntry> retrieve() {
+        if (holder != null)
+            return holder.get();
+        else
+            return Collections.emptyList();
+    }
+
+    /** Detach stored entries. */
+    public void detach() {
+        if (holder != null)
+            holder.remove();
+    }
+}

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

@@ -133,6 +133,7 @@ public class LogicalRelImplementorTest extends GridCommonAbstractTest {
             NoOpIoTracker.INSTANCE,
             0,
             null,
+            null,
             null
         ) {
             @Override public ColocationGroup group(long srcId) {

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

@@ -121,6 +121,7 @@ private RuntimeSortedIndex<Object[]> generate(RelDataType rowType, final List<In
                 NoOpIoTracker.INSTANCE,
                 0,
                 null,
+                null,
                 null),
             RelCollations.of(ImmutableIntList.copyOf(idxCols)),
             (o1, o2) -> {

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractExecutionTest.java

@@ -362,6 +362,7 @@ protected ExecutionContext<Object[]> executionContext(UUID nodeId, UUID qryId, l
             NoOpIoTracker.INSTANCE,
             0,
             ImmutableMap.of(),
+            null,
             null
         );
     }