From e10a1ff89ba109b06356cab7332e04521ace51cc Mon Sep 17 00:00:00 2001 From: zstan Date: Fri, 7 Aug 2026 10:54:05 +0300 Subject: [PATCH 1/6] IGNITE-28959 Calcite. Merge Join occasionally return wrong results --- .../query/calcite/exec/rel/MergeJoinNode.java | 78 ++++++++++--------- .../exec/rel/JoinBuffersExecutionTest.java | 3 - .../CorrelatesIntegrationTest.java | 35 ++++++--- 3 files changed, 63 insertions(+), 53 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 800e66e918b72..57b8a488130b4 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 extends AbstractNode { */ 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 @@ private void pushLeft(Row row) throws Exception { waitingLeft--; - if (!finishing) - leftInBuf.add(row); + leftInBuf.add(row); - join(); + if (waitingLeft == 0 && waitingRight <= 0) + join(); } /** */ @@ -191,10 +188,10 @@ private void pushRight(Row row) throws Exception { waitingRight--; - if (!finishing) - rightInBuf.add(row); + rightInBuf.add(row); - join(); + if (waitingRight == 0 && waitingLeft <= 0) + join(); } /** */ @@ -204,7 +201,8 @@ private void endLeft() throws Exception { waitingLeft = NOT_WAITING; - join(); + if (waitingRight <= 0) + join(); } /** */ @@ -214,7 +212,8 @@ private void endRight() throws Exception { waitingRight = NOT_WAITING; - join(); + if (waitingLeft <= 0) + join(); } /** */ @@ -238,27 +237,6 @@ protected boolean rightFinished(boolean withMaterialization) { && (!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 @@ else if (cmp > 0) { 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 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && checkJoinFinished()) + if (requested > 0 && leftFinished()) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -734,8 +720,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && rightFinished(true) && checkJoinFinished()) + if (requested > 0 && rightFinished(true)) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -939,8 +929,12 @@ else if (cmp > 0) { 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 @@ else if (cmp > 0) { 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 @@ else if (cmp > 0) { 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 4ff685131cea0..0d52216bd9836 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.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.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 bc7df8a544a23..6d7f885ae798f 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 @@ -19,10 +19,16 @@ import java.sql.Date; import java.time.Instant; +import java.util.List; import java.util.UUID; import org.apache.ignite.internal.processors.query.calcite.QueryChecker; +import org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition; import org.junit.Test; +import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NO_CNL_JOIN; +import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NO_HASH_JOIN; +import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NO_NL_JOIN; + /** * Tests correlated queries. */ @@ -90,19 +96,24 @@ public void testCorrelatesCollision() { sql("INSERT INTO test1 VALUES (11, 1), (12, 2), (13, 3)"); sql("INSERT INTO test2 VALUES (11, 1), (12, 1), (13, 4)"); - // Collision by correlate variables in the left hand. - assertQuery("SELECT * FROM test1 WHERE " + - "EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b<>test2.c) " + - "AND NOT EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b>> Check with: " + noHint); + // Collision by correlate variables in the left hand. + assertQuery("SELECT /*+ %s */ * FROM test1 WHERE ".formatted(noHint) + + "EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b<>test2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.btest2.c) " + - "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)test2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b) Date: Sun, 9 Aug 2026 16:36:12 +0300 Subject: [PATCH 2/6] fix --- .../query/calcite/exec/rel/MergeJoinNode.java | 80 +++++++++---------- .../CorrelatesIntegrationTest.java | 30 ++++--- 2 files changed, 54 insertions(+), 56 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 57b8a488130b4..eb38d9f6d2b1f 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,6 +84,9 @@ public abstract class MergeJoinNode extends AbstractNode { */ 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. @@ -175,10 +178,10 @@ private void pushLeft(Row row) throws Exception { waitingLeft--; - leftInBuf.add(row); + if (!finishing) + leftInBuf.add(row); - if (waitingLeft == 0 && waitingRight <= 0) - join(); + join(); } /** */ @@ -188,10 +191,10 @@ private void pushRight(Row row) throws Exception { waitingRight--; - rightInBuf.add(row); + if (!finishing) + rightInBuf.add(row); - if (waitingRight == 0 && waitingLeft <= 0) - join(); + join(); } /** */ @@ -201,8 +204,7 @@ private void endLeft() throws Exception { waitingLeft = NOT_WAITING; - if (waitingRight <= 0) - join(); + join(); } /** */ @@ -212,8 +214,7 @@ private void endRight() throws Exception { waitingRight = NOT_WAITING; - if (waitingLeft <= 0) - join(); + join(); } /** */ @@ -237,6 +238,29 @@ protected boolean rightFinished(boolean withMaterialization) { && (!withMaterialization || rightMaterialization == null); } + /** */ + protected boolean checkJoinFinished() throws Exception { + if (!finishing) { + if (leftFinished() && rightFinished(true)) + 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) @@ -392,12 +416,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && (leftFinished() || rightFinished(true))) { - requested = 0; - downstream().end(); - + if (requested > 0 && (leftFinished() || rightFinished(true)) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -550,12 +570,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished()) { - requested = 0; - downstream().end(); - + if (requested > 0 && leftFinished() && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -720,12 +736,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && rightFinished(true)) { - requested = 0; - downstream().end(); - + if (requested > 0 && rightFinished(true) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -929,12 +941,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && rightFinished(true)) { - requested = 0; - downstream().end(); - + if (requested > 0 && leftFinished() && rightFinished(true) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -989,12 +997,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && (leftFinished() || rightFinished(false))) { - requested = 0; - downstream().end(); - + if (requested > 0 && (leftFinished() || rightFinished(false)) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -1052,12 +1056,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished()) { - requested = 0; - downstream().end(); - + if (requested > 0 && leftFinished() && checkJoinFinished()) return; - } tryToRequestInputs(); } 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 6d7f885ae798f..4c20e8d0faeb0 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 @@ -97,22 +97,20 @@ public void testCorrelatesCollision() { sql("INSERT INTO test2 VALUES (11, 1), (12, 1), (13, 4)"); for (HintDefinition noHint : List.of(NO_NL_JOIN, NO_CNL_JOIN, NO_HASH_JOIN)) { - if (noHint.toString().toUpperCase().startsWith("NO_")) { - System.out.println(">>> Check with: " + noHint); - // Collision by correlate variables in the left hand. - assertQuery("SELECT /*+ %s */ * FROM test1 WHERE ".formatted(noHint) + - "EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b<>test2.c) " + - "AND NOT EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.btest2.c) " + - "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)>> Check with: " + noHint); + // Collision by correlate variables in the left hand. + assertQuery("SELECT /*+ %s */ * FROM test1 WHERE ".formatted(noHint) + + "EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b<>test2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.btest2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b) Date: Mon, 10 Aug 2026 13:24:58 +0300 Subject: [PATCH 3/6] fix after review --- .../query/calcite/exec/rel/MergeJoinNode.java | 80 +++++++++---------- .../query/calcite/prepare/PlannerPhase.java | 4 +- .../CorrelatesIntegrationTest.java | 49 +++++++++--- 3 files changed, 78 insertions(+), 55 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 eb38d9f6d2b1f..57b8a488130b4 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 extends AbstractNode { */ 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 @@ private void pushLeft(Row row) throws Exception { waitingLeft--; - if (!finishing) - leftInBuf.add(row); + leftInBuf.add(row); - join(); + if (waitingLeft == 0 && waitingRight <= 0) + join(); } /** */ @@ -191,10 +188,10 @@ private void pushRight(Row row) throws Exception { waitingRight--; - if (!finishing) - rightInBuf.add(row); + rightInBuf.add(row); - join(); + if (waitingRight == 0 && waitingLeft <= 0) + join(); } /** */ @@ -204,7 +201,8 @@ private void endLeft() throws Exception { waitingLeft = NOT_WAITING; - join(); + if (waitingRight <= 0) + join(); } /** */ @@ -214,7 +212,8 @@ private void endRight() throws Exception { waitingRight = NOT_WAITING; - join(); + if (waitingLeft <= 0) + join(); } /** */ @@ -238,29 +237,6 @@ protected boolean rightFinished(boolean withMaterialization) { && (!withMaterialization || rightMaterialization == null); } - /** */ - protected boolean checkJoinFinished() throws Exception { - if (!finishing) { - if (leftFinished() && rightFinished(true)) - 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) @@ -416,8 +392,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && (leftFinished() || rightFinished(true)) && checkJoinFinished()) + if (requested > 0 && (leftFinished() || rightFinished(true))) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -570,8 +550,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && checkJoinFinished()) + if (requested > 0 && leftFinished()) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -736,8 +720,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && rightFinished(true) && checkJoinFinished()) + if (requested > 0 && rightFinished(true)) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -941,8 +929,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && rightFinished(true) && checkJoinFinished()) + if (requested > 0 && leftFinished() && rightFinished(true)) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -997,8 +989,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && (leftFinished() || rightFinished(false)) && checkJoinFinished()) + if (requested > 0 && (leftFinished() || rightFinished(false))) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -1056,8 +1052,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && checkJoinFinished()) + if (requested > 0 && leftFinished()) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java index 1ac34eab52688..860bb868e0889 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java @@ -298,8 +298,8 @@ public enum PlannerPhase { MergeJoinConverterRule.INSTANCE, CorrelatedNestedLoopJoinRule.INSTANCE, CorrelateToNestedLoopRule.INSTANCE, - NestedLoopJoinConverterRule.INSTANCE, - HashJoinConverterRule.INSTANCE, + //NestedLoopJoinConverterRule.INSTANCE, + //HashJoinConverterRule.INSTANCE, // This rule replaces input refs to literals in the window agg calls. // Since ignite aggregate calculation bounded to input field index - this rule should be excluded from rule set. 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 4c20e8d0faeb0..832ea88fec7cf 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 @@ -25,9 +25,9 @@ import org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition; import org.junit.Test; -import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NO_CNL_JOIN; -import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NO_HASH_JOIN; -import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NO_NL_JOIN; +import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.HASH_JOIN; +import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NL_JOIN; +import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.MERGE_JOIN; /** * Tests correlated queries. @@ -96,24 +96,47 @@ public void testCorrelatesCollision() { sql("INSERT INTO test1 VALUES (11, 1), (12, 2), (13, 3)"); sql("INSERT INTO test2 VALUES (11, 1), (12, 1), (13, 4)"); - for (HintDefinition noHint : List.of(NO_NL_JOIN, NO_CNL_JOIN, NO_HASH_JOIN)) { - log.info(">>> Check with: " + noHint); - // Collision by correlate variables in the left hand. - assertQuery("SELECT /*+ %s */ * FROM test1 WHERE ".formatted(noHint) + - "EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b<>test2.c) " + - "AND NOT EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.btest2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE test1.a=test2.a AND test1.b>> Check with: " + noHint); // Collision by correlate variables in both, left and right hands. - assertQuery("SELECT /*+ %s */ * FROM test1 WHERE ".formatted(noHint) + - "EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)<>test2.c) " + + assertQuery("SELECT * FROM test1 WHERE " + + "EXISTS(SELECT /*+ %s */ * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)<>test2.c) ".formatted(noHint) + "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b) Date: Mon, 10 Aug 2026 14:06:25 +0300 Subject: [PATCH 4/6] fix --- .../query/calcite/exec/rel/MergeJoinNode.java | 78 +++++++++---------- .../query/calcite/prepare/PlannerPhase.java | 4 +- .../CorrelatesIntegrationTest.java | 16 ++-- 3 files changed, 47 insertions(+), 51 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 57b8a488130b4..800e66e918b72 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,6 +84,9 @@ public abstract class MergeJoinNode extends AbstractNode { */ 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. @@ -175,10 +178,10 @@ private void pushLeft(Row row) throws Exception { waitingLeft--; - leftInBuf.add(row); + if (!finishing) + leftInBuf.add(row); - if (waitingLeft == 0 && waitingRight <= 0) - join(); + join(); } /** */ @@ -188,10 +191,10 @@ private void pushRight(Row row) throws Exception { waitingRight--; - rightInBuf.add(row); + if (!finishing) + rightInBuf.add(row); - if (waitingRight == 0 && waitingLeft <= 0) - join(); + join(); } /** */ @@ -201,8 +204,7 @@ private void endLeft() throws Exception { waitingLeft = NOT_WAITING; - if (waitingRight <= 0) - join(); + join(); } /** */ @@ -212,8 +214,7 @@ private void endRight() throws Exception { waitingRight = NOT_WAITING; - if (waitingLeft <= 0) - join(); + join(); } /** */ @@ -237,6 +238,27 @@ protected boolean rightFinished(boolean withMaterialization) { && (!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) @@ -392,12 +414,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && (leftFinished() || rightFinished(true))) { - requested = 0; - downstream().end(); - + if (requested > 0 && (leftFinished() || rightFinished(true)) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -550,12 +568,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished()) { - requested = 0; - downstream().end(); - + if (requested > 0 && leftFinished() && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -720,12 +734,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && rightFinished(true)) { - requested = 0; - downstream().end(); - + if (requested > 0 && rightFinished(true) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -929,12 +939,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && rightFinished(true)) { - requested = 0; - downstream().end(); - + if (requested > 0 && leftFinished() && rightFinished(true) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -989,12 +995,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && (leftFinished() || rightFinished(false))) { - requested = 0; - downstream().end(); - + if (requested > 0 && (leftFinished() || rightFinished(false)) && checkJoinFinished()) return; - } tryToRequestInputs(); } @@ -1052,12 +1054,8 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished()) { - requested = 0; - downstream().end(); - + if (requested > 0 && leftFinished() && checkJoinFinished()) return; - } tryToRequestInputs(); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java index 860bb868e0889..1ac34eab52688 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java @@ -298,8 +298,8 @@ public enum PlannerPhase { MergeJoinConverterRule.INSTANCE, CorrelatedNestedLoopJoinRule.INSTANCE, CorrelateToNestedLoopRule.INSTANCE, - //NestedLoopJoinConverterRule.INSTANCE, - //HashJoinConverterRule.INSTANCE, + NestedLoopJoinConverterRule.INSTANCE, + HashJoinConverterRule.INSTANCE, // This rule replaces input refs to literals in the window agg calls. // Since ignite aggregate calculation bounded to input field index - this rule should be excluded from rule set. 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 832ea88fec7cf..d78e9aea62c1b 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 @@ -103,15 +103,13 @@ public void testCorrelatesCollision() { .returns(12, 2) .check(); - for (HintDefinition noHint : List.of(MERGE_JOIN, HASH_JOIN, NL_JOIN)) { - log.info(">>> Check with: " + noHint); - // Collision by correlate variables in both, left and right hands. - assertQuery("SELECT * FROM test1 WHERE " + - "EXISTS(SELECT /*+ %s */ * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)<>test2.c) ".formatted(noHint) + - "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)test2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b) Date: Mon, 10 Aug 2026 14:08:42 +0300 Subject: [PATCH 5/6] Revert "fix" This reverts commit 608d01774bc3f7ca82303ee2ecaad041bbafe670. --- .../query/calcite/exec/rel/MergeJoinNode.java | 78 ++++++++++--------- .../query/calcite/prepare/PlannerPhase.java | 4 +- .../CorrelatesIntegrationTest.java | 16 ++-- 3 files changed, 51 insertions(+), 47 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 800e66e918b72..57b8a488130b4 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 extends AbstractNode { */ 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 @@ private void pushLeft(Row row) throws Exception { waitingLeft--; - if (!finishing) - leftInBuf.add(row); + leftInBuf.add(row); - join(); + if (waitingLeft == 0 && waitingRight <= 0) + join(); } /** */ @@ -191,10 +188,10 @@ private void pushRight(Row row) throws Exception { waitingRight--; - if (!finishing) - rightInBuf.add(row); + rightInBuf.add(row); - join(); + if (waitingRight == 0 && waitingLeft <= 0) + join(); } /** */ @@ -204,7 +201,8 @@ private void endLeft() throws Exception { waitingLeft = NOT_WAITING; - join(); + if (waitingRight <= 0) + join(); } /** */ @@ -214,7 +212,8 @@ private void endRight() throws Exception { waitingRight = NOT_WAITING; - join(); + if (waitingLeft <= 0) + join(); } /** */ @@ -238,27 +237,6 @@ protected boolean rightFinished(boolean withMaterialization) { && (!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 @@ else if (cmp > 0) { 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 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && checkJoinFinished()) + if (requested > 0 && leftFinished()) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -734,8 +720,12 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && rightFinished(true) && checkJoinFinished()) + if (requested > 0 && rightFinished(true)) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } @@ -939,8 +929,12 @@ else if (cmp > 0) { 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 @@ else if (cmp > 0) { 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 @@ else if (cmp > 0) { inLoop = false; } - if (requested > 0 && leftFinished() && checkJoinFinished()) + if (requested > 0 && leftFinished()) { + requested = 0; + downstream().end(); + return; + } tryToRequestInputs(); } diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java index 1ac34eab52688..860bb868e0889 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java @@ -298,8 +298,8 @@ public enum PlannerPhase { MergeJoinConverterRule.INSTANCE, CorrelatedNestedLoopJoinRule.INSTANCE, CorrelateToNestedLoopRule.INSTANCE, - NestedLoopJoinConverterRule.INSTANCE, - HashJoinConverterRule.INSTANCE, + //NestedLoopJoinConverterRule.INSTANCE, + //HashJoinConverterRule.INSTANCE, // This rule replaces input refs to literals in the window agg calls. // Since ignite aggregate calculation bounded to input field index - this rule should be excluded from rule set. 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 d78e9aea62c1b..832ea88fec7cf 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 @@ -103,13 +103,15 @@ public void testCorrelatesCollision() { .returns(12, 2) .check(); - // Collision by correlate variables in both, left and right hands. - assertQuery("SELECT * FROM test1 WHERE " + - "EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)<>test2.c) " + - "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)>> Check with: " + noHint); + // Collision by correlate variables in both, left and right hands. + assertQuery("SELECT * FROM test1 WHERE " + + "EXISTS(SELECT /*+ %s */ * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)<>test2.c) ".formatted(noHint) + + "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b) Date: Mon, 10 Aug 2026 14:09:45 +0300 Subject: [PATCH 6/6] fix --- .../query/calcite/prepare/PlannerPhase.java | 4 ++-- .../CorrelatesIntegrationTest.java | 22 ++++++------------- 2 files changed, 9 insertions(+), 17 deletions(-) diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java index 860bb868e0889..1ac34eab52688 100644 --- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java +++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java @@ -298,8 +298,8 @@ public enum PlannerPhase { MergeJoinConverterRule.INSTANCE, CorrelatedNestedLoopJoinRule.INSTANCE, CorrelateToNestedLoopRule.INSTANCE, - //NestedLoopJoinConverterRule.INSTANCE, - //HashJoinConverterRule.INSTANCE, + NestedLoopJoinConverterRule.INSTANCE, + HashJoinConverterRule.INSTANCE, // This rule replaces input refs to literals in the window agg calls. // Since ignite aggregate calculation bounded to input field index - this rule should be excluded from rule set. 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 832ea88fec7cf..61be928ac2c71 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 @@ -19,16 +19,10 @@ import java.sql.Date; import java.time.Instant; -import java.util.List; import java.util.UUID; import org.apache.ignite.internal.processors.query.calcite.QueryChecker; -import org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition; import org.junit.Test; -import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.HASH_JOIN; -import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.NL_JOIN; -import static org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition.MERGE_JOIN; - /** * Tests correlated queries. */ @@ -103,17 +97,15 @@ public void testCorrelatesCollision() { .returns(12, 2) .check(); - for (HintDefinition noHint : List.of(MERGE_JOIN, HASH_JOIN, NL_JOIN)) { - log.info(">>> Check with: " + noHint); - // Collision by correlate variables in both, left and right hands. - assertQuery("SELECT * FROM test1 WHERE " + - "EXISTS(SELECT /*+ %s */ * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)<>test2.c) ".formatted(noHint) + - "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)test2.c) " + + "AND NOT EXISTS(SELECT * FROM test2 WHERE (SELECT test1.a)=test2.a AND (SELECT test1.b)