From 25e6104ada5ef3672dec0f090dd242eecd45b1df Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 26 Oct 2026 10:13:01 +0100 Subject: [PATCH] [FLINK-40936][table] Reject VARIANT elements and keys in functions and operators that compare them Flink compares VARIANT values by their binary encoding, but one value has many encodings. So functions and operators that compare array elements, map keys or collected values treated equal VARIANT values as different and silently returned wrong results. ComparableTypeValidator now checks the array elements and map keys of ARRAY_CONTAINS, ARRAY_DISTINCT, ARRAY_POSITION, ARRAY_REMOVE, ARRAY_UNION, ARRAY_EXCEPT, ARRAY_INTERSECT, MAP_CONTAINS_KEY and MAP_UNION, the map key of an item access m[k], and the argument of COLLECT, also in an OVER window. An index into an array and a field access on a VARIANT compare nothing and stay allowed. The validator names what it checks in sets: the functions, the operators that compare elements, and the comparison operators that need equality or order. The comparison operator sets hold the same kinds that the validator matched before. The check lives in the validator and not in the input type strategy of each function. That would need a new argument strategy and a flag in three collection strategies, for a check that a future VARIANT equality may remove. LogicalTypeChecks.areComparable rejects every type that is not comparable, see LogicalTypeChecks.isComparableType. This also covers a structured type with a VARIANT attribute. GREATEST, LEAST, MAP_FROM_ENTRIES and Table API comparisons reject VARIANT through it during type inference. Ten functions build their equality check again during code generation: ARRAY_CONTAINS, ARRAY_DISTINCT, ARRAY_POSITION, ARRAY_REMOVE, ARRAY_UNION, ARRAY_EXCEPT, ARRAY_INTERSECT, MAP_CONTAINS_KEY, MAP_UNION and MAP_FROM_ENTRIES. That check goes through LogicalTypeChecks.areComparable, not through the validator, which only runs when a query is planned. So a job restored from a compiled plan that calls one of them on VARIANT elements or keys now fails when it is executed, before the job is submitted. Such a plan returned wrong results before. A user SpecializedFunction that builds an = evaluator on VARIANT values now fails too, in new and in restored plans. The VARIANT section of the data types page gives a few examples of where a query compares values, instead of a full list that would soon be out of date. KeyTypeValidator is now ComparableTypeValidator. The planner also uses key for upsert, unique and primary keys, and the validator rejects every comparison of a type that is not comparable, for equality or for order. LogicalTypeChecks.isComparableKeyType is now isComparableType, and the Javadocs say which comparisons need equality and which need order. --- .../docs/sql/reference/data-types.md | 12 +- docs/content/docs/sql/reference/data-types.md | 12 +- .../logical/utils/LogicalTypeChecks.java | 20 ++- .../ComparableInputTypeStrategyTest.java | 7 ++ .../logical/utils/LogicalTypeChecksTest.java | 8 +- ...ator.java => ComparableTypeValidator.java} | 114 +++++++++++++++--- .../CommonSubGraphBasedOptimizer.scala | 4 +- .../planner/functions/CastFunctionITCase.java | 26 +--- .../planner/functions/MapFunctionITCase.java | 9 +- ....java => ComparableTypeValidatorTest.java} | 33 ++++- 10 files changed, 165 insertions(+), 80 deletions(-) rename flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/{KeyTypeValidator.java => ComparableTypeValidator.java} (66%) rename flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/{KeyTypeValidatorTest.java => ComparableTypeValidatorTest.java} (91%) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 2ed6bbe298398e..600dec9ef102d1 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/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 diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index a13e4f2b990e35..f3235b83f5b7ae 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/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 diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecks.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecks.java index af66e5edc07318..57e876288a1212 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecks.java +++ b/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. + * + *

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. * - *

A VARIANT value has many valid binary encodings, so two equal values can differ in their - * bytes. + *

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); } diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ComparableInputTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ComparableInputTypeStrategyTest.java index 3c4b9696f12949..7baa9da260c29d 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ComparableInputTypeStrategyTest.java +++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/ComparableInputTypeStrategyTest.java @@ -313,6 +313,13 @@ protected Stream 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) diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecksTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecksTest.java index ca9775be893a38..df5248ba42e0d8 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecksTest.java +++ b/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 comparableKeyTypes() { + private static Stream comparableTypes() { return Stream.of( Arguments.of(INT(), true), Arguments.of(ROW(FIELD("f0", INT())), true), diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidator.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/ComparableTypeValidator.java similarity index 66% rename from flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidator.java rename to flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/ComparableTypeValidator.java index bf3cfb577479b2..ce556b6ac3cfbe 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidator.java +++ b/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)}. * - *

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. + *

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 >=}. + * + *

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 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 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 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 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 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,7 +272,7 @@ 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 = @@ -211,12 +280,25 @@ public RexNode visitSubQuery(final RexSubQuery subQuery) { 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 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. " diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala index d0eb2603deec1e..330d5a6d6ab2a6 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/CommonSubGraphBasedOptimizer.scala +++ b/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() diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java index 6163ee022f3feb..321bc19f2d7dd3 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CastFunctionITCase.java @@ -682,16 +682,8 @@ private static List 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) AS ARRAY)", @@ -707,19 +699,7 @@ private static List primitiveToVariantCasts() { $("f2").cast(MAP(STRING(), VARIANT())).cast(MAP(STRING(), INT())), "CAST(CAST(f2 AS MAP) AS MAP)", 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)[CAST(1 AS VARIANT)]", - "a", - STRING()) - .testResult( - $("f3").cast(MAP(VARIANT(), STRING())).at(lit(1L).cast(VARIANT())), - "CAST(f3 AS MAP)[CAST(CAST(1 AS BIGINT) AS VARIANT)]", - null, - STRING())); + MAP(STRING(), INT()))); } private static List constructedToVariantCasts() { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java index 1b02ffc587b50a..92f6a6cb2c3d5c 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java @@ -660,14 +660,7 @@ private Stream mapContainsKeyTestCases() { $("f3").mapContainsKey(array(1, 2)), "MAP_CONTAINS_KEY(f3, ARRAY[1, 2])", true, - DataTypes.BOOLEAN()) - // there is no VARIANT literal, so the key is built with PARSE_JSON - .testResult( - map(call("PARSE_JSON", "1"), lit(1)) - .mapContainsKey(call("PARSE_JSON", "1")), - "MAP_CONTAINS_KEY(MAP[PARSE_JSON('1'), 1], PARSE_JSON('1'))", - true, - DataTypes.BOOLEAN().notNull()), + DataTypes.BOOLEAN()), TestSetSpec.forFunction( BuiltInFunctionDefinitions.MAP_CONTAINS_KEY, "Documented examples") .onFieldsWithData(1) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidatorTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/ComparableTypeValidatorTest.java similarity index 91% rename from flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidatorTest.java rename to flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/ComparableTypeValidatorTest.java index 0703af6f2f6bd7..ba36c217c50efa 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidatorTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/ComparableTypeValidatorTest.java @@ -48,8 +48,8 @@ import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; -/** Tests for {@link KeyTypeValidator}. */ -class KeyTypeValidatorTest { +/** Tests for {@link ComparableTypeValidator}. */ +class ComparableTypeValidatorTest { private static final String GROUPING_KEY = "Column 'v' of type VARIANT cannot be used as a grouping key"; @@ -62,6 +62,11 @@ class KeyTypeValidatorTest { private static final String EQUALS = COMPARISON + " '='"; + private static final String ARRAY_ELEMENT = + "An array element of type VARIANT cannot be used in"; + + private static final String MAP_KEY = "A map key of type VARIANT cannot be used in"; + private static Stream rejectedQueries() { final Stream bothModes = inBothModes( @@ -172,6 +177,27 @@ private static Stream rejectedQueries() { Arguments.of( "SELECT CASE v WHEN PARSE_JSON('1') THEN 1 END FROM t", EQUALS), Arguments.of("SELECT NULLIF(v, PARSE_JSON('1')) FROM t", EQUALS), + Arguments.of( + "SELECT ARRAY_CONTAINS(a, v) FROM t", + ARRAY_ELEMENT + + " ARRAY_CONTAINS, because the type has no equality"), + Arguments.of("SELECT ARRAY_DISTINCT(a) FROM t", ARRAY_ELEMENT), + Arguments.of("SELECT ARRAY_POSITION(a, v) FROM t", ARRAY_ELEMENT), + Arguments.of("SELECT ARRAY_REMOVE(a, v) FROM t", ARRAY_ELEMENT), + Arguments.of("SELECT ARRAY_UNION(a, a) FROM t", ARRAY_ELEMENT), + Arguments.of("SELECT ARRAY_EXCEPT(a, a) FROM t", ARRAY_ELEMENT), + Arguments.of("SELECT ARRAY_INTERSECT(a, a) FROM t", ARRAY_ELEMENT), + Arguments.of("SELECT MAP_CONTAINS_KEY(m, v) FROM t", MAP_KEY), + Arguments.of("SELECT MAP_UNION(m, m) FROM t", MAP_KEY), + Arguments.of("SELECT m[v] FROM t", MAP_KEY + " ITEM"), + Arguments.of( + "SELECT COLLECT(v) FROM t", + "Column 'v' of type VARIANT cannot be used as an argument of " + + "COLLECT"), + Arguments.of( + "SELECT COLLECT(v) OVER (PARTITION BY id ORDER BY ts) FROM t", + "An expression of type VARIANT cannot be used as an argument of " + + "COLLECT"), Arguments.of( "SELECT * FROM t MATCH_RECOGNIZE (PARTITION BY v ORDER BY ts " + "MEASURES A.id AS aid PATTERN (A) DEFINE A AS A.id > 0)", @@ -209,6 +235,8 @@ private static Stream allowedQueries() { Arguments.of("SELECT v, ts FROM t ORDER BY ts"), Arguments.of("SELECT DISTINCT id, CAST(v AS STRING) FROM t"), Arguments.of("SELECT v FROM t WHERE v IS NOT NULL AND r IS NULL"), + Arguments.of("SELECT MAP_CONTAINS_KEY(MAP['k', v], 'k') FROM t"), + Arguments.of("SELECT a[1], v['k'] FROM t"), Arguments.of( "SELECT v FROM t WHERE v IS DISTINCT FROM NULL " + "OR v IS NOT DISTINCT FROM NULL")); @@ -373,6 +401,7 @@ private static TableEnvironment createEnvironment(final RuntimeExecutionMode mod + "v VARIANT, " + "r ROW, " + "a ARRAY, " + + "m MAP, " + "ts TIMESTAMP(3), " + "WATERMARK FOR ts AS ts" + ") WITH ('connector' = 'values', 'bounded' = '%s')",