HIVE-29176: Wrong result when HiveAntiJoin is replacing an IS NULL filter on a nullable column

Author: thomasrebele

ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveCalciteUtil.java

@@ -1233,43 +1233,17 @@ public FixNullabilityShuttle(RexBuilder rexBuilder,
   }
 
   /**
-   * Checks if any of the expression given as list expressions are from right side of the join.
-   *  This is used during anti join conversion.
-   *
-   * @param joinRel Join node whose right side has to be searched.
-   * @param expressions The list of expression to search.
-   * @return true if any of the expressions is from right side of join.
+   * Given a join, creates a bitset of the joined columns originating from the right-hand side.
+   * @param joinRel a join that concatenates all columns from its inputs (so no semi-join)
+   * @return a bitset
    */
-  public static boolean hasAnyExpressionFromRightSide(RelNode joinRel, List<RexNode> expressions)  {
-    List<RelDataTypeField> joinFields = joinRel.getRowType().getFieldList();
-    int nTotalFields = joinFields.size();
-    List<RelDataTypeField> leftFields = (joinRel.getInputs().get(0)).getRowType().getFieldList();
-    int nFieldsLeft = leftFields.size();
-    ImmutableBitSet rightBitmap = ImmutableBitSet.range(nFieldsLeft, nTotalFields);
-
-    for (RexNode node : expressions) {
-      ImmutableBitSet inputBits = RelOptUtil.InputFinder.bits(node);
-      if (rightBitmap.contains(inputBits)) {
-        return true;
-      }
-    }
-    return false;
-  }
-
-  public static boolean hasAllExpressionsFromRightSide(RelNode joinRel, List<RexNode> expressions) {
-    List<RelDataTypeField> joinFields = joinRel.getRowType().getFieldList();
-    int nTotalFields = joinFields.size();
-    List<RelDataTypeField> leftFields = (joinRel.getInputs().get(0)).getRowType().getFieldList();
-    int nFieldsLeft = leftFields.size();
-    ImmutableBitSet rightBitmap = ImmutableBitSet.range(nFieldsLeft, nTotalFields);
-
-    for (RexNode node : expressions) {
-      ImmutableBitSet inputBits = RelOptUtil.InputFinder.bits(node);
-      if (!rightBitmap.contains(inputBits)) {
-        return false;
-      }
+  public static ImmutableBitSet getRightSideBitset(RelNode joinRel) {
+    if(joinRel.getInputs().size() != 2) {
+      throw new IllegalArgumentException("The relation must have exactly two children:\n" + RelOptUtil.toString(joinRel));
     }
-    return true;
+    int nTotalFields = joinRel.getRowType().getFieldCount();
+    int nFieldsLeft = (joinRel.getInputs().get(0)).getRowType().getFieldCount();
+    return ImmutableBitSet.range(nFieldsLeft, nTotalFields);
   }
 
   /**

ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveAntiSemiJoinRule.java

@@ -26,11 +26,15 @@
 import org.apache.calcite.rel.core.Join;
 import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexVisitorImpl;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Util;
+import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil;
 import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveAntiJoin;
 import org.slf4j.Logger;
@@ -39,7 +43,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Optional;
 
 /**
  * Planner rule that converts a join plus filter to anti join.
@@ -86,14 +90,17 @@ protected void perform(RelOptRuleCall call, Project project, Filter filter, Join
 
     assert (filter != null);
 
-    List<RexNode> filterList = getResidualFilterNodes(filter, join);
-    if (filterList == null) {
+    ImmutableBitSet rhsFields = HiveCalciteUtil.getRightSideBitset(join);
+    Optional<List<RexNode>> optFilterList = getResidualFilterNodes(filter, join, rhsFields);
+    if (optFilterList.isEmpty()) {
       return;
     }
+    List<RexNode> filterList = optFilterList.get();
 
     // If any projection is there from right side, then we can not convert to anti join.
-    boolean hasProjection = HiveCalciteUtil.hasAnyExpressionFromRightSide(join, project.getProjects());
-    if (hasProjection) {
+    ImmutableBitSet projectedFields = RelOptUtil.InputFinder.bits(project.getProjects(), null);
+    boolean projectionUsesRHS = projectedFields.intersects(rhsFields);
+    if (projectionUsesRHS) {
       return;
     }
 
@@ -119,13 +126,14 @@ protected void perform(RelOptRuleCall call, Project project, Filter filter, Join
   /**
    * Extracts the non-null filter conditions from given filter node.
    *
-   * @param filter The filter condition to be checked.
-   * @param join Join node whose right side has to be searched.
+   * @param filter    The filter condition to be checked.
+   * @param join      Join node whose right side has to be searched.
+   * @param rhsFields
    * @return null : Anti join condition is not matched for filter.
-   *         Empty list : No residual filter conditions present.
-   *         Valid list containing the filter to be applied after join.
+   *     Empty list : No residual filter conditions present.
+   *     Valid list containing the filter to be applied after join.
    */
-  private List<RexNode> getResidualFilterNodes(Filter filter, Join join) {
+  private Optional<List<RexNode>> getResidualFilterNodes(Filter filter, Join join, ImmutableBitSet rhsFields) {
     // 1. If null filter is not present from right side then we can not convert to anti join.
     // 2. If any non-null filter is present from right side, we can not convert it to anti join.
     // 3. Keep other filters which needs to be executed after join.
@@ -135,43 +143,76 @@ private List<RexNode> getResidualFilterNodes(Filter filter, Join join) {
     List<RexNode> aboveFilters = RelOptUtil.conjunctions(filter.getCondition());
     boolean hasNullFilterOnRightSide = false;
     List<RexNode> filterList = new ArrayList<>();
+    final ImmutableBitSet notNullColumnsFromRightSide = getNotNullColumnsFromRightSide(join);
+
     for (RexNode filterNode : aboveFilters) {
-      if (filterNode.getKind() == SqlKind.IS_NULL) {
-        // Null filter from right side table can be removed and its a pre-condition for anti join conversion.
-        if (HiveCalciteUtil.hasAllExpressionsFromRightSide(join, Collections.singletonList(filterNode))
-            && isStrong(((RexCall) filterNode).getOperands().get(0))) {
-          hasNullFilterOnRightSide = true;
-        } else {
-          filterList.add(filterNode);
-        }
-      } else {
-        if (HiveCalciteUtil.hasAnyExpressionFromRightSide(join, Collections.singletonList(filterNode))) {
-          // If some non null condition is present from right side, we can not convert the join to anti join as
-          // anti join does not project the fields from right side.
-          return null;
-        } else {
-          filterList.add(filterNode);
-        }
+      final ImmutableBitSet usedFields = RelOptUtil.InputFinder.bits(filterNode);
+      boolean usesFieldFromRHS = usedFields.intersects(rhsFields);
+
+      if(!usesFieldFromRHS) {
+        // Only LHS fields or constants, so the filterNode is part of the residual filter
+        filterList.add(filterNode);
+        continue;
+      }
+
+      // In the following we check for filter nodes that let us deduce that
+      // "an (originally) not-null column of RHS IS NULL because the LHS row will not be matched"
+
+      if(filterNode.getKind() != SqlKind.IS_NULL) {
+        return Optional.empty();
+      }
+
+      boolean usesRHSFieldsOnly = rhsFields.contains(usedFields);
+      if (!usesRHSFieldsOnly) {
+        // If there is a mix between LHS and RHS fields, don't convert to anti-join
+        return Optional.empty();
+      }
+
+      // Null filter from right side table can be removed and it is a pre-condition for anti join conversion.
+      RexNode arg = ((RexCall) filterNode).getOperands().get(0);
+      if (isStrong(arg, notNullColumnsFromRightSide)) {
+        hasNullFilterOnRightSide = true;
+      } else if(!isStrong(arg, rhsFields)) {
+        // if all RHS fields are null and the IS NULL is still not fulfilled, bail out
+        return Optional.empty();
       }
     }
 
     if (!hasNullFilterOnRightSide) {
-      return null;
+      return Optional.empty();
     }
-    return filterList;
+    return Optional.of(filterList);
   }
 
-  private boolean isStrong(RexNode rexNode) {
-    AtomicBoolean hasCast = new AtomicBoolean(false);
-    rexNode.accept(new RexVisitorImpl<Void>(true) {
-      @Override
-      public Void visitCall(RexCall call) {
-        if (call.getKind() == SqlKind.CAST) {
-          hasCast.set(true);
-        }
-        return super.visitCall(call);
+  private ImmutableBitSet getNotNullColumnsFromRightSide(RelNode joinRel) {
+    // we need to shift the indices of the second child to the right
+    int shift = (joinRel.getInput(0)).getRowType().getFieldCount();
+
+    ImmutableBitSet.Builder builder = ImmutableBitSet.builder();
+    List<RelDataTypeField> rhsFields = joinRel.getInput(1).getRowType().getFieldList();
+    for(RelDataTypeField field : rhsFields) {
+      if(!field.getType().isNullable()) {
+        builder.set(shift+field.getIndex());
       }
-    });
-    return !hasCast.get() && Strong.isStrong(rexNode);
+    }
+    return builder.build();
+  }
+
+  private boolean isStrong(RexNode rexNode, ImmutableBitSet rightSideBitset) {
+    try {
+      rexNode.accept(new RexVisitorImpl<Void>(true) {
+        @Override
+        public Void visitCall(RexCall call) {
+          if (call.getKind() == SqlKind.CAST) {
+            throw Util.FoundOne.NULL;
+          }
+          return super.visitCall(call);
+        }
+      });
+    } catch (Util.FoundOne e) {
+      // Hive's CAST might introduce NULL for NOT NULL fields
+      return false;
+    }
+    return Strong.isNull(rexNode, rightSideBitset);
   }
 }

ql/src/test/org/apache/hadoop/hive/ql/optimizer/calcite/rules/TestHiveAntiSemiJoinRule.java

@@ -0,0 +1,168 @@
+/*
+ * 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.hadoop.hive.ql.optimizer.calcite.rules;
+
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.runtime.Hook;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.tools.RelBuilder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.util.Collections;
+
+import static org.apache.hadoop.hive.ql.optimizer.calcite.rules.TestRuleHelper.*;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestHiveAntiSemiJoinRule {
+
+  PlanFixture fixture() {
+    RelOptPlanner planner = buildPlanner(Collections.singletonList(HiveAntiSemiJoinRule.INSTANCE));
+    return new PlanFixture(planner).registerTable("t1", T1Record.class).registerTable("t2", T2Record.class);
+  }
+
+  @Test
+  public void testFilterOnNullableColumn() {
+    PlanFixture fixture = fixture();
+    RelBuilder b = fixture.createRelBuilder();
+
+    // @formatter:off
+    RelNode plan = b
+        .scan("t1")
+        .scan("t2")
+        .join(JoinRelType.LEFT, b.equals(
+            b.field(2, 0, "t1nullable"),
+            b.field(2, 1, "t2id")))
+        .filter(b.isNull(b.field("t2nullable")))
+        .project(b.field("t1id"))
+        .build();
+
+    String expectedPlan = "HiveProject(t1id=[$0])\n"
+                        + "  HiveFilter(condition=[IS NULL($5)])\n"
+                        + "    HiveJoin(condition=[=($2, $3)], joinType=[left], algorithm=[none], cost=[not available])\n"
+                        + "      LogicalTableScan(table=[[t1]])\n"
+                        + "      LogicalTableScan(table=[[t2]])\n";
+    // @formatter:on
+
+    assertPlans(fixture.getPlanner(), plan, expectedPlan, expectedPlan);
+  }
+
+  @Test
+  public void testFilterIsNullFromBothSides() {
+    PlanFixture fixture = fixture();
+
+    RelNode plan;
+    try (Hook.Closeable ignore = Hook.REL_BUILDER_SIMPLIFY.addThread(Hook.propertyJ(false))) {
+      RelBuilder b = fixture.createRelBuilder();
+      // @formatter:off
+      plan = b.scan("t1")
+              .scan("t2")
+              .join(JoinRelType.LEFT, b.equals(b.field(2, 0, "t1nullable"), b.field(2, 1, "t2id")))
+              .filter(b.isNull(b.call(SqlStdOperatorTable.PLUS, b.field("t2nullable"), b.field("t1nullable"))))
+              .project(b.field("t1id")).build();
+      // @formatter:on
+    }
+
+    // @formatter:off
+    String expectedPlan = "HiveProject(t1id=[$0])\n"
+                        + "  HiveFilter(condition=[IS NULL(+($5, $2))])\n"
+                        + "    HiveJoin(condition=[=($2, $3)], joinType=[left], algorithm=[none], cost=[not available])\n"
+                        + "      LogicalTableScan(table=[[t1]])\n"
+                        + "      LogicalTableScan(table=[[t2]])\n";
+    // @formatter:on
+
+    assertPlans(fixture.getPlanner(), plan, expectedPlan, expectedPlan);
+  }
+
+  @Test
+  public void testFilterOnNotNullColumn() {
+    PlanFixture fixture = fixture();
+    RelBuilder b = fixture.createRelBuilder();
+
+    // @formatter:off
+    RelNode plan = b
+        .scan("t1")
+        .scan("t2")
+        .join(JoinRelType.LEFT, b.equals(
+            b.field(2, 0, "t1nullable"),
+            b.field(2, 1, "t2id")))
+        .filter(b.isNull(b.field("t2notnull")))
+        .project(b.field("t1id"))
+        .build();
+
+    String prePlan = "HiveProject(t1id=[$0])\n"
+                   + "  HiveFilter(condition=[IS NULL($4)])\n"
+                   + "    HiveJoin(condition=[=($2, $3)], joinType=[left], algorithm=[none], cost=[not available])\n"
+                   + "      LogicalTableScan(table=[[t1]])\n"
+                   + "      LogicalTableScan(table=[[t2]])\n";
+
+    String postPlan = "HiveProject(t1id=[$0])\n"
+                   + "  HiveAntiJoin(condition=[=($2, $3)], joinType=[anti])\n"
+                   + "    LogicalTableScan(table=[[t1]])\n"
+                   + "    LogicalTableScan(table=[[t2]])\n";
+    // @formatter:on
+
+    assertPlans(fixture.getPlanner(), plan, prePlan, postPlan);
+  }
+
+  @Test
+  public void testFilterOnNullAndNotNullColumn() {
+    PlanFixture fixture = fixture();
+    RelBuilder b = fixture.createRelBuilder();
+
+    // @formatter:off
+    RelNode plan = b
+        .scan("t1")
+        .scan("t2")
+        .join(JoinRelType.LEFT, b.equals(
+            b.field(2, 0, "t1nullable"),
+            b.field(2, 1, "t2id")))
+        .filter(b.and(b.isNull(b.field("t2notnull")), b.isNull((b.field("t2nullable")))))
+        .project(b.field("t1id"))
+        .build();
+
+    String prePlan = "HiveProject(t1id=[$0])\n"
+        + "  HiveFilter(condition=[AND(IS NULL($4), IS NULL($5))])\n"
+        + "    HiveJoin(condition=[=($2, $3)], joinType=[left], algorithm=[none], cost=[not available])\n"
+        + "      LogicalTableScan(table=[[t1]])\n"
+        + "      LogicalTableScan(table=[[t2]])\n";
+
+    String postPlan = "HiveProject(t1id=[$0])\n"
+        + "  HiveAntiJoin(condition=[=($2, $3)], joinType=[anti])\n"
+        + "    LogicalTableScan(table=[[t1]])\n"
+        + "    LogicalTableScan(table=[[t2]])\n";
+    // @formatter:on
+
+    assertPlans(fixture.getPlanner(), plan, prePlan, postPlan);
+  }
+
+  static class T1Record {
+    public int t1id;
+    public int t1notnull;
+    public Integer t1nullable;
+  }
+
+  static class T2Record {
+    public int t2id;
+    public int t2notnull;
+    public Integer t2nullable;
+  }
+}