[CALCITE-7817] Merge ANTI JOIN should preserve unmatched NULL keys

core/src/test/java/org/apache/calcite/runtime/EnumerablesTest.java

@@ -192,30 +192,35 @@ class EnumerablesTest {
         newArrayList(1, 3, 4),
         newArrayList(1, 4),
         equalTo("[3]"),
+        equalTo("[3, null]"),
         JoinType.ANTI);
     // Matching key at start and end of right, not of left
     testIntersect(
         newArrayList(0, 1, 3, 4, 5),
         newArrayList(1, 4),
         equalTo("[0, 3, 5]"),
+        equalTo("[0, 3, 5, null]"),
         JoinType.ANTI);
     // Matching key at start and end of left, not right
     testIntersect(
         newArrayList(1, 3, 4),
         newArrayList(0, 1, 4, 5),
         equalTo("[3]"),
+        equalTo("[3, null]"),
         JoinType.ANTI);
     // Matching key not at start or end of left or right
     testIntersect(
         newArrayList(0, 2, 3, 4, 5),
         newArrayList(1, 3, 4, 6),
         equalTo("[0, 2, 5]"),
+        equalTo("[0, 2, 5, null]"),
         JoinType.ANTI);
     // Matching duplicated keys
     testIntersect(
         newArrayList(1, 3, 4),
         newArrayList(1, 1, 4, 4),
         equalTo("[3]"),
+        equalTo("[3, null]"),
         JoinType.ANTI);
 
     // LEFT join tests:
@@ -291,24 +296,42 @@ class EnumerablesTest {
         newArrayList(0, 2, 4),
         newArrayList(1, 3, 5),
         equalTo("[0, 2, 4]"),
+        equalTo("[0, 2, 4, null]"),
         JoinType.ANTI);
     // Left empty
     testIntersect(
         new ArrayList<>(),
         newArrayList(1, 3, 4, 6),
         equalTo("[]"),
+        equalTo("[null]"),
         JoinType.ANTI);
     // Right empty
     testIntersect(
         newArrayList(3, 7),
         new ArrayList<>(),
         equalTo("[3, 7]"),
+        equalTo("[3, 7, null]"),
         JoinType.ANTI);
     // Both empty
     testIntersect(
         new ArrayList<Integer>(),
         new ArrayList<>(),
         equalTo("[]"),
+        equalTo("[null]"),
+        JoinType.ANTI);
+    // Left reaches NULL keys after right runs out
+    testIntersect(
+        newArrayList(1, 2, null, null),
+        newArrayList(1),
+        equalTo("[2, null, null]"),
+        equalTo("[2, null, null, null]"),
+        JoinType.ANTI);
+    // Left reaches NULL keys in advanceLeft, immediately after a match
+    testIntersect(
+        newArrayList(1, null, null),
+        newArrayList(1, 2),
+        equalTo("[null, null]"),
+        equalTo("[null, null, null]"),
         JoinType.ANTI);
 
     // LEFT join tests:
@@ -636,7 +659,8 @@ private static <T extends Comparable<T>> Enumerable<String> intersect(
             (v0, v1) -> v0,
             JoinType.ANTI,
             null, null).toList(),
-        hasToString("[Emp(30, Fred), Emp(20, Sebastian), Emp(20, Zoey)]"));
+        hasToString("[Emp(30, Fred), Emp(20, Sebastian), Emp(20, Zoey), "
+            + "Emp(40, null), Emp(30, null)]"));
   }
 
   @Test void testMergeLeftJoin() {

core/src/test/java/org/apache/calcite/test/enumerable/EnumerableJoinTest.java

@@ -336,6 +336,30 @@ class EnumerableJoinTest {
         "empid=7");
   }
 
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7817">[CALCITE-7817]
+   * Enumerable merge ANTI join drops left rows with NULL keys</a>. */
+  @Test void testMergeJoinAntiWithSingleKeyAndNullValues() {
+    tester(false, new HrSchema())
+        .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner -> {
+          planner.addRule(EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE);
+          planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE);
+        })
+        .withRel(builder -> builder
+            .scan("s", "emps").as("e1")
+            .sort(builder.field("commission"))
+            .scan("s", "emps").as("e2")
+            .filter(builder.equals(builder.field("deptno"), builder.literal(20)))
+            .sort(builder.field("commission"))
+            .antiJoin(
+                builder.equals(builder.field(2, 0, "commission"),
+                    builder.field(2, 1, "commission")))
+            .project(builder.field("empid"))
+            .build())
+        .explainHookContains("EnumerableMergeJoin")
+        .returnsUnordered("empid=100", "empid=110", "empid=150");
+  }
+
   private void checkMergeJoinWithCompositeKeyAndNullValues(boolean bigSchema, JoinRelType joinType,
       String... expected) {
     CalciteAssert.AssertQuery checker =

linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java

@@ -5047,11 +5047,11 @@ private Enumerator<TInner> getRightEnumerator() {
     }
 
     /** Returns whether the left enumerator was successfully advanced to the next
-     * element, and it does not have a null key (except for LEFT join, that needs to process
-     * all elements from left. */
+     * element, and it does not have a null key (except for LEFT and ANTI joins, which need to
+     * process all elements from left). */
     private boolean leftMoveNext() {
       return getLeftEnumerator().moveNext()
-          && (joinType == JoinType.LEFT
+          && (isLeftOrAntiJoin()
               || outerKeySelector.apply(getLeftEnumerator().current()) != null);
     }
 
@@ -5123,9 +5123,9 @@ private boolean advance() {
         // iterate until finding matching keys (or ANTI join results)
         for (;;) {
           // mergeJoin assumes inputs sorted in ascending order with nulls last,
-          // if we reach a null key, we are done.
+          // if we reach a null key, there are no more matches.
           if (leftKey == null || rightKey == null) {
-            if (joinType == JoinType.LEFT || (joinType == JoinType.ANTI && leftKey != null)) {
+            if (isLeftOrAntiJoin()) {
               // all remaining items in left are results for left/anti join
               remainingLeft = true;
               return true;
@@ -5226,17 +5226,19 @@ private boolean advance() {
      * is over or a different key is found.
      *
      * @return {@code true} if there are still elements to be processed on the left enumerator,
-     * {@code false} otherwise (left enumerator is over or null key is found).
+     * {@code false} otherwise (left enumerator is over, or null key is found and the join
+     * is not LEFT or ANTI).
      */
     private boolean advanceLeft(TSource left, TKey leftKey) {
       lefts.clear();
       lefts.add(left);
       while (getLeftEnumerator().moveNext()) {
         left = getLeftEnumerator().current();
         TKey leftKey2 = outerKeySelector.apply(left);
-        if (leftKey2 == null && joinType != JoinType.LEFT) {
+        if (leftKey2 == null && !isLeftOrAntiJoin()) {
           // mergeJoin assumes inputs sorted in ascending order with nulls last,
-          // if we reach a null key, we are done (except LEFT join, that needs to process LHS fully)
+          // if we reach a null key, we are done (except LEFT and ANTI joins, which
+          // process LHS fully)
           break;
         }
         if (!compareEquals(leftKey, leftKey2)) {