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')",