This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new e59277bfcf2 IGNITE-28925 SQL Calcite: Fix unexpected
downstream().end() call for execution nodes - Fixes #13407.
e59277bfcf2 is described below
commit e59277bfcf27099dee9802be3ac37014d017c35c
Author: Aleksey Plekhanov <[email protected]>
AuthorDate: Fri Jul 31 09:46:29 2026 +0300
IGNITE-28925 SQL Calcite: Fix unexpected downstream().end() call for
execution nodes - Fixes #13407.
Signed-off-by: Aleksey Plekhanov <[email protected]>
---
.../query/calcite/exec/rel/AbstractNode.java | 2 +-
.../query/calcite/exec/rel/CollectNode.java | 11 ++++-
.../exec/rel/CorrelatedNestedLoopJoinNode.java | 12 +++--
.../query/calcite/exec/rel/SortAggregateNode.java | 10 ++++-
.../query/calcite/exec/rel/SortNode.java | 7 +--
.../integration/CollectIntegrationTest.java | 52 ++++++++++++++++++++++
.../integration/SortAggregateIntegrationTest.java | 26 +++++++++++
.../ignite/testsuites/IntegrationTestSuite.java | 2 +
8 files changed, 112 insertions(+), 10 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java
index ef372b62f2a..e426069a32d 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AbstractNode.java
@@ -38,7 +38,7 @@ import static
org.apache.ignite.IgniteSystemProperties.IGNITE_CALCITE_EXEC_MODIF
*/
public abstract class AbstractNode<Row> implements Node<Row> {
/** */
- protected static final int IN_BUFFER_SIZE =
IgniteSystemProperties.getInteger(IGNITE_CALCITE_EXEC_IN_BUFFER_SIZE, 512);
+ public static final int IN_BUFFER_SIZE =
IgniteSystemProperties.getInteger(IGNITE_CALCITE_EXEC_IN_BUFFER_SIZE, 512);
/** */
protected static final int MODIFY_BATCH_SIZE =
IgniteSystemProperties.getInteger(IGNITE_CALCITE_EXEC_MODIFY_BATCH_SIZE, 100);
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CollectNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CollectNode.java
index af13ba2b415..468d7b3a837 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CollectNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CollectNode.java
@@ -114,6 +114,11 @@ public class CollectNode<Row> extends
MemoryTrackingNode<Row> implements SingleN
if (waiting == 0)
source().request(waiting = IN_BUFFER_SIZE);
+ else if (waiting < 0) {
+ requested = 0;
+
+ downstream().end();
+ }
}
/** {@inheritDoc} */
@@ -138,6 +143,7 @@ public class CollectNode<Row> extends
MemoryTrackingNode<Row> implements SingleN
@Override public void end() throws Exception {
assert downstream() != null;
assert waiting > 0;
+ assert requested > 0;
checkState();
@@ -146,10 +152,13 @@ public class CollectNode<Row> extends
MemoryTrackingNode<Row> implements SingleN
if (isClosed())
return;
+ requested--;
+
+ downstream().push(collector.get());
+
if (requested > 0) {
requested = 0;
- downstream().push(collector.get());
downstream().end();
}
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CorrelatedNestedLoopJoinNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CorrelatedNestedLoopJoinNode.java
index 8860cc10570..48ddd9fe628 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CorrelatedNestedLoopJoinNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/CorrelatedNestedLoopJoinNode.java
@@ -282,6 +282,8 @@ public class CorrelatedNestedLoopJoinNode<Row> extends
AbstractNode<Row> {
break;
case END:
+ requested = 0;
+
downstream().end();
break;
@@ -338,8 +340,11 @@ public class CorrelatedNestedLoopJoinNode<Row> extends
AbstractNode<Row> {
state = State.END;
- if (requested > 0)
+ if (requested > 0) {
+ requested = 0;
+
downstream().end();
+ }
}
else {
prepareCorrelations();
@@ -469,8 +474,9 @@ public class CorrelatedNestedLoopJoinNode<Row> extends
AbstractNode<Row> {
state = State.END;
- if (requested > 0)
- downstream().end();
+ requested = 0;
+
+ downstream().end();
return;
}
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortAggregateNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortAggregateNode.java
index 3b261b936b3..e03552427d1 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortAggregateNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortAggregateNode.java
@@ -101,8 +101,11 @@ public class SortAggregateNode<Row> extends
AggregateNode<Row> {
source().request(IN_BUFFER_SIZE);
}
- else if (waiting < 0)
+ else if (waiting < 0) {
+ requested = 0;
+
downstream().end();
+ }
}
/** {@inheritDoc} */
@@ -163,8 +166,11 @@ public class SortAggregateNode<Row> extends
AggregateNode<Row> {
doPush();
}
- if (requested > 0)
+ if (requested > 0) {
+ requested = 0;
+
downstream().end();
+ }
grp = null;
prevRow = null;
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
index 09376b210f6..1347fced4b6 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/SortNode.java
@@ -205,10 +205,11 @@ public class SortNode<Row> extends
MemoryTrackingNode<Row> implements SingleNode
}
if (reversed == null ? rows.isEmpty() : reversed.isEmpty()) {
- if (requested > 0)
- downstream().end();
+ if (requested > 0) {
+ requested = 0;
- requested = 0;
+ downstream().end();
+ }
}
}
finally {
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CollectIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CollectIntegrationTest.java
new file mode 100644
index 00000000000..726a3453e1b
--- /dev/null
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CollectIntegrationTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.integration;
+
+import
org.apache.ignite.internal.processors.query.calcite.exec.rel.AbstractNode;
+import org.junit.Test;
+
+/**
+ * Integration test for collect node.
+ */
+public class CollectIntegrationTest extends AbstractBasicIntegrationTest {
+ /**
+ * Tests that collect node correctly handles the case when downstream
requests
+ * limited number of rows, where collect must push one row and then
+ * properly terminate downstream.
+ */
+ @Test
+ public void testRequestLimitedRowsCountFromCollect() {
+ sql("CREATE TABLE t(a INT)");
+
+ sql("INSERT INTO t (a) VALUES (?)", 0);
+
+ String sql = "SELECT /*+ CNL_JOIN */ ARRAY(SELECT a FROM t) FROM t
LIMIT 1";
+
+ assertQuery(sql).resultSize(1).check();
+
+ /**
+ * The data source size of (buffer size + 1) is used to ensure that
multiple batches are needed
+ * on right hand of CNLJ to process all input rows, in this case left
hand is not requested
+ * immediately after endLeft() call.
+ */
+ for (int i = 1; i < AbstractNode.IN_BUFFER_SIZE + 1; i++)
+ sql("INSERT INTO t (a) VALUES (?)", i);
+
+ assertQuery(sql).resultSize(1).check();
+ }
+}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
index 8d9f9cc8535..dd802750931 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SortAggregateIntegrationTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.cache.QueryIndexType;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.configuration.IgniteConfiguration;
+import
org.apache.ignite.internal.processors.query.calcite.exec.rel.AbstractNode;
import org.apache.ignite.internal.util.typedef.F;
import org.junit.Test;
@@ -155,6 +156,31 @@ public class SortAggregateIntegrationTest extends
AbstractBasicIntegrationTransa
.check();
}
+ /**
+ * Tests that sort aggregate node correctly handles the case when input
data
+ * ends exactly when the requested number of rows is satisfied.
+ */
+ @Test
+ public void testRequestRowsAfterInputEnds() {
+ /**
+ * With input rows count equals to the buffer size, the last row
completes both
+ * the input data and the requested count in the same cycle. This
triggers
+ * a synchronous request() call from within push() to fill the buffer,
and
+ * the node must properly handle the termination on the subsequent
request()
+ * call rather than on end().
+ */
+ int bufSize = AbstractNode.IN_BUFFER_SIZE;
+
+ sql("CREATE TABLE t0(a INTEGER PRIMARY KEY, b INTEGER) WITH
template=replicated," + atomicity());
+
+ for (int i = 0; i < bufSize; i++)
+ sql("INSERT INTO t0 VALUES (?, ?)", i, i);
+
+ assertQuery("SELECT t1.a FROM t0 AS t1 JOIN (SELECT a, count(a) FROM
t0 GROUP BY a) AS t2 ON t1.a = t2.a")
+ .resultSize(bufSize)
+ .check();
+ }
+
/**
* @param c Cache.
* @param rows Rows count.
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
index e392d7c78bb..7c32feb3186 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
@@ -32,6 +32,7 @@ import
org.apache.ignite.internal.processors.query.calcite.integration.CacheWith
import
org.apache.ignite.internal.processors.query.calcite.integration.CalciteBasicSecondaryIndexIntegrationTest;
import
org.apache.ignite.internal.processors.query.calcite.integration.CalciteErrorHandlilngIntegrationTest;
import
org.apache.ignite.internal.processors.query.calcite.integration.CalcitePlanningDumpTest;
+import
org.apache.ignite.internal.processors.query.calcite.integration.CollectIntegrationTest;
import
org.apache.ignite.internal.processors.query.calcite.integration.CorrelatesIntegrationTest;
import
org.apache.ignite.internal.processors.query.calcite.integration.DataTypesTest;
import
org.apache.ignite.internal.processors.query.calcite.integration.DateTimeTest;
@@ -187,6 +188,7 @@ import org.junit.runners.Suite;
SelectByKeyFieldTest.class,
WindowIntegrationTest.class,
CalciteMessageUnmarshalThreadIntegrationTest.class,
+ CollectIntegrationTest.class,
})
public class IntegrationTestSuite {
}
