[CALCITE-7757] RelMdFunctionalDependency can infer unsound dependencies and cause incorrect query results

core/src/main/java/org/apache/calcite/rel/metadata/RelMdFunctionalDependency.java

@@ -24,16 +24,19 @@
 import org.apache.calcite.rel.core.Correlate;
 import org.apache.calcite.rel.core.Filter;
 import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
 import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.calcite.rel.core.Project;
 import org.apache.calcite.rel.core.SetOp;
 import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeUtil;
 import org.apache.calcite.util.Arrow;
 import org.apache.calcite.util.ArrowSet;
 import org.apache.calcite.util.ImmutableBitSet;
@@ -272,7 +275,8 @@
         ImmutableBitSet bitSet = expr instanceof RexInputRef
             ? ImmutableBitSet.of(((RexInputRef) expr).getIndex())
             : inputBits[i];
-        if (inputFdSet.implies(refIndex, bitSet)) {
+        if (typeSupportsGroupKeyInference(k.getType())
+            && inputFdSet.implies(refIndex, bitSet)) {
           fdBuilder.addArrow(v, i);
         }
       });
@@ -292,7 +296,7 @@
 
       // Map all determinant columns
       ImmutableBitSet mappedDeterminants = mapAllCols(determinants, mapping);
-      if (mappedDeterminants.isEmpty()) {
+      if (mappedDeterminants.isEmpty() && !determinants.isEmpty()) {
         continue;
       }
 
@@ -345,26 +349,22 @@
     ArrowSet inputFdSet = mq.getFDs(rel.getInput());
 
     ImmutableBitSet groupSet = rel.getGroupSet();
+    Mappings.TargetMapping inputToOutputMap =
+        Mappings.target(groupSet::indexOf,
+            rel.getInput().getRowType().getFieldCount(), rel.getGroupCount());
 
     // Preserve input FDs that only involve group columns
     if (Aggregate.isSimple(rel)) {
-      for (Arrow inputFd : inputFdSet.getArrows()) {
-        ImmutableBitSet determinants = inputFd.getDeterminants();
-        ImmutableBitSet dependents = inputFd.getDependents();
-
-        // Only preserve if both determinants and dependents are within group columns
-        if (groupSet.contains(determinants) && groupSet.contains(dependents)) {
-          fdBuilder.addArrow(determinants, dependents);
-        }
-      }
+      mapInputFDs(inputFdSet, inputToOutputMap, fdBuilder);
 
       // Add transitive dependencies within group columns
       for (int groupCol : groupSet) {
         ImmutableBitSet singleton = ImmutableBitSet.of(groupCol);
         ImmutableBitSet closure = inputFdSet.dependents(singleton);
         ImmutableBitSet groupDependents = closure.intersect(groupSet).except(singleton);
         if (!groupDependents.isEmpty()) {
-          fdBuilder.addArrow(singleton, groupDependents);
+          fdBuilder.addArrow(mapAllCols(singleton, inputToOutputMap),
+              mapAllCols(groupDependents, inputToOutputMap));
         }
       }
     }
@@ -373,7 +373,7 @@
     if (!groupSet.isEmpty() && !rel.getAggCallList().isEmpty()) {
       ImmutableBitSet aggCols =
           ImmutableBitSet.range(rel.getGroupCount(), rel.getRowType().getFieldCount());
-      fdBuilder.addArrow(groupSet, aggCols);
+      fdBuilder.addArrow(ImmutableBitSet.range(rel.getGroupCount()), aggCols);
     }
 
     return fdBuilder.build();
@@ -387,7 +387,7 @@
     ArrowSet.Builder fdBuilder = new ArrowSet.Builder();
 
     // Adds equality dependencies from filter conditions.
-    addFDsFromEqualityCondition(rel.getCondition(), fdBuilder);
+    addBidirectionalFDsFromEqualityCondition(rel.getCondition(), fdBuilder);
 
     return fdBuilder.build().union(inputSet);
   }
@@ -407,9 +407,12 @@
     case INNER:
     case LEFT:
     case RIGHT:
-      ArrowSet.Builder joinFdBuilder = new ArrowSet.Builder()
-          .addArrowSet(leftFdSet.union(shiftFdSet(rightFdSet, leftFieldCount)));
-      addFDsFromEqualityCondition(rel.getCondition(), joinFdBuilder);
+      ArrowSet.Builder joinFdBuilder = new ArrowSet.Builder();
+      addJoinInputFDs(leftFdSet, rel.getLeft(), 0,
+          joinType.generatesNullsOnLeft(), joinFdBuilder);
+      addJoinInputFDs(rightFdSet, rel.getRight(), leftFieldCount,
+          joinType.generatesNullsOnRight(), joinFdBuilder);
+      addFDsFromJoinCondition(rel, leftFieldCount, joinFdBuilder);
       return joinFdBuilder.build();
     case SEMI:
     case ANTI:
@@ -419,6 +422,41 @@
     }
   }
 
+  /**
+   * Copies input dependencies into a join, optionally filtering dependencies
+   * that can be invalidated when the input is null-generated.
+   */
+  private static void addJoinInputFDs(ArrowSet inputFdSet, RelNode input,
+      int offset, boolean nullGenerated, ArrowSet.Builder fdBuilder) {
+    for (Arrow inputFd : inputFdSet.getArrows()) {
+      if (nullGenerated && !hasNonNullableDeterminant(inputFd, input)) {
+        continue;
+      }
+      fdBuilder.addArrow(inputFd.getDeterminants().shift(offset),
+          inputFd.getDependents().shift(offset));
+    }
+  }
+
+  /**
+   * Returns whether a dependency's determinant contains a non-nullable input
+   * field. Such a determinant cannot collide with the all-NULL determinant of
+   * a padded outer-join row.
+   */
+  private static boolean hasNonNullableDeterminant(Arrow fd, RelNode input) {
+    final int fieldCount = input.getRowType().getFieldCount();
+    boolean hasNonNullableField = false;
+    for (int determinant : fd.getDeterminants()) {
+      if (determinant >= fieldCount) {
+        return false;
+      }
+      if (!input.getRowType().getFieldList().get(determinant)
+          .getType().isNullable()) {
+        hasNonNullableField = true;
+      }
+    }
+    return hasNonNullableField;
+  }
+
   /**
    * Returns functional dependencies for Calc.
    */
@@ -428,50 +466,93 @@
   }
 
   /**
-   * Shifts column indices in functional dependencies (for right table in Joins).
-   *
-   * @param fdSet Functional dependency set
-   * @param offset Index offset
-   * @return Shifted functional dependency set
+   * Adds functional dependencies implied by a join condition.
    */
-  private ArrowSet shiftFdSet(ArrowSet fdSet, int offset) {
-    ArrowSet.Builder shiftedFdSetBuilder = new ArrowSet.Builder();
-    for (Arrow fd : fdSet.getArrows()) {
-      ImmutableBitSet shiftedDeterminants = fd.getDeterminants().shift(offset);
-      ImmutableBitSet shiftedDependents = fd.getDependents().shift(offset);
-      shiftedFdSetBuilder.addArrow(shiftedDeterminants, shiftedDependents);
+  private static void addFDsFromJoinCondition(Join rel, int leftFieldCount,
+      ArrowSet.Builder builder) {
+    final JoinRelType joinType = rel.getJoinType();
+    if (joinType == JoinRelType.INNER) {
+      addBidirectionalFDsFromEqualityCondition(rel.getCondition(), builder);
+      return;
     }
-    return shiftedFdSetBuilder.build();
+
+    if (joinType == JoinRelType.LEFT || joinType == JoinRelType.RIGHT) {
+      final JoinInfo joinInfo = rel.analyzeCondition();
+      if (!joinInfo.isEqui() || joinInfo.leftKeys.isEmpty()
+          || !fieldsSupportEqualityInference(rel.getLeft(), joinInfo.leftSet())
+          || !fieldsSupportEqualityInference(rel.getRight(), joinInfo.rightSet())) {
+        return;
+      }
+
+      final ImmutableBitSet leftKeys = joinInfo.leftSet();
+      final ImmutableBitSet rightKeys = joinInfo.rightSet().shift(leftFieldCount);
+      if (joinType == JoinRelType.LEFT) {
+        builder.addArrow(leftKeys, rightKeys);
+      } else {
+        builder.addArrow(rightKeys, leftKeys);
+      }
+      return;
+    }
+
+    throw new AssertionError("unsupported join type: " + joinType);
   }
 
   /**
-   * Extracts functional dependencies from equality and AND conditions.
-   * Handles col1 = col2, col1 IS NOT DISTINCT FROM col2, and AND conditions.
+   * Adds bidirectional dependencies for input-reference equalities in a
+   * condition. Callers are responsible for ensuring that every output row
+   * satisfies the condition, as is true for Filters and inner joins.
    */
-  private static void addFDsFromEqualityCondition(RexNode condition, ArrowSet.Builder builder) {
-    if (!(condition instanceof RexCall)) {
-      return;
-    }
+  private static void addBidirectionalFDsFromEqualityCondition(
+      RexNode condition, ArrowSet.Builder builder) {
+    for (RexNode conjunct : RelOptUtil.conjunctions(condition)) {
+      if (!(conjunct instanceof RexCall)) {
+        continue;
+      }
 
-    RexCall call = (RexCall) condition;
-    if (call.getOperator().getKind() == SqlKind.EQUALS
-        || call.getOperator().getKind() == SqlKind.IS_NOT_DISTINCT_FROM) {
+      RexCall call = (RexCall) conjunct;
+      if (call.getOperator().getKind() != SqlKind.EQUALS
+          && call.getOperator().getKind() != SqlKind.IS_NOT_DISTINCT_FROM) {
+        continue;
+      }
       List<RexNode> operands = call.getOperands();
       if (operands.size() == 2) {
         RexNode left = operands.get(0);
         RexNode right = operands.get(1);
 
-        if (left instanceof RexInputRef && right instanceof RexInputRef) {
+        if (left instanceof RexInputRef && right instanceof RexInputRef
+            && typeSupportsGroupKeyInference(left.getType())
+            && typeSupportsGroupKeyInference(right.getType())) {
           int leftRef = ((RexInputRef) left).getIndex();
           int rightRef = ((RexInputRef) right).getIndex();
 
           builder.addBidirectionalArrow(leftRef, rightRef);
         }
       }
-    } else if (call.getOperator().getKind() == SqlKind.AND) {
-      for (RexNode operand : call.getOperands()) {
-        addFDsFromEqualityCondition(operand, builder);
+    }
+  }
+
+  /**
+   * Returns whether equality on the given fields can safely imply a functional
+   * dependency for grouping purposes.
+   */
+  private static boolean fieldsSupportEqualityInference(RelNode input,
+      ImmutableBitSet fields) {
+    for (int field : fields) {
+      if (!typeSupportsGroupKeyInference(
+          input.getRowType().getFieldList().get(field).getType())) {
+        return false;
       }
     }
+    return true;
+  }
+
+  /**
+   * Returns whether a type can safely be used to infer that one grouping key
+   * determines another. Approximate numerics and intervals are unsafe,
+   * including when nested in rows, collections, or maps.
+   */
+  private static boolean typeSupportsGroupKeyInference(RelDataType type) {
+    return !SqlTypeUtil.containsType(type,
+        t -> SqlTypeUtil.isApproximateNumeric(t) || SqlTypeUtil.isInterval(t));
   }
 }

core/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java

@@ -62,6 +62,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -405,6 +406,36 @@ public static boolean containsNullable(RelDataType type) {
     return false;
   }
 
+  /**
+   * Returns whether a type or any type nested within its fields, collection
+   * component, map key, or map value matches a predicate.
+   */
+  public static boolean containsType(RelDataType type,
+      Predicate<? super RelDataType> predicate) {
+    requireNonNull(type, "type");
+    requireNonNull(predicate, "predicate");
+    if (predicate.test(type)) {
+      return true;
+    }
+    if (type.isStruct()) {
+      for (RelDataTypeField field : type.getFieldList()) {
+        if (containsType(field.getType(), predicate)) {
+          return true;
+        }
+      }
+    }
+    final RelDataType componentType = type.getComponentType();
+    if (componentType != null && containsType(componentType, predicate)) {
+      return true;
+    }
+    final RelDataType keyType = type.getKeyType();
+    if (keyType != null && containsType(keyType, predicate)) {
+      return true;
+    }
+    final RelDataType valueType = type.getValueType();
+    return valueType != null && containsType(valueType, predicate);
+  }
+
   /**
    * Creates a RelDataType having the same type of the sourceRelDataType,
    * and the same nullability as the targetRelDataType.

core/src/test/java/org/apache/calcite/sql/type/SqlTypeUtilTest.java

@@ -122,6 +122,38 @@ private RelDataType struct(RelDataType...relDataTypes) {
     return builder.build();
   }
 
+  @Test void testContainsType() {
+    assertThat(
+        SqlTypeUtil.containsType(f.sqlInt,
+        SqlTypeUtil::isApproximateNumeric), is(false));
+    assertThat(
+        SqlTypeUtil.containsType(f.sqlFloat,
+        SqlTypeUtil::isApproximateNumeric), is(true));
+    assertThat(
+        SqlTypeUtil.containsType(struct(f.sqlInt, f.arrayFloat),
+        SqlTypeUtil::isApproximateNumeric), is(true));
+    assertThat(
+        SqlTypeUtil.containsType(f.arrayOfArrayFloat,
+        SqlTypeUtil::isApproximateNumeric), is(true));
+    assertThat(
+        SqlTypeUtil.containsType(f.multisetFloat,
+        SqlTypeUtil::isApproximateNumeric), is(true));
+
+    final RelDataType floatKeyMap =
+        f.typeFactory.createMapType(f.sqlFloat, f.sqlInt);
+    final RelDataType floatValueMap =
+        f.typeFactory.createMapType(f.sqlInt, f.sqlFloat);
+    assertThat(
+        SqlTypeUtil.containsType(floatKeyMap,
+        SqlTypeUtil::isApproximateNumeric), is(true));
+    assertThat(
+        SqlTypeUtil.containsType(floatValueMap,
+        SqlTypeUtil::isApproximateNumeric), is(true));
+    assertThat(
+        SqlTypeUtil.containsType(struct(f.arrayBigInt, f.mapOfInt),
+        SqlTypeUtil::isApproximateNumeric), is(false));
+  }
+
   @Test void testModifyTypeCoercionMappings() {
     SqlTypeMappingRules.Builder builder = SqlTypeMappingRules.builder();
     final SqlTypeCoercionRule defaultRules = SqlTypeCoercionRule.instance();