IGNITE-24274: Sql. Inbox does not handle closing properly

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractNode.java

@@ -137,6 +137,7 @@ protected void closeInternal() {
     /**
      * Get closed flag: {@code true} if the subtree is canceled.
      */
+    @Override
     public boolean isClosed() {
         return closed;
     }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractRightMaterializedJoinNode.java

@@ -51,7 +51,7 @@ public void request(int rowsCnt) throws Exception {
         requested = rowsCnt;
 
         if (!inLoop) {
-            context().execute(this::doJoin, this::onError);
+            this.execute(this::doJoin, this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractSetOpNode.java

@@ -71,7 +71,7 @@ public void request(int rowsCnt) throws Exception {
         if (waiting == 0) {
             sources().get(curSrcIdx).request(waiting = inBufSize);
         } else if (!inLoop) {
-            context().execute(this::flush, this::onError);
+            this.execute(this::flush, this::onError);
         }
     }
 
@@ -179,7 +179,7 @@ private void flush() throws Exception {
 
                 if (processed >= inBufSize && requested > 0) {
                     // Allow others to do their job.
-                    context().execute(this::flush, this::onError);
+                    this.execute(this::flush, this::onError);
 
                     return;
                 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/CorrelatedNestedLoopJoinNode.java

@@ -258,7 +258,7 @@ private void onRequest() throws Exception {
                 assert nullOrEmpty(leftInBuf);
                 assert nullOrEmpty(rightInBuf);
 
-                context().execute(() -> {
+                this.execute(() -> {
                     checkState();
 
                     state = State.FILLING_LEFT;
@@ -272,7 +272,7 @@ private void onRequest() throws Exception {
                 assert waitingRight == -1 || waitingRight == 0 && rightInBuf.size() == rightInBufferSize;
                 assert waitingLeft == -1 || waitingLeft == 0 && leftInBuf.size() == leftInBufferSize;
 
-                context().execute(() -> {
+                this.execute(() -> {
                     checkState();
 
                     join();

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/FilterNode.java

@@ -63,7 +63,7 @@ public void request(int rowsCnt) throws Exception {
         requested = rowsCnt;
 
         if (!inLoop) {
-            context().execute(this::doFilter, this::onError);
+            this.execute(this::doFilter, this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/HashAggregateNode.java

@@ -107,7 +107,7 @@ public void request(int rowsCnt) throws Exception {
         if (waiting == 0) {
             source().request(waiting = inBufSize);
         } else if (!inLoop) {
-            context().execute(this::flush, this::onError);
+            this.execute(this::flush, this::onError);
         }
     }
 
@@ -191,7 +191,7 @@ private void flush() throws Exception {
 
                 if (processed >= inBufSize && requested > 0) {
                     // allow others to do their job
-                    context().execute(this::flush, this::onError);
+                    this.execute(this::flush, this::onError);
 
                     return;
                 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java

@@ -117,7 +117,7 @@ public void request(int rowsCnt) throws Exception {
         requested = rowsCnt;
 
         if (!inLoop) {
-            context().execute(this::doPush, this::onError);
+            this.execute(this::doPush, this::onError);
         }
     }
 
@@ -356,7 +356,7 @@ private void requestBatches(String nodeName, int cnt, @Nullable SharedState stat
                                 ex
                         );
 
-                        context().execute(() -> onError(wrapperEx), this::onError);
+                        this.execute(() -> onError(wrapperEx), this::onError);
                     }
                 });
     }
@@ -367,9 +367,9 @@ private void requestBatches(String nodeName, int cnt, @Nullable SharedState stat
      */
     public void onNodeLeft(String nodeName) {
         if (context().originatingNodeName().equals(nodeName) && srcNodeNames == null) {
-            context().execute(this::close, this::onError);
+            this.execute(this::close, this::onError);
         } else if (srcNodeNames != null && srcNodeNames.contains(nodeName)) {
-            context().execute(() -> onNodeLeft0(nodeName), this::onError);
+            this.execute(() -> onNodeLeft0(nodeName), this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/IndexSpoolNode.java

@@ -126,7 +126,7 @@ public void push(RowT row) throws Exception {
         waiting--;
 
         if (waiting == 0) {
-            context().execute(this::requestSource, this::onError);
+            this.execute(this::requestSource, this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinNode.java

@@ -82,7 +82,7 @@ public void request(int rowsCnt) throws Exception {
         requested = rowsCnt;
 
         if (!inLoop) {
-            context().execute(this::doJoin, this::onError);
+            this.execute(this::doJoin, this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ModifyNode.java

@@ -290,7 +290,7 @@ private void flushTuples() {
                 throw new UnsupportedOperationException(modifyOp.name());
         }
 
-        modifyResult.whenComplete((r, e) -> context().execute(() -> {
+        modifyResult.whenComplete((r, e) -> this.execute(() -> {
             if (e != null) {
                 onError(e);
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Node.java

@@ -18,6 +18,8 @@
 package org.apache.ignite.internal.sql.engine.exec.rel;
 
 import java.util.List;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.lang.RunnableX;
 import org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
 
 /**
@@ -71,4 +73,31 @@ public interface Node<RowT> extends AutoCloseable {
      * Rewinds upstream.
      */
     void rewind();
+
+    /**
+     * Schedules the given action of this execution node.
+     *
+     * @param action Task.
+     * @param onError Error handler.
+     */
+    default void execute(RunnableX action, Consumer<Throwable> onError) {
+        if (this.isClosed()) {
+            return;
+        }
+
+        context().execute(() -> {
+            // If the node is closed, the task must be ignored.
+            if (this.isClosed()) {
+                return;
+            }
+            action.run();
+        }, onError);
+    }
+
+    /**
+     * Returns {@code true} if this node was closed.
+     *
+     * @return {@code true} if not was closed.
+     */
+    boolean isClosed();
 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java

@@ -265,7 +265,7 @@ private void sendBatch(String nodeName, int batchId, boolean last, List<RowT> ro
                             ex
                     );
 
-                    context().execute(() -> onError(wrapperEx), this::onError);
+                    this.execute(() -> onError(wrapperEx), this::onError);
                 });
     }
 
@@ -344,7 +344,7 @@ private void flush() throws Exception {
      */
     public void onNodeLeft(String nodeName) {
         if (nodeName.equals(context().originatingNodeName())) {
-            context().execute(this::close, this::onError);
+            this.execute(this::close, this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/RootNode.java

@@ -130,7 +130,7 @@ public boolean isClosed() {
     /** {@inheritDoc} */
     @Override
     public void closeInternal() {
-        context().execute(() -> sources().forEach(Commons::closeQuiet), this::onError);
+        this.execute(() -> sources().forEach(Commons::closeQuiet), this::onError);
     }
 
     /** {@inheritDoc} */
@@ -255,7 +255,7 @@ private void exchangeBuffers() {
                     close();
                 } else if (inBuff.isEmpty() && waiting == 0) {
                     int req = waiting = inBufSize;
-                    context().execute(() -> source().request(req), this::onError);
+                    this.execute(() -> source().request(req), this::onError);
                 }
 
                 if (!outBuff.isEmpty() || waiting == -1) {

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/ScanNode.java

@@ -70,7 +70,7 @@ public void request(int rowsCnt) throws Exception {
         requested = rowsCnt;
 
         if (!inLoop) {
-            context().execute(this::push, this::onError);
+            this.execute(this::push, this::onError);
         }
     }
 
@@ -126,7 +126,7 @@ private void push() throws Exception {
 
                 if (++processed == inBufSize && requested > 0) {
                     // allow others to do their job
-                    context().execute(this::push, this::onError);
+                    this.execute(this::push, this::onError);
 
                     return;
                 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortAggregateNode.java

@@ -154,7 +154,7 @@ public void push(RowT row) throws Exception {
         if (waiting == 0 && requested > 0) {
             waiting = inBufSize;
 
-            context().execute(() -> source().request(inBufSize), this::onError);
+            this.execute(() -> source().request(inBufSize), this::onError);
         }
     }
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/SortNode.java

@@ -119,7 +119,7 @@ public void request(int rowsCnt) throws Exception {
         if (waiting == 0) {
             source().request(waiting = inBufSize);
         } else if (!inLoop) {
-            context().execute(this::flush, this::onError);
+            this.execute(this::flush, this::onError);
         }
     }
 
@@ -176,7 +176,7 @@ private void flush() throws Exception {
 
                     if (++processed >= inBufSize) {
                         // Allow the others to do their job.
-                        context().execute(this::flush, this::onError);
+                        this.execute(this::flush, this::onError);
 
                         return;
                     }
@@ -194,7 +194,7 @@ private void flush() throws Exception {
 
                 if (++processed >= inBufSize && requested > 0) {
                     // allow others to do their job
-                    context().execute(this::flush, this::onError);
+                    this.execute(this::flush, this::onError);
 
                     return;
                 }

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/StorageScanNode.java

@@ -83,7 +83,7 @@ public void request(int rowsCnt) throws Exception {
         requested = rowsCnt;
 
         if (!inLoop) {
-            context().execute(this::push, this::onError);
+            this.execute(this::push, this::onError);
         }
     }
 
@@ -166,7 +166,7 @@ private void push() throws Exception {
                 requested = 0;
                 downstream().end();
             } else {
-                context().execute(this::push, this::onError);
+                this.execute(this::push, this::onError);
             }
         }
     }
@@ -228,7 +228,7 @@ public void onNext(RowT row) {
             inBuffInner.add(row);
 
             if (inBuffInner.size() == inBufSize) {
-                context().execute(() -> {
+                StorageScanNode.this.execute(() -> {
                     waiting = 0;
                     push();
                 }, StorageScanNode.this::onError);
@@ -238,15 +238,15 @@ public void onNext(RowT row) {
         /** {@inheritDoc} */
         @Override
         public void onError(Throwable throwable) {
-            context().execute(() -> {
+            StorageScanNode.this.execute(() -> {
                 throw throwable;
             }, StorageScanNode.this::onError);
         }
 
         /** {@inheritDoc} */
         @Override
         public void onComplete() {
-            context().execute(() -> {
+            StorageScanNode.this.execute(() -> {
                 activeSubscription = null;
                 waiting = 0;
 

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/TableSpoolNode.java

@@ -100,7 +100,7 @@ public void request(int rowsCnt) throws Exception {
         requested += rowsCnt;
 
         if ((waiting == -1 || rowIdx < rows.size()) && !inLoop) {
-            context().execute(this::doPush, this::onError);
+            this.execute(this::doPush, this::onError);
         } else if (waiting == 0) {
             source().request(waiting = inBufSize);
         }
@@ -132,7 +132,7 @@ private void doPush() throws Exception {
             requested = 0;
             downstream().end();
         } else if (requested > 0 && processed >= inBufSize) {
-            context().execute(this::doPush, this::onError);
+            this.execute(this::doPush, this::onError);
         }
     }
 
@@ -167,6 +167,6 @@ public void end() throws Exception {
 
         waiting = -1;
 
-        context().execute(this::doPush, this::onError);
+        this.execute(this::doPush, this::onError);
     }
 }

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractJoinExecutionTest.java

@@ -458,7 +458,7 @@ private void verifyJoin(Object[][] left, Object[][] right, JoinRelType joinType,
             TestDownstream<Object[]> downstream = new TestDownstream<>();
             project.onRegister(downstream);
 
-            ctx.execute(() -> project.request(1024), project::onError);
+            project.execute(() -> project.request(1024), project::onError);
 
             Object[][] res = await(downstream.result()).toArray(EMPTY);
 

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/DataSourceScanNodeSelfTest.java

@@ -244,7 +244,7 @@ context, factory, fromRowSchema(ROW_SCHEMA), dataSource, predicate, projection,
 
         node.onRegister(downstream);
 
-        context.execute(() -> node.request(Integer.MAX_VALUE), node::onError);
+        node.execute(() -> node.request(Integer.MAX_VALUE), node::onError);
 
         return await(downstream.result());
     }

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/ExecutionTest.java

@@ -446,6 +446,8 @@ protected Object[] row(Object... fields) {
      * Node that always throws {@link IllegalAccessError} except for {@link #close()} and {@link #onRegister(Downstream)} methods.
      */
     static class CorruptedNode<T> implements Node<T> {
+        private volatile boolean closed;
+
         /** {@inheritDoc} */
         @Override
         public ExecutionContext<T> context() {
@@ -488,9 +490,15 @@ public void rewind() {
             throw new IllegalAccessError();
         }
 
+        @Override
+        public boolean isClosed() {
+            return closed;
+        }
+
         /** {@inheritDoc} */
         @Override
         public void close() {
+            this.closed = true;
         }
     }
 

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/LimitExecutionTest.java

@@ -209,7 +209,7 @@ protected void rewindInternal() {
         @Override public void request(int rowsCnt) {
             int r = requested.getAndAdd(rowsCnt);
 
-            context().execute(() -> {
+            this.execute(() -> {
                 for (int i = 0; i < rowsCnt; i++) {
                     downstream().push(new Object[]{r + i});
                 }