This is an automated email from the ASF dual-hosted git repository.
raminqaf pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 417b615d80c [FLINK-40936][table] Reject VARIANT elements and keys in
functions and operators that compare them (#29610)
417b615d80c is described below
commit 417b615d80c7eb0e9f87e8716a33b3bcbda30ade
Author: Ramin Gharib <[email protected]>
AuthorDate: Mon Oct 26 11:24:09 2026 +0100
[FLINK-40936][table] Reject VARIANT elements and keys in functions and
operators that compare them (#29610)
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 returned wrong results for equal VARIANT values.
ComparableTypeValidator now rejects VARIANT elements and keys in the array
and map functions that compare them, such as ARRAY_CONTAINS and
MAP_CONTAINS_KEY. It also rejects a VARIANT key in a map lookup m[k] and a
VARIANT argument of COLLECT. An array index and a field access on a VARIANT
stay allowed. The check lives in the validator, not in each function's input
type strategy, since a future VARIANT equality may remove it.
LogicalTypeChecks.areComparable now rejects every type that is not
comparable, so GREATEST, LEAST and Table API comparisons reject VARIANT too.
The array and map functions check it again during code generation. So a
restored compiled plan that calls them on VARIANT now fails before the job is
submitted, instead of returning wrong results. A user SpecializedFunction that
builds an = evaluator on VARIANT fails too.
KeyTypeValidator is now ComparableTypeValidator, and isComparableKeyType is
now isComparableType, since both cover every comparison, not only keys.
(cherry picked from commit 5c9455b113ecdf7fb1d67e3d37e98a2dea229660)
---
docs/content.zh/docs/sql/reference/data-types.md | 12 +--
docs/content/docs/sql/reference/data-types.md | 12 +--
.../types/logical/utils/LogicalTypeChecks.java | 20 +++-
.../inference/ComparableInputTypeStrategyTest.java | 7 ++
.../types/logical/utils/LogicalTypeChecksTest.java | 8 +-
...Validator.java => ComparableTypeValidator.java} | 114 ++++++++++++++++++---
.../optimize/CommonSubGraphBasedOptimizer.scala | 4 +-
.../planner/functions/CastFunctionITCase.java | 26 +----
.../table/planner/functions/MapFunctionITCase.java | 9 +-
...rTest.java => ComparableTypeValidatorTest.java} | 33 +++++-
10 files changed, 165 insertions(+), 80 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index 2ed6bbe2983..600dec9ef10 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 a13e4f2b990..f3235b83f5b 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 af66e5edc07..57e876288a1 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 final class LogicalTypeChecks {
}
}
+ /**
+ * 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 final class LogicalTypeChecks {
}
/**
- * 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 @@ public final class LogicalTypeChecks {
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 3c4b9696f12..7baa9da260c 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 @@ class ComparableInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
.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 ca9775be893..df5248ba42e 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 @@ class LogicalTypeChecksTest {
}
@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),
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 bf3cfb57747..ce556b6ac3c 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.rex.RexUtil;
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 @@ public final class KeyTypeValidator {
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 final class KeyTypeValidator {
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 final class KeyTypeValidator {
"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 final class KeyTypeValidator {
@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 final class KeyTypeValidator {
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 @@ public final class KeyTypeValidator {
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 d0eb2603dee..330d5a6d6ab 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 6163ee022f3..321bc19f2d7 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 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
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 @@ public class CastFunctionITCase extends
BuiltInFunctionTestBase {
$("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() {
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 1b02ffc587b..92f6a6cb2c3 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 @@ public class MapFunctionITCase extends
BuiltInFunctionTestBase {
$("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 0703af6f2f6..ba36c217c50 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.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXE
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<Arguments> rejectedQueries() {
final Stream<Arguments> bothModes =
inBothModes(
@@ -172,6 +177,27 @@ class KeyTypeValidatorTest {
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 @@ class KeyTypeValidatorTest {
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 @@ class KeyTypeValidatorTest {
+ "v VARIANT, "
+ "r ROW<x VARIANT>, "
+ "a ARRAY<VARIANT>, "
+ + "m MAP<VARIANT, INT>, "
+ "ts TIMESTAMP(3), "
+ "WATERMARK FOR ts AS ts"
+ ") WITH ('connector' = 'values',
'bounded' = '%s')",
