This is an automated email from the ASF dual-hosted git repository.
zstan 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 c21741a0c78 IGNITE-28959 Calcite. Merge Join occasionally return wrong
results (#13455)
c21741a0c78 is described below
commit c21741a0c7852ad2ae213dd129d8c7a82a222e4e
Author: Evgeniy Stanilovskiy <[email protected]>
AuthorDate: Tue Aug 11 08:04:13 2026 +0300
IGNITE-28959 Calcite. Merge Join occasionally return wrong results (#13455)
---
.../query/calcite/exec/rel/MergeJoinNode.java | 78 +++++++++++-----------
.../calcite/exec/rel/JoinBuffersExecutionTest.java | 3 -
.../integration/CorrelatesIntegrationTest.java | 24 +++++++
3 files changed, 64 insertions(+), 41 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
index 800e66e918b..57b8a488130 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/MergeJoinNode.java
@@ -84,9 +84,6 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
*/
protected final boolean distributed;
- /** Flag indicating that join is in finishing stage (one of the inputs are
ended, no more rows will be produced). */
- protected boolean finishing;
-
/**
* @param ctx Execution context.
* @param comp Join expression.
@@ -178,10 +175,10 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
waitingLeft--;
- if (!finishing)
- leftInBuf.add(row);
+ leftInBuf.add(row);
- join();
+ if (waitingLeft == 0 && waitingRight <= 0)
+ join();
}
/** */
@@ -191,10 +188,10 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
waitingRight--;
- if (!finishing)
- rightInBuf.add(row);
+ rightInBuf.add(row);
- join();
+ if (waitingRight == 0 && waitingLeft <= 0)
+ join();
}
/** */
@@ -204,7 +201,8 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
waitingLeft = NOT_WAITING;
- join();
+ if (waitingRight <= 0)
+ join();
}
/** */
@@ -214,7 +212,8 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
waitingRight = NOT_WAITING;
- join();
+ if (waitingLeft <= 0)
+ join();
}
/** */
@@ -238,27 +237,6 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
&& (!withMaterialization || rightMaterialization == null);
}
- /** */
- protected boolean checkJoinFinished() throws Exception {
- if (!finishing) {
- finishing = true;
- leftInBuf.clear();
- rightInBuf.clear();
- rightMaterialization = null;
- rightIdx = 0;
- drainMaterialization = false;
- }
-
- if (!distributed || (waitingLeft == NOT_WAITING && waitingRight ==
NOT_WAITING)) {
- requested = 0;
- downstream().end();
-
- return true;
- }
-
- return false;
- }
-
/** */
protected void tryToRequestInputs() throws Exception {
if (waitingLeft == 0 && leftInBuf.size() <= HALF_BUF_SIZE)
@@ -414,8 +392,12 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
inLoop = false;
}
- if (requested > 0 && (leftFinished() || rightFinished(true)) &&
checkJoinFinished())
+ if (requested > 0 && (leftFinished() || rightFinished(true))) {
+ requested = 0;
+ downstream().end();
+
return;
+ }
tryToRequestInputs();
}
@@ -568,8 +550,12 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
inLoop = false;
}
- if (requested > 0 && leftFinished() && checkJoinFinished())
+ if (requested > 0 && leftFinished()) {
+ requested = 0;
+ downstream().end();
+
return;
+ }
tryToRequestInputs();
}
@@ -734,8 +720,12 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
inLoop = false;
}
- if (requested > 0 && rightFinished(true) && checkJoinFinished())
+ if (requested > 0 && rightFinished(true)) {
+ requested = 0;
+ downstream().end();
+
return;
+ }
tryToRequestInputs();
}
@@ -939,8 +929,12 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
inLoop = false;
}
- if (requested > 0 && leftFinished() && rightFinished(true) &&
checkJoinFinished())
+ if (requested > 0 && leftFinished() && rightFinished(true)) {
+ requested = 0;
+ downstream().end();
+
return;
+ }
tryToRequestInputs();
}
@@ -995,8 +989,12 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
inLoop = false;
}
- if (requested > 0 && (leftFinished() || rightFinished(false)) &&
checkJoinFinished())
+ if (requested > 0 && (leftFinished() || rightFinished(false))) {
+ requested = 0;
+ downstream().end();
+
return;
+ }
tryToRequestInputs();
}
@@ -1054,8 +1052,12 @@ public abstract class MergeJoinNode<Row> extends
AbstractNode<Row> {
inLoop = false;
}
- if (requested > 0 && leftFinished() && checkJoinFinished())
+ if (requested > 0 && leftFinished()) {
+ requested = 0;
+ downstream().end();
+
return;
+ }
tryToRequestInputs();
}
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
index 4ff685131ce..0d52216bd98 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/JoinBuffersExecutionTest.java
@@ -34,8 +34,6 @@ import
org.apache.ignite.internal.processors.query.calcite.util.TypeUtils;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.testframework.GridTestUtils;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
import static org.apache.calcite.rel.core.JoinRelType.ANTI;
import static org.apache.calcite.rel.core.JoinRelType.FULL;
@@ -45,7 +43,6 @@ import static org.apache.calcite.rel.core.JoinRelType.RIGHT;
import static org.apache.calcite.rel.core.JoinRelType.SEMI;
/** Tests that buffers of join nodes are cleared at the join end and that a
join node is not stuck. */
-@RunWith(Parameterized.class)
public class JoinBuffersExecutionTest extends AbstractExecutionTest {
/** */
@Test
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
index bc7df8a544a..61be928ac2c 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/CorrelatesIntegrationTest.java
@@ -105,6 +105,30 @@ public class CorrelatesIntegrationTest extends
AbstractBasicIntegrationTransacti
.check();
}
+ /** */
+ @Test
+ public void testMergeJoinUnderCorrelateSurvivesRewind() {
+ sql("CREATE TABLE t0 (a INTEGER, b INTEGER) WITH " + atomicity());
+ sql("CREATE TABLE t1 (a INTEGER, b INTEGER) WITH " + atomicity());
+ sql("CREATE TABLE t2 (a INTEGER, b INTEGER) WITH " + atomicity());
+
+ sql("INSERT INTO t0 VALUES (1, 1), (2, 2), (3, 3)");
+ sql("INSERT INTO t1 VALUES (1, 1), (2, 2), (3, 3)");
+ sql("INSERT INTO t2 VALUES (1, 1)");
+
+ String qry = "SELECT t0.a, (SELECT /*+ MERGE_JOIN */ count(*) FROM t1
LEFT JOIN t2 ON t1.a = t2.a " +
+ "WHERE t1.a = (SELECT t0.a)) FROM t0";
+
+ // The defect is only reachable while the plan keeps a merge join
under the correlate.
+ assertQuery(qry)
+
.matches(QueryChecker.containsSubPlan("IgniteCorrelatedNestedLoopJoin"))
+ .matches(QueryChecker.containsSubPlan("IgniteMergeJoin"))
+ .returns(1, 1L)
+ .returns(2, 1L)
+ .returns(3, 1L)
+ .check();
+ }
+
/**
* Tests colocated join possible with the help of correlated distribution.
*/
