[FLINK-40936][table] Reject VARIANT elements and keys in functions and operators that compare them

docs/content.zh/docs/sql/reference/data-types.md

@@ -1726,16 +1726,8 @@ CAST(ARRAY[1, 2] AS VARIANT)                 -- fails at validation, not support
 A `VARIANT` has no equality and no order, because one value has many valid binary encodings. The
 objects `{"a": 1, "b": 2}` and `{"b": 2, "a": 1}` hold the same value, and so do a `1` stored as a
 `TINYINT` and a `1` stored as a `BIGINT`. So a query cannot use a `VARIANT`, or a type that contains
-one, where it compares values:
-
-- as a grouping key, including `GROUPING SETS`, window aggregations, and `SELECT DISTINCT`
-- as the argument of a `DISTINCT` aggregate, such as `COUNT(DISTINCT v)`
-- in `UNION`, `INTERSECT`, `INTERSECT ALL`, `EXCEPT`, and `EXCEPT ALL`, while `UNION ALL` is allowed
-- as an `ORDER BY` key
-- as a `PARTITION BY` or `ORDER BY` key of an `OVER` window, of `MATCH_RECOGNIZE`, or of a table
-  argument
-- as an operand of a comparison such as `=`, `<>`, `<`, `IS DISTINCT FROM`, or `IN`, which includes
-  join conditions
+one, where it compares values. Examples are a `GROUP BY`, `ORDER BY`, or join key,
+`SELECT DISTINCT`, a comparison such as `=` or `<`, and the array elements of `ARRAY_CONTAINS`.
 
 Such a query fails before it runs. Cast the value to a concrete type first, or use `TRY_CAST` when
 some values do not fit the type. A `VARIANT` can still be selected, read with field access, checked

docs/content/docs/sql/reference/data-types.md

@@ -1769,16 +1769,8 @@ CAST(ARRAY[INTERVAL '1' DAY] AS VARIANT)     -- fails at validation
 A `VARIANT` has no equality and no order, because one value has many valid binary encodings. The
 objects `{"a": 1, "b": 2}` and `{"b": 2, "a": 1}` hold the same value, and so do a `1` stored as a
 `TINYINT` and a `1` stored as a `BIGINT`. So a query cannot use a `VARIANT`, or a type that contains
-one, where it compares values:
-
-- as a grouping key, including `GROUPING SETS`, window aggregations, and `SELECT DISTINCT`
-- as the argument of a `DISTINCT` aggregate, such as `COUNT(DISTINCT v)`
-- in `UNION`, `INTERSECT`, `INTERSECT ALL`, `EXCEPT`, and `EXCEPT ALL`, while `UNION ALL` is allowed
-- as an `ORDER BY` key
-- as a `PARTITION BY` or `ORDER BY` key of an `OVER` window, of `MATCH_RECOGNIZE`, or of a table
-  argument
-- as an operand of a comparison such as `=`, `<>`, `<`, `IS DISTINCT FROM`, or `IN`, which includes
-  join conditions
+one, where it compares values. Examples are a `GROUP BY`, `ORDER BY`, or join key,
+`SELECT DISTINCT`, a comparison such as `=` or `<`, and the array elements of `ARRAY_CONTAINS`.
 
 Such a query fails before it runs. Cast the value to a concrete type first, or use `TRY_CAST` when
 some values do not fit the type. A `VARIANT` can still be selected, read with field access, checked

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecks.java

@@ -249,6 +249,12 @@ public static boolean hasWellDefinedString(LogicalType logicalType) {
         }
     }
 
+    /**
+     * Checks whether two types support the required comparison with each other.
+     *
+     * <p>NULL is comparable with every type. A type that is not comparable on its own is never
+     * comparable with another type, see {@link #isComparableType(LogicalType)}.
+     */
     public static boolean areComparable(
             LogicalType firstType,
             LogicalType secondType,
@@ -258,13 +264,13 @@ public static boolean areComparable(
     }
 
     /**
-     * Checks whether a (possibly nested) logical type contains no VARIANT, which every grouping,
-     * join, partition and sort key and every operand of a comparison operator needs.
+     * Checks whether values of a type can be compared at all. A type that is not comparable has
+     * neither equality nor order.
      *
-     * <p>A VARIANT value has many valid binary encodings, so two equal values can differ in their
-     * bytes.
+     * <p>VARIANT and types that contain a VARIANT are not comparable, because one VARIANT value has
+     * many valid binary encodings.
      */
-    public static boolean isComparableKeyType(LogicalType logicalType) {
+    public static boolean isComparableType(LogicalType logicalType) {
         return !hasNested(logicalType, t -> t.is(LogicalTypeRoot.VARIANT));
     }
 
@@ -283,6 +289,10 @@ private static boolean areComparableWithNormalizedNullability(
             return true;
         }
 
+        if (!isComparableType(firstType) || !isComparableType(secondType)) {
+            return false;
+        }
+
         if (firstType.getTypeRoot() == secondType.getTypeRoot()) {
             return areTypesOfSameRootComparable(firstType, secondType, requiredComparison);
         }

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ComparableInputTypeStrategyTest.java

@@ -313,6 +313,13 @@ protected Stream<TestSpec> testData() {
                         .expectErrorMessage(
                                 "All types in a comparison should support 'EQUALS' comparison with each other."
                                         + " Can not compare `cat`.`db`.`type1` with `cat`.`db`.`type2`"),
+                TestSpec.forStrategy(
+                                "VARIANT types are not comparable",
+                                SpecificInputTypeStrategies.TWO_EQUALS_COMPARABLE)
+                        .calledWithArgumentTypes(DataTypes.VARIANT(), DataTypes.VARIANT())
+                        .expectErrorMessage(
+                                "All types in a comparison should support 'EQUALS' comparison with each other."
+                                        + " Can not compare VARIANT with VARIANT"),
                 TestSpec.forStrategy(
                                 "Not comparable array types",
                                 SpecificInputTypeStrategies.TWO_FULLY_COMPARABLE)

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecksTest.java

@@ -142,13 +142,13 @@ void testFieldCountExtraction() {
     }
 
     @ParameterizedTest(name = "{0}")
-    @MethodSource("comparableKeyTypes")
-    void testIsComparableKeyType(DataType dataType, boolean expected) {
-        assertThat(LogicalTypeChecks.isComparableKeyType(dataType.getLogicalType()))
+    @MethodSource("comparableTypes")
+    void testIsComparableType(DataType dataType, boolean expected) {
+        assertThat(LogicalTypeChecks.isComparableType(dataType.getLogicalType()))
                 .isEqualTo(expected);
     }
 
-    private static Stream<Arguments> comparableKeyTypes() {
+    private static Stream<Arguments> comparableTypes() {
         return Stream.of(
                 Arguments.of(INT(), true),
                 Arguments.of(ROW(FIELD("f0", INT())), true),

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidator.java → flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/ComparableTypeValidator.java

@@ -19,10 +19,13 @@
 package org.apache.flink.table.planner.plan.optimize;
 
 import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
 import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
 import org.apache.flink.table.planner.calcite.RexTableArgCall;
 import org.apache.flink.table.planner.plan.nodes.calcite.TableAggregate;
 import org.apache.flink.table.planner.plan.nodes.calcite.WatermarkAssigner;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
 
@@ -48,29 +51,74 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlUtil;
 import org.apache.calcite.sql.fun.SqlQuantifyOperator;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.util.ImmutableBitSet;
 
 import javax.annotation.Nullable;
 
 import java.util.List;
+import java.util.Set;
 
 /**
- * Rejects queries that group, join, partition, sort or compare values of a type that is not a
- * comparable key type, see {@link LogicalTypeChecks#isComparableKeyType(LogicalType)}.
+ * Rejects a query that compares values of a type that is not comparable, see {@link
+ * LogicalTypeChecks#isComparableType(LogicalType)}.
  *
- * <p>The check runs on the logical plan before optimization. Keys that the optimizer adds later
- * only match a value with copies of itself and need no check.
+ * <p>A query checks values for equality when it groups, joins or partitions by them, applies
+ * DISTINCT or a set operation other than UNION ALL to them, or compares them with an operator such
+ * as {@code =}, {@code <>}, {@code IN} or {@code IS DISTINCT FROM}. It also checks array elements
+ * and map keys for equality when it passes them to a function such as {@code ARRAY_CONTAINS} or
+ * {@code MAP_CONTAINS_KEY}, looks up a map key with {@code m[k]}, or collects values with {@code
+ * COLLECT}. It orders values when it sorts by them with ORDER BY, or compares them with {@code <},
+ * {@code <=}, {@code >} or {@code >=}.
+ *
+ * <p>The check runs on the logical plan before optimization. Comparisons that the optimizer adds
+ * later only match a value with copies of itself and need no check.
  */
-public final class KeyTypeValidator {
+public final class ComparableTypeValidator {
 
     private static final String EQUALITY = "equality";
     private static final String ORDER = "order";
 
-    private KeyTypeValidator() {}
+    /** Operators that check their operands for equality, also in a join condition. */
+    private static final Set<SqlKind> EQUALITY_OPERATORS =
+            Set.of(
+                    SqlKind.EQUALS,
+                    SqlKind.NOT_EQUALS,
+                    SqlKind.IS_DISTINCT_FROM,
+                    SqlKind.IS_NOT_DISTINCT_FROM,
+                    SqlKind.IN,
+                    SqlKind.NOT_IN);
+
+    /** Operators that order their operands. */
+    private static final Set<SqlKind> ORDER_OPERATORS =
+            Set.of(
+                    SqlKind.LESS_THAN,
+                    SqlKind.LESS_THAN_OR_EQUAL,
+                    SqlKind.GREATER_THAN,
+                    SqlKind.GREATER_THAN_OR_EQUAL);
+
+    /** Built-in functions that compare the elements of an array or the keys of a map. */
+    private static final Set<FunctionDefinition> ELEMENT_COMPARING_FUNCTIONS =
+            Set.of(
+                    BuiltInFunctionDefinitions.ARRAY_CONTAINS,
+                    BuiltInFunctionDefinitions.ARRAY_DISTINCT,
+                    BuiltInFunctionDefinitions.ARRAY_POSITION,
+                    BuiltInFunctionDefinitions.ARRAY_REMOVE,
+                    BuiltInFunctionDefinitions.ARRAY_UNION,
+                    BuiltInFunctionDefinitions.ARRAY_EXCEPT,
+                    BuiltInFunctionDefinitions.ARRAY_INTERSECT,
+                    BuiltInFunctionDefinitions.MAP_CONTAINS_KEY,
+                    BuiltInFunctionDefinitions.MAP_UNION);
+
+    /** Operators that compare the keys of a map or the values they collect. */
+    private static final Set<SqlKind> ELEMENT_COMPARING_OPERATORS =
+            Set.of(SqlKind.ITEM, SqlKind.COLLECT);
+
+    private ComparableTypeValidator() {}
 
     /**
-     * Throws a {@link ValidationException} if a plan groups, joins, partitions, sorts or compares
-     * values of a type that is not a comparable key type.
+     * Throws a {@link ValidationException} if a plan checks values of a type that is not comparable
+     * for equality or orders them.
      */
     public static void validate(final List<RelNode> roots) {
         roots.forEach(root -> new PlanChecker().go(root));
@@ -99,6 +147,13 @@ private void checkKeys(final RelNode node) {
                                 ImmutableBitSet.of(call.getArgList()),
                                 "as an argument of a DISTINCT aggregate",
                                 EQUALITY);
+                    } else if (ELEMENT_COMPARING_OPERATORS.contains(
+                            call.getAggregation().getKind())) {
+                        checkFields(
+                                inputType,
+                                ImmutableBitSet.of(call.getArgList()),
+                                "as an argument of " + call.getAggregation().getName(),
+                                EQUALITY);
                     }
                 }
             } else if (node instanceof TableAggregate) {
@@ -164,11 +219,17 @@ public RexNode visitCall(final RexCall call) {
                         ImmutableBitSet.of(tableArg.getOrderKeys()),
                         "as an ORDER BY key of a table argument",
                         ORDER);
-            } else if (call.isA(SqlKind.BINARY_COMPARISON) || call.isA(SqlKind.COMPARISON)) {
-                checkComparison(
-                        call.getOperands(),
-                        call.getOperator().getName(),
-                        call.isA(SqlKind.ORDER_COMPARISON) ? ORDER : EQUALITY);
+            } else if (call.isA(EQUALITY_OPERATORS)) {
+                checkComparison(call.getOperands(), call.getOperator().getName(), EQUALITY);
+            } else if (call.isA(ORDER_OPERATORS)) {
+                checkComparison(call.getOperands(), call.getOperator().getName(), ORDER);
+            } else if (call.isA(ELEMENT_COMPARING_OPERATORS)) {
+                checkElements(call);
+            } else {
+                final FunctionDefinition definition = ShortcutUtils.unwrapFunctionDefinition(call);
+                if (definition != null && ELEMENT_COMPARING_FUNCTIONS.contains(definition)) {
+                    checkElements(call);
+                }
             }
             return super.visitCall(call);
         }
@@ -183,6 +244,14 @@ public RexNode visitOver(final RexOver over) {
                             "as an argument of a DISTINCT aggregate",
                             EQUALITY);
                 }
+            } else if (over.isA(ELEMENT_COMPARING_OPERATORS)) {
+                for (final RexNode operand : over.getOperands()) {
+                    checkType(
+                            "An expression",
+                            operand.getType(),
+                            "as an argument of " + over.getOperator().getName(),
+                            EQUALITY);
+                }
             }
             for (final RexNode key : over.getWindow().partitionKeys) {
                 checkType(
@@ -203,20 +272,33 @@ public RexNode visitOver(final RexOver over) {
 
         @Override
         public RexNode visitSubQuery(final RexSubQuery subQuery) {
-            if (subQuery.isA(SqlKind.IN) || subQuery.isA(SqlKind.NOT_IN)) {
+            if (subQuery.isA(EQUALITY_OPERATORS)) {
                 checkComparison(subQuery.getOperands(), subQuery.getOperator().getName(), EQUALITY);
             } else if (subQuery.getOperator() instanceof SqlQuantifyOperator) {
                 final SqlKind comparisonKind =
                         ((SqlQuantifyOperator) subQuery.getOperator()).comparisonKind;
                 checkComparison(
                         subQuery.getOperands(),
                         subQuery.getOperator().getName(),
-                        SqlKind.ORDER_COMPARISON.contains(comparisonKind) ? ORDER : EQUALITY);
+                        ORDER_OPERATORS.contains(comparisonKind) ? ORDER : EQUALITY);
             }
             new PlanChecker().go(subQuery.rel);
             return super.visitSubQuery(subQuery);
         }
 
+        private static void checkElements(final RexCall call) {
+            final String usage = "in " + call.getOperator().getName();
+            for (final RexNode operand : call.getOperands()) {
+                final RelDataType type = operand.getType();
+                // an index into an array compares no elements
+                if (type.getSqlTypeName() == SqlTypeName.ARRAY && !call.isA(SqlKind.ITEM)) {
+                    checkType("An array element", type.getComponentType(), usage, EQUALITY);
+                } else if (type.getSqlTypeName() == SqlTypeName.MAP) {
+                    checkType("A map key", type.getKeyType(), usage, EQUALITY);
+                }
+            }
+        }
+
         private static void checkComparison(
                 final List<RexNode> operands, final String operatorName, final String requirement) {
             // a comparison with NULL never compares two values
@@ -264,7 +346,7 @@ private static void checkType(
             final String usage,
             final String requirement) {
         final LogicalType logicalType = FlinkTypeFactory.toLogicalType(type);
-        if (!LogicalTypeChecks.isComparableKeyType(logicalType)) {
+        if (!LogicalTypeChecks.isComparableType(logicalType)) {
             throw new ValidationException(
                     String.format(
                             "%s of type %s cannot be used %s, because the type has no %s. "

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala

@@ -76,8 +76,8 @@ abstract class CommonSubGraphBasedOptimizer extends Optimizer {
    *   a list of RelNode represents an optimized RelNode DAG.
    */
   override def optimize(roots: Seq[RelNode]): Seq[RelNode] = {
-    // validate before the optimizer rewrites the keys that the query asks for
-    KeyTypeValidator.validate(toJava(roots))
+    // validate before the optimizer rewrites the comparisons that the query asks for
+    ComparableTypeValidator.validate(toJava(roots))
 
     // resolve hints before optimizing
     val queryHintsResolver = new QueryHintsResolver()

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java

@@ -682,16 +682,8 @@ private static List<TestSetSpec> primitiveToVariantCasts() {
                                 BYTES()),
                 // a constructed type casts element by element when each child casts to VARIANT
                 TestSetSpec.forExpression("Cast a constructed type to VARIANT elements and back")
-                        .onFieldsWithData(
-                                new Integer[] {1, null},
-                                Row.of(7),
-                                map(entry("a", 1)),
-                                map(entry(1, "a")))
-                        .andDataTypes(
-                                ARRAY(INT()),
-                                ROW(FIELD("a", INT())),
-                                MAP(STRING(), INT()),
-                                MAP(INT(), STRING()))
+                        .onFieldsWithData(new Integer[] {1, null}, Row.of(7), map(entry("a", 1)))
+                        .andDataTypes(ARRAY(INT()), ROW(FIELD("a", INT())), MAP(STRING(), INT()))
                         .testResult(
                                 $("f0").cast(ARRAY(VARIANT())).cast(ARRAY(INT())),
                                 "CAST(CAST(f0 AS ARRAY<VARIANT>) AS ARRAY<INT>)",
@@ -707,19 +699,7 @@ private static List<TestSetSpec> primitiveToVariantCasts() {
                                 $("f2").cast(MAP(STRING(), VARIANT())).cast(MAP(STRING(), INT())),
                                 "CAST(CAST(f2 AS MAP<STRING, VARIANT>) AS MAP<STRING, INT>)",
                                 map(entry("a", 1)),
-                                MAP(STRING(), INT()))
-                        // a VARIANT key matches by its bytes, so a key cast from INT does not
-                        // match the same number cast from BIGINT
-                        .testResult(
-                                $("f3").cast(MAP(VARIANT(), STRING())).at(lit(1).cast(VARIANT())),
-                                "CAST(f3 AS MAP<VARIANT, STRING>)[CAST(1 AS VARIANT)]",
-                                "a",
-                                STRING())
-                        .testResult(
-                                $("f3").cast(MAP(VARIANT(), STRING())).at(lit(1L).cast(VARIANT())),
-                                "CAST(f3 AS MAP<VARIANT, STRING>)[CAST(CAST(1 AS BIGINT) AS VARIANT)]",
-                                null,
-                                STRING()));
+                                MAP(STRING(), INT())));
     }
 
     private static List<TestSetSpec> constructedToVariantCasts() {