This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 5d17b29d312 [FLINK-40910][table] Reject `VARIANT` where a query needs
equality or ordering
5d17b29d312 is described below
commit 5d17b29d3122476690d657f390e6df56bee57153
Author: Ramin Gharib <[email protected]>
AuthorDate: Thu Oct 8 15:02:14 2026 +0200
[FLINK-40910][table] Reject `VARIANT` where a query needs equality or
ordering
---
docs/content.zh/docs/sql/reference/data-types.md | 30 +-
docs/content/docs/sql/reference/data-types.md | 30 +-
.../types/logical/utils/LogicalTypeChecks.java | 11 +
.../types/logical/utils/LogicalTypeChecksTest.java | 33 ++
.../planner/plan/optimize/KeyTypeValidator.java | 275 ++++++++++++++
.../optimize/CommonSubGraphBasedOptimizer.scala | 3 +
.../nodes/exec/stream/VariantSemanticTest.java | 149 ++++++--
.../plan/optimize/KeyTypeValidatorTest.java | 399 +++++++++++++++++++++
8 files changed, 899 insertions(+), 31 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index e5344621c3f..04accbeb996 100644
--- a/docs/content.zh/docs/sql/reference/data-types.md
+++ b/docs/content.zh/docs/sql/reference/data-types.md
@@ -1705,8 +1705,8 @@ A scalar value can also be cast to a `VARIANT` with
`CAST` or `TRY_CAST`. Only a
- A constructed type casts element by element when its element, field, or
value type casts to
`VARIANT`, for example `ARRAY<INT>` to `ARRAY<VARIANT>`. A `NULL` element
stays a SQL `NULL`. A
whole `ARRAY`, `MAP`, or `ROW` does not cast into a single `VARIANT` yet.
-- Two `VARIANT` values are equal only when their binary encodings match. So as
a `MAP` key or a
- `MULTISET` element, a `1` cast from `INT` does not match a `1` cast from
`BIGINT` or parsed by
+- As a `MAP` key or a `MULTISET` element, two `VARIANT` values match only when
their binary
+ encodings match. So a `1` cast from `INT` does not match a `1` cast from
`BIGINT` or parsed by
`PARSE_JSON('1')`.
```sql
@@ -1720,6 +1720,32 @@ CAST(ARRAY[1, NULL] AS ARRAY<VARIANT>) -- [1,
NULL], each element a VARIAN
CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not
supported yet
```
+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
+
+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
+with `IS NULL`, cast, and passed to aggregate functions such as `COUNT`,
`FIRST_VALUE`, and
+`LAST_VALUE`.
+
+```sql
+SELECT v['a'], COUNT(*) FROM t GROUP BY v['a'] --
fails
+SELECT CAST(v['a'] AS INT), COUNT(*) FROM t GROUP BY CAST(v['a'] AS INT) --
one group per number
+SELECT * FROM t WHERE v['a'] = PARSE_JSON('1') --
fails
+SELECT * FROM t WHERE CAST(v['a'] AS INT) = 1 --
compares the number
+```
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
diff --git a/docs/content/docs/sql/reference/data-types.md
b/docs/content/docs/sql/reference/data-types.md
index c80cb694fc2..66dba8e499d 100644
--- a/docs/content/docs/sql/reference/data-types.md
+++ b/docs/content/docs/sql/reference/data-types.md
@@ -1715,8 +1715,8 @@ A scalar value can also be cast to a `VARIANT` with
`CAST` or `TRY_CAST`. Only a
- A constructed type casts element by element when its element, field, or
value type casts to
`VARIANT`, for example `ARRAY<INT>` to `ARRAY<VARIANT>`. A `NULL` element
stays a SQL `NULL`. A
whole constructed value can also become a single `VARIANT`, see below.
-- Two `VARIANT` values are equal only when their binary encodings match. So as
a `MAP` key or a
- `MULTISET` element, a `1` cast from `INT` does not match a `1` cast from
`BIGINT` or parsed by
+- As a `MAP` key or a `MULTISET` element, two `VARIANT` values match only when
their binary
+ encodings match. So a `1` cast from `INT` does not match a `1` cast from
`BIGINT` or parsed by
`PARSE_JSON('1')`.
```sql
@@ -1763,6 +1763,32 @@ CAST(MAP[1, 'a'] AS VARIANT) -- fails at
validation, a MAP key m
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
+
+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
+with `IS NULL`, cast, and passed to aggregate functions such as `COUNT`,
`FIRST_VALUE`, and
+`LAST_VALUE`.
+
+```sql
+SELECT v['a'], COUNT(*) FROM t GROUP BY v['a'] --
fails
+SELECT CAST(v['a'] AS INT), COUNT(*) FROM t GROUP BY CAST(v['a'] AS INT) --
one group per number
+SELECT * FROM t WHERE v['a'] = PARSE_JSON('1') --
fails
+SELECT * FROM t WHERE CAST(v['a'] AS INT) = 1 --
compares the number
+```
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
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 9c1a791472a..af66e5edc07 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
@@ -257,6 +257,17 @@ public final class LogicalTypeChecks {
firstType.copy(true), secondType.copy(true),
requiredComparison);
}
+ /**
+ * 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.
+ *
+ * <p>A VARIANT value has many valid binary encodings, so two equal values
can differ in their
+ * bytes.
+ */
+ public static boolean isComparableKeyType(LogicalType logicalType) {
+ return !hasNested(logicalType, t -> t.is(LogicalTypeRoot.VARIANT));
+ }
+
private static boolean areComparableWithNormalizedNullability(
LogicalType firstType,
LogicalType secondType,
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 d0ede0f0fd9..ca9775be893 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
@@ -30,14 +30,21 @@ import org.apache.flink.table.types.logical.StructuredType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.Arrays;
import java.util.List;
+import java.util.stream.Stream;
+import static org.apache.flink.table.api.DataTypes.ARRAY;
import static org.apache.flink.table.api.DataTypes.FIELD;
import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.MAP;
import static org.apache.flink.table.api.DataTypes.ROW;
import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.DataTypes.VARIANT;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link LogicalTypeChecks}. */
@@ -133,4 +140,30 @@ class LogicalTypeChecksTest {
DataType dataType = ROW(FIELD("f0", INT()), FIELD("f1", STRING()));
assertThat(LogicalTypeChecks.getFieldCount(dataType.getLogicalType())).isEqualTo(2);
}
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("comparableKeyTypes")
+ void testIsComparableKeyType(DataType dataType, boolean expected) {
+
assertThat(LogicalTypeChecks.isComparableKeyType(dataType.getLogicalType()))
+ .isEqualTo(expected);
+ }
+
+ private static Stream<Arguments> comparableKeyTypes() {
+ return Stream.of(
+ Arguments.of(INT(), true),
+ Arguments.of(ROW(FIELD("f0", INT())), true),
+ Arguments.of(VARIANT(), false),
+ Arguments.of(ROW(FIELD("f0", VARIANT())), false),
+ Arguments.of(ARRAY(VARIANT()), false),
+ Arguments.of(MAP(STRING(), VARIANT()), false),
+ Arguments.of(
+ TypeConversions.fromLogicalToDataType(
+
StructuredType.newBuilder(ObjectIdentifier.of("cat", "db", "type"))
+ .attributes(
+ List.of(
+ new
StructuredType.StructuredAttribute(
+ "v",
VARIANT().getLogicalType())))
+ .build()),
+ false));
+ }
}
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/KeyTypeValidator.java
new file mode 100644
index 00000000000..bf3cfb57747
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidator.java
@@ -0,0 +1,275 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.optimize;
+
+import org.apache.flink.table.api.ValidationException;
+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.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
+
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelVisitor;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.core.Match;
+import org.apache.calcite.rel.core.SetOp;
+import org.apache.calcite.rel.core.Sort;
+import org.apache.calcite.rel.core.Union;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexFieldCollation;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexOver;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.rex.RexSubQuery;
+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.util.ImmutableBitSet;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/**
+ * 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)}.
+ *
+ * <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.
+ */
+public final class KeyTypeValidator {
+
+ private static final String EQUALITY = "equality";
+ private static final String ORDER = "order";
+
+ private KeyTypeValidator() {}
+
+ /**
+ * Throws a {@link ValidationException} if a plan groups, joins,
partitions, sorts or compares
+ * values of a type that is not a comparable key type.
+ */
+ public static void validate(final List<RelNode> roots) {
+ roots.forEach(root -> new PlanChecker().go(root));
+ }
+
+ private static final class PlanChecker extends RelVisitor {
+
+ private final RexShuttle expressionChecker = new ExpressionChecker();
+
+ @Override
+ public void visit(final RelNode node, final int ordinal, @Nullable
final RelNode parent) {
+ checkKeys(node);
+ node.accept(expressionChecker);
+ super.visit(node, ordinal, parent);
+ }
+
+ private void checkKeys(final RelNode node) {
+ if (node instanceof Aggregate) {
+ final Aggregate aggregate = (Aggregate) node;
+ final RelDataType inputType =
aggregate.getInput().getRowType();
+ checkFields(inputType, aggregate.getGroupSet(), "as a grouping
key", EQUALITY);
+ for (final AggregateCall call : aggregate.getAggCallList()) {
+ if (call.isDistinct()) {
+ checkFields(
+ inputType,
+ ImmutableBitSet.of(call.getArgList()),
+ "as an argument of a DISTINCT aggregate",
+ EQUALITY);
+ }
+ }
+ } else if (node instanceof TableAggregate) {
+ final TableAggregate aggregate = (TableAggregate) node;
+ checkFields(
+ aggregate.getInput().getRowType(),
+ aggregate.getGroupSet(),
+ "as a grouping key",
+ EQUALITY);
+ } else if (node instanceof SetOp) {
+ final SetOp setOp = (SetOp) node;
+ if (!(setOp instanceof Union && setOp.all)) {
+ final RelDataType rowType = setOp.getRowType();
+ checkFields(
+ rowType,
+ ImmutableBitSet.range(rowType.getFieldCount()),
+ "in " + setOp.kind.sql + (setOp.all ? " ALL" : ""),
+ EQUALITY);
+ }
+ } else if (node instanceof Sort) {
+ final Sort sort = (Sort) node;
+ checkCollation(
+ sort.getInput().getRowType(), sort.getCollation(), "as
an ORDER BY key");
+ } else if (node instanceof Match) {
+ // Match and WatermarkAssigner do not expose their expressions
to a RexShuttle
+ final Match match = (Match) node;
+ final RelDataType inputType = match.getInput().getRowType();
+ checkFields(
+ inputType,
+ match.getPartitionKeys(),
+ "as a PARTITION BY key of MATCH_RECOGNIZE",
+ EQUALITY);
+ checkCollation(
+ inputType, match.getOrderKeys(), "as an ORDER BY key
of MATCH_RECOGNIZE");
+ match.getPatternDefinitions().values().forEach(e ->
e.accept(expressionChecker));
+ match.getMeasures().values().forEach(e ->
e.accept(expressionChecker));
+ } else if (node instanceof WatermarkAssigner) {
+ ((WatermarkAssigner)
node).watermarkExpr().accept(expressionChecker);
+ }
+ }
+
+ private void checkCollation(
+ final RelDataType rowType, final RelCollation collation, final
String usage) {
+ for (final RelFieldCollation fieldCollation :
collation.getFieldCollations()) {
+ checkField(rowType, fieldCollation.getFieldIndex(), usage,
ORDER);
+ }
+ }
+ }
+
+ private static final class ExpressionChecker extends RexShuttle {
+
+ @Override
+ public RexNode visitCall(final RexCall call) {
+ if (call instanceof RexTableArgCall) {
+ final RexTableArgCall tableArg = (RexTableArgCall) call;
+ checkFields(
+ tableArg.getType(),
+ ImmutableBitSet.of(tableArg.getPartitionKeys()),
+ "as a PARTITION BY key of a table argument",
+ EQUALITY);
+ checkFields(
+ tableArg.getType(),
+ 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);
+ }
+ return super.visitCall(call);
+ }
+
+ @Override
+ public RexNode visitOver(final RexOver over) {
+ if (over.isDistinct()) {
+ for (final RexNode operand : over.getOperands()) {
+ checkType(
+ "An expression",
+ operand.getType(),
+ "as an argument of a DISTINCT aggregate",
+ EQUALITY);
+ }
+ }
+ for (final RexNode key : over.getWindow().partitionKeys) {
+ checkType(
+ "An expression",
+ key.getType(),
+ "as a PARTITION BY key of an OVER window",
+ EQUALITY);
+ }
+ for (final RexFieldCollation key : over.getWindow().orderKeys) {
+ checkType(
+ "An expression",
+ key.getKey().getType(),
+ "as an ORDER BY key of an OVER window",
+ ORDER);
+ }
+ return super.visitOver(over);
+ }
+
+ @Override
+ public RexNode visitSubQuery(final RexSubQuery subQuery) {
+ if (subQuery.isA(SqlKind.IN) || subQuery.isA(SqlKind.NOT_IN)) {
+ 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);
+ }
+ new PlanChecker().go(subQuery.rel);
+ return super.visitSubQuery(subQuery);
+ }
+
+ private static void checkComparison(
+ final List<RexNode> operands, final String operatorName, final
String requirement) {
+ // a comparison with NULL never compares two values
+ if (operands.stream().anyMatch(operand ->
RexUtil.isNullLiteral(operand, true))) {
+ return;
+ }
+ for (final RexNode operand : operands) {
+ checkType(
+ "An expression",
+ operand.getType(),
+ "in a comparison with '" + operatorName + "'",
+ requirement);
+ }
+ }
+ }
+
+ private static void checkFields(
+ final RelDataType rowType,
+ final ImmutableBitSet fields,
+ final String usage,
+ final String requirement) {
+ for (final int field : fields) {
+ checkField(rowType, field, usage, requirement);
+ }
+ }
+
+ private static void checkField(
+ final RelDataType rowType,
+ final int index,
+ final String usage,
+ final String requirement) {
+ final RelDataTypeField field = rowType.getFieldList().get(index);
+ final String name = field.getName();
+ // names such as EXPR$0 or $f0 come from the planner and mean nothing
to the user
+ final String subject =
+ name.startsWith("$") || SqlUtil.isGeneratedAlias(name)
+ ? "An expression"
+ : String.format("Column '%s'", name);
+ checkType(subject, field.getType(), usage, requirement);
+ }
+
+ private static void checkType(
+ final String subject,
+ final RelDataType type,
+ final String usage,
+ final String requirement) {
+ final LogicalType logicalType = FlinkTypeFactory.toLogicalType(type);
+ if (!LogicalTypeChecks.isComparableKeyType(logicalType)) {
+ throw new ValidationException(
+ String.format(
+ "%s of type %s cannot be used %s, because the type
has no %s. "
+ + "Cast the value to a comparable type
first.",
+ subject, logicalType.copy(true).asSummaryString(),
usage, requirement));
+ }
+ }
+}
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 92c7f316ae0..d0eb2603dee 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,6 +76,9 @@ 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))
+
// resolve hints before optimizing
val queryHintsResolver = new QueryHintsResolver()
val resolvedHintRoots = queryHintsResolver.resolve(toJava(roots))
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
index 86e5b660210..c769e326f25 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/VariantSemanticTest.java
@@ -145,16 +145,16 @@ public class VariantSemanticTest extends SemanticTestBase
{
.build())
.setupTableSink(
SinkTestStep.newBuilder("sink_t")
- .addSchema("fv VARIANT", "lv VARIANT", "c
BIGINT", "dc BIGINT")
+ .addSchema("fv VARIANT", "lv VARIANT", "c
BIGINT")
.consumedValues(
- Row.of(V1, V1, 1, 1),
- Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V1, 1, 1),
- Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 2, 2),
- Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V2, 2, 2),
- Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 3, 2))
+ Row.of(V1, V1, 1),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V1, 1),
+ Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 2),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V2, 2),
+ Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 3))
.build())
.runSql(
- "INSERT INTO sink_t SELECT FIRST_VALUE(v),
LAST_VALUE(v), COUNT(v), COUNT(DISTINCT v) FROM t")
+ "INSERT INTO sink_t SELECT FIRST_VALUE(v),
LAST_VALUE(v), COUNT(v) FROM t")
.build();
static final TableTestProgram BUILTIN_AGG_WITH_RETRACTION =
@@ -172,18 +172,18 @@ public class VariantSemanticTest extends SemanticTestBase
{
.build())
.setupTableSink(
SinkTestStep.newBuilder("sink_t")
- .addSchema("fv VARIANT", "lv VARIANT", "c
BIGINT", "dc BIGINT")
+ .addSchema("fv VARIANT", "lv VARIANT", "c
BIGINT")
.consumedValues(
- Row.of(V1, V1, 1, 1),
- Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V1, 1, 1),
- Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 2, 2),
- Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V2, 2, 2),
- Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 3, 2),
- Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V2, 3, 2),
- Row.ofKind(RowKind.UPDATE_AFTER,
V2, V2, 2, 1))
+ Row.of(V1, V1, 1),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V1, 1),
+ Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 2),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V2, 2),
+ Row.ofKind(RowKind.UPDATE_AFTER,
V1, V2, 3),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
V1, V2, 3),
+ Row.ofKind(RowKind.UPDATE_AFTER,
V2, V2, 2))
.build())
.runSql(
- "INSERT INTO sink_t SELECT FIRST_VALUE(v),
LAST_VALUE(v), COUNT(v), COUNT(DISTINCT v) FROM t")
+ "INSERT INTO sink_t SELECT FIRST_VALUE(v),
LAST_VALUE(v), COUNT(v) FROM t")
.build();
static final TableTestProgram VARIANT_AS_UDF_ARG =
@@ -277,25 +277,115 @@ public class VariantSemanticTest extends
SemanticTestBase {
.build();
static final TableTestProgram VARIANT_AS_AGG_KEY =
- TableTestProgram.of("variant-as-agg-key", "validates variant as
agg key")
+ TableTestProgram.of(
+ "variant-as-agg-key", "validates that a variant
cannot be an agg key")
.setupTableSource(
SourceTestStep.newBuilder("t")
.addSchema("k VARIANT", "v INTEGER")
- .producedValues(
- Row.of(BUILDER.of(1), 1),
- Row.of(BUILDER.of(2), 2),
- Row.of(BUILDER.of(1), 2))
+ .producedValues(Row.of(BUILDER.of(1), 1))
.build())
+ .runFailingSql(
+ "SELECT k, SUM(v) AS total FROM t GROUP BY k",
+ ValidationException.class,
+ "Column 'k' of type VARIANT cannot be used as a
grouping key, "
+ + "because the type has no equality. "
+ + "Cast the value to a comparable type
first.")
+ .build();
+
+ static final SourceTestStep OBJECTS_WITH_DIFFERENT_KEYS_SOURCE =
+ SourceTestStep.newBuilder("t")
+ .addSchema("s STRING")
+ .producedValues(
+ Row.of("{\"a\":1,\"b\":2}"),
+ Row.of("{\"b\":2,\"a\":1}"),
+ Row.of("{\"a\":1,\"c\":3}"))
+ .build();
+
+ static final TableTestProgram VARIANT_FIELD_CAST_AS_AGG_KEY =
+ TableTestProgram.of(
+ "variant-field-cast-as-agg-key",
+ "validates that casting a variant field before
grouping forms one group")
+ .setupTableSource(OBJECTS_WITH_DIFFERENT_KEYS_SOURCE)
.setupTableSink(
SinkTestStep.newBuilder("sink_t")
- .addSchema("k VARIANT", "total INTEGER")
+ .addSchema("a INT", "cnt BIGINT")
.consumedValues(
- Row.of(1, 1),
- Row.of(2, 2),
- Row.ofKind(RowKind.UPDATE_BEFORE,
1, 1),
- Row.ofKind(RowKind.UPDATE_AFTER,
1, 3))
+ Row.of(1, 1L),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
1, 1L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
1, 2L),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
1, 2L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
1, 3L))
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT a, COUNT(*) FROM "
+ + "(SELECT CAST(PARSE_JSON(s)['a'] AS INT)
AS a FROM t) GROUP BY a")
+ .build();
+
+ static final TableTestProgram VARIANT_FIELD_AS_AGG_KEY =
+ TableTestProgram.of(
+ "variant-field-as-agg-key",
+ "validates that a variant field cannot be an agg
key")
+ .setupTableSource(OBJECTS_WITH_DIFFERENT_KEYS_SOURCE)
+ .runFailingSql(
+ "SELECT a, COUNT(*) FROM (SELECT
PARSE_JSON(s)['a'] AS a FROM t) GROUP BY a",
+ ValidationException.class,
+ "Column 'a' of type VARIANT cannot be used as a
grouping key")
+ .build();
+
+ static final TableTestProgram VARIANT_OBJECT_AS_DISTINCT_KEY =
+ TableTestProgram.of(
+ "variant-object-as-distinct-key",
+ "validates that a variant cannot be a DISTINCT
key")
+ .setupTableSource(OBJECTS_WITH_DIFFERENT_KEYS_SOURCE)
+ .runFailingSql(
+ "SELECT COUNT(DISTINCT PARSE_JSON(s)) FROM t",
+ ValidationException.class,
+ "of type VARIANT cannot be used as an argument of
a DISTINCT aggregate")
+ .runFailingSql(
+ "SELECT DISTINCT PARSE_JSON(s) FROM t",
+ ValidationException.class,
+ "of type VARIANT cannot be used as a grouping key")
+ .build();
+
+ static final TableTestProgram VARIANT_FIELD_EQUALS_VARIANT =
+ TableTestProgram.of(
+ "variant-field-equals-variant",
+ "validates that variants cannot be compared with =
or IS [NOT] DISTINCT FROM")
+ .setupTableSource(OBJECTS_WITH_DIFFERENT_KEYS_SOURCE)
+ .runFailingSql(
+ "SELECT s FROM t WHERE PARSE_JSON(s)['a'] =
PARSE_JSON('1')",
+ ValidationException.class,
+ "An expression of type VARIANT cannot be used in a
comparison with '='")
+ // the SQL converter expands IS [NOT] DISTINCT FROM into =
+ .runFailingSql(
+ "SELECT s FROM t WHERE PARSE_JSON(s)['a'] IS
DISTINCT FROM PARSE_JSON('1')",
+ ValidationException.class,
+ "An expression of type VARIANT cannot be used in a
comparison with '='")
+ .runFailingSql(
+ "SELECT s FROM t WHERE PARSE_JSON(s)['a'] IS NOT
DISTINCT FROM PARSE_JSON('1')",
+ ValidationException.class,
+ "An expression of type VARIANT cannot be used in a
comparison with '='")
+ .build();
+
+ static final TableTestProgram VARIANT_FIELD_AS_JOIN_KEY =
+ TableTestProgram.of(
+ "variant-field-as-join-key",
+ "validates that a variant field cannot be a join
key")
+ .setupTableSource(
+ SourceTestStep.newBuilder("l")
+ .addSchema("s STRING")
+
.producedValues(Row.of("{\"id\":7,\"x\":1}"))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("r")
+ .addSchema("s STRING")
+
.producedValues(Row.of("{\"y\":2,\"id\":7}"))
.build())
- .runSql("INSERT INTO sink_t SELECT k, SUM(v) AS total FROM
t GROUP BY k")
+ .runFailingSql(
+ "SELECT PARSE_JSON(l.s)['x'], PARSE_JSON(r.s)['y']
"
+ + "FROM l JOIN r ON PARSE_JSON(l.s)['id']
= PARSE_JSON(r.s)['id']",
+ ValidationException.class,
+ "An expression of type VARIANT cannot be used in a
comparison with '='")
.build();
public static final SourceTestStep VARIANT_ARRAY_SOURCE =
@@ -512,6 +602,11 @@ public class VariantSemanticTest extends SemanticTestBase {
VARIANT_IN_VIEW,
VARIANT_AS_UDAF_ARG,
VARIANT_AS_AGG_KEY,
+ VARIANT_FIELD_CAST_AS_AGG_KEY,
+ VARIANT_FIELD_AS_AGG_KEY,
+ VARIANT_OBJECT_AS_DISTINCT_KEY,
+ VARIANT_FIELD_EQUALS_VARIANT,
+ VARIANT_FIELD_AS_JOIN_KEY,
VARIANT_ARRAY_ACCESS,
VARIANT_ARRAY_ACCESS_WITH_DIFFERENT_INDEX_TYPES,
VARIANT_OBJECT_ACCESS,
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/KeyTypeValidatorTest.java
new file mode 100644
index 00000000000..0703af6f2f6
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/KeyTypeValidatorTest.java
@@ -0,0 +1,399 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.optimize;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.ArgumentTrait;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.Over;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.ProcessTableFunction;
+import org.apache.flink.types.Row;
+
+import org.junit.jupiter.api.Named;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import static org.apache.flink.api.common.RuntimeExecutionMode.BATCH;
+import static org.apache.flink.api.common.RuntimeExecutionMode.STREAMING;
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.UNBOUNDED_ROW;
+import static
org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link KeyTypeValidator}. */
+class KeyTypeValidatorTest {
+
+ private static final String GROUPING_KEY =
+ "Column 'v' of type VARIANT cannot be used as a grouping key";
+
+ private static final String DISTINCT_ARGUMENT =
+ "of type VARIANT cannot be used as an argument of a DISTINCT
aggregate";
+
+ private static final String COMPARISON =
+ "An expression of type VARIANT cannot be used in a comparison
with";
+
+ private static final String EQUALS = COMPARISON + " '='";
+
+ private static Stream<Arguments> rejectedQueries() {
+ final Stream<Arguments> bothModes =
+ inBothModes(
+ Arguments.of(
+ "SELECT v, COUNT(*) FROM t GROUP BY v",
+ "Column 'v' of type VARIANT cannot be used as
a grouping key, "
+ + "because the type has no equality. "
+ + "Cast the value to a comparable type
first."),
+ Arguments.of(
+ "SELECT COUNT(*) FROM t GROUP BY r",
+ "Column 'r' of type ROW<`x` VARIANT> cannot be
used as a grouping key"),
+ Arguments.of(
+ "SELECT COUNT(*) FROM t GROUP BY a",
+ "Column 'a' of type ARRAY<VARIANT> cannot be
used as a grouping key"),
+ Arguments.of(
+ "SELECT COUNT(*) FROM t GROUP BY v['a']",
+ "An expression of type VARIANT cannot be used
as a grouping key"),
+ Arguments.of(
+ "SELECT COUNT(*) FROM t GROUP BY GROUPING SETS
((v), (id))",
+ GROUPING_KEY),
+ Arguments.of(
+ "SELECT COUNT(*) FROM t GROUP BY ROLLUP (v,
id)", GROUPING_KEY),
+ Arguments.of("SELECT DISTINCT v FROM t", GROUPING_KEY),
+ Arguments.of(
+ "SELECT COUNT(DISTINCT v) FROM t",
+ "Column 'v' " + DISTINCT_ARGUMENT),
+ Arguments.of(
+ "SELECT COUNT(DISTINCT v) OVER (PARTITION BY
id ORDER BY ts) FROM t",
+ "An expression " + DISTINCT_ARGUMENT),
+ Arguments.of(
+ "SELECT window_start, v, COUNT(*) FROM TABLE("
+ + "TUMBLE(TABLE t, DESCRIPTOR(ts),
INTERVAL '1' MINUTE)) "
+ + "GROUP BY window_start, window_end,
v",
+ GROUPING_KEY),
+ Arguments.of(
+ "SELECT window_start, COUNT(*) FROM TABLE("
+ + "SESSION(TABLE t PARTITION BY v,
DESCRIPTOR(ts), INTERVAL '1' MINUTE)) "
+ + "GROUP BY window_start, window_end",
+ "Column 'v' of type VARIANT cannot be used as
a PARTITION BY key"),
+ Arguments.of(
+ "SELECT v FROM t UNION SELECT v FROM u",
+ "Column 'v' of type VARIANT cannot be used in
UNION"),
+ Arguments.of(
+ "SELECT v FROM t INTERSECT SELECT v FROM u",
+ "Column 'v' of type VARIANT cannot be used in
INTERSECT"),
+ Arguments.of(
+ "SELECT v FROM t INTERSECT ALL SELECT v FROM
u",
+ "Column 'v' of type VARIANT cannot be used in
INTERSECT ALL"),
+ Arguments.of(
+ "SELECT v FROM t EXCEPT SELECT v FROM u",
+ "Column 'v' of type VARIANT cannot be used in
EXCEPT"),
+ Arguments.of(
+ "SELECT v FROM t EXCEPT ALL SELECT v FROM u",
+ "Column 'v' of type VARIANT cannot be used in
EXCEPT ALL"),
+ Arguments.of(
+ "SELECT v FROM t ORDER BY v",
+ "Column 'v' of type VARIANT cannot be used as
an ORDER BY key, "
+ + "because the type has no order."),
+ Arguments.of(
+ "SELECT ts, v FROM t ORDER BY ts, v",
+ "Column 'v' of type VARIANT cannot be used as
an ORDER BY key"),
+ Arguments.of(
+ "SELECT COUNT(*) OVER (PARTITION BY v ORDER BY
ts) FROM t",
+ "An expression of type VARIANT cannot be used
as a PARTITION BY key "
+ + "of an OVER window"),
+ Arguments.of(
+ "SELECT * FROM (SELECT id, ROW_NUMBER() OVER "
+ + "(PARTITION BY v ORDER BY ts) AS rn
FROM t) WHERE rn = 1",
+ "An expression of type VARIANT cannot be used
as a PARTITION BY key "
+ + "of an OVER window"),
+ Arguments.of(
+ "SELECT * FROM (SELECT id, ROW_NUMBER() OVER "
+ + "(PARTITION BY id ORDER BY v) AS rn
FROM t) WHERE rn <= 3",
+ "An expression of type VARIANT cannot be used
as an ORDER BY key "
+ + "of an OVER window"),
+ Arguments.of("SELECT t.id FROM t JOIN u ON t.v = u.v",
EQUALS),
+ Arguments.of("SELECT t.id FROM t JOIN u USING (v)",
EQUALS),
+ Arguments.of("SELECT id FROM t WHERE v =
PARSE_JSON('1')", EQUALS),
+ Arguments.of("SELECT id FROM t WHERE v <> v",
COMPARISON + " '<>'"),
+ Arguments.of(
+ "SELECT id FROM t WHERE v < v",
+ COMPARISON + " '<', because the type has no
order."),
+ Arguments.of(
+ "SELECT id FROM t WHERE v BETWEEN
PARSE_JSON('1') AND PARSE_JSON('2')",
+ COMPARISON),
+ // the SQL converter expands IS [NOT] DISTINCT FROM
into =
+ Arguments.of(
+ "SELECT id FROM t WHERE v IS DISTINCT FROM
PARSE_JSON('1')",
+ EQUALS),
+ Arguments.of(
+ "SELECT id FROM t WHERE v IS NOT DISTINCT FROM
PARSE_JSON('1')",
+ EQUALS),
+ Arguments.of(
+ "SELECT id FROM t WHERE r = r",
+ "An expression of type ROW<`x` VARIANT> cannot
be used in a comparison "
+ + "with '='"),
+ Arguments.of(
+ "SELECT id FROM t WHERE v IN (PARSE_JSON('1'),
PARSE_JSON('2'))",
+ EQUALS),
+ Arguments.of(
+ "SELECT id FROM t WHERE v IN (SELECT v FROM
u)",
+ COMPARISON + " 'IN'"),
+ Arguments.of(
+ "SELECT id FROM t WHERE v > ALL (SELECT v FROM
u)", COMPARISON),
+ Arguments.of(
+ "SELECT id FROM t WHERE EXISTS (SELECT 1 FROM
u WHERE u.v = t.v)",
+ EQUALS),
+ 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 * FROM t MATCH_RECOGNIZE (PARTITION BY
v ORDER BY ts "
+ + "MEASURES A.id AS aid PATTERN (A)
DEFINE A AS A.id > 0)",
+ "Column 'v' of type VARIANT cannot be used as
a PARTITION BY key "
+ + "of MATCH_RECOGNIZE"),
+ Arguments.of(
+ "SELECT * FROM t MATCH_RECOGNIZE (PARTITION BY
id ORDER BY ts, v "
+ + "MEASURES A.id AS aid PATTERN (A)
DEFINE A AS A.id > 0)",
+ "Column 'v' of type VARIANT cannot be used as
an ORDER BY key "
+ + "of MATCH_RECOGNIZE"),
+ Arguments.of(
+ "SELECT * FROM t MATCH_RECOGNIZE (PARTITION BY
id ORDER BY ts "
+ + "MEASURES A.id AS aid PATTERN (A B) "
+ + "DEFINE B AS B.v = A.v)",
+ EQUALS),
+ Arguments.of(
+ "SELECT * FROM f(r => TABLE t PARTITION BY v)",
+ "Column 'v' of type VARIANT cannot be used as
a PARTITION BY key "
+ + "of a table argument"),
+ Arguments.of(
+ "SELECT * FROM f(r => TABLE t PARTITION BY id
ORDER BY (ts, v))",
+ "Column 'v' of type VARIANT cannot be used as
an ORDER BY key "
+ + "of a table argument"));
+ // batch mode does not apply watermarks
+ final Stream<Arguments> streamingOnly =
+ inMode(STREAMING, Arguments.of("SELECT id FROM w", EQUALS));
+ return Stream.concat(bothModes, streamingOnly);
+ }
+
+ private static Stream<Arguments> allowedQueries() {
+ return inBothModes(
+ Arguments.of("SELECT v FROM t UNION ALL SELECT v FROM u"),
+ Arguments.of("SELECT t.v, u.v FROM t JOIN u ON t.id = u.id"),
+ Arguments.of("SELECT v, COUNT(*) OVER (PARTITION BY id ORDER
BY ts) FROM t"),
+ 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 v FROM t WHERE v IS DISTINCT FROM NULL "
+ + "OR v IS NOT DISTINCT FROM NULL"));
+ }
+
+ private static Stream<Arguments> rejectedJoinsPerBatchStrategy() {
+ return Stream.of(
+ Named.of("SortMergeJoin", "HashJoin, NestedLoopJoin"),
+ Named.of(
+ "ShuffleHashJoin",
+ "SortMergeJoin, NestedLoopJoin,
BroadcastHashJoin"),
+ Named.of(
+ "BroadcastHashJoin",
+ "SortMergeJoin, NestedLoopJoin,
ShuffleHashJoin"),
+ Named.of("NestedLoopJoin", "SortMergeJoin, HashJoin"))
+ .flatMap(
+ disabledOperators ->
+ Stream.of(
+ Arguments.of(
+ "SELECT t.id FROM t JOIN u ON
t.v = u.v",
+ disabledOperators),
+ Arguments.of(
+ "SELECT t.id FROM t JOIN u ON
t.v['id'] = u.v['id']",
+ disabledOperators)));
+ }
+
+ private static Stream<Arguments> rejectedTableApiQueries() {
+ final Stream<Arguments> bothModes =
+ inBothModes(
+ tableApi(
+ "groupBy",
+ env ->
+ env.from("t")
+ .groupBy($("v"))
+ .select($("v"),
$("id").count()),
+ GROUPING_KEY),
+ tableApi(
+ "distinct",
+ env -> env.from("t").select($("v")).distinct(),
+ GROUPING_KEY),
+ tableApi(
+ "orderBy",
+ env -> env.from("t").orderBy($("v")),
+ "Column 'v' of type VARIANT cannot be used as
an ORDER BY key"));
+ // the Table API supports INTERSECT and MINUS only on bounded tables
+ final Stream<Arguments> batchOnly =
+ inMode(
+ BATCH,
+ tableApi(
+ "intersect",
+ env ->
+ env.from("t")
+ .select($("v"))
+
.intersect(env.from("u").select($("v"))),
+ "Column 'v' of type VARIANT cannot be used in
INTERSECT"),
+ tableApi(
+ "minus",
+ env ->
+ env.from("t")
+ .select($("v"))
+
.minus(env.from("u").select($("v"))),
+ "Column 'v' of type VARIANT cannot be used in
EXCEPT"));
+ // the Table API supports these OVER windows only in streaming mode
+ final Stream<Arguments> streamingOnly =
+ inMode(
+ STREAMING,
+ tableApi(
+ "over window partitionBy",
+ env ->
+ env.from("t")
+ .window(
+
Over.partitionBy($("v"))
+
.orderBy($("ts"))
+
.preceding(UNBOUNDED_ROW)
+ .as("w"))
+
.select($("id").count().over($("w"))),
+ "An expression of type VARIANT cannot be used
as a PARTITION BY key "
+ + "of an OVER window"),
+ tableApi(
+ "distinct count over window",
+ env ->
+ env.from("t")
+ .window(
+
Over.partitionBy($("id"))
+
.orderBy($("ts"))
+
.preceding(UNBOUNDED_ROW)
+ .as("w"))
+
.select($("v").count().distinct().over($("w"))),
+ "An expression " + DISTINCT_ARGUMENT));
+ return Stream.concat(bothModes, Stream.concat(batchOnly,
streamingOnly));
+ }
+
+ @ParameterizedTest(name = "[{index}] {2}: {0}")
+ @MethodSource("rejectedQueries")
+ void testRejectedQuery(
+ final String sql, final String expectedMessage, final
RuntimeExecutionMode mode) {
+ final TableEnvironment env = createEnvironment(mode);
+ assertThatThrownBy(() -> env.explainSql(sql))
+ .satisfies(anyCauseMatches(ValidationException.class,
expectedMessage));
+ }
+
+ @ParameterizedTest(name = "[{index}] {1}: {0}")
+ @MethodSource("allowedQueries")
+ void testAllowedQuery(final String sql, final RuntimeExecutionMode mode) {
+ final TableEnvironment env = createEnvironment(mode);
+ assertThatCode(() -> env.explainSql(sql)).doesNotThrowAnyException();
+ }
+
+ @ParameterizedTest(name = "[{index}] {1} only: {0}")
+ @MethodSource("rejectedJoinsPerBatchStrategy")
+ void testRejectedJoinPerBatchStrategy(final String sql, final String
disabledOperators) {
+ final TableEnvironment env = createEnvironment(BATCH);
+ env.getConfig().set(TABLE_EXEC_DISABLED_OPERATORS, disabledOperators);
+ assertThatThrownBy(() -> env.explainSql(sql))
+ .satisfies(anyCauseMatches(ValidationException.class, EQUALS));
+ }
+
+ @ParameterizedTest(name = "[{index}] {2}: {0}")
+ @MethodSource("rejectedTableApiQueries")
+ void testRejectedTableApiQuery(
+ final Function<TableEnvironment, Table> query,
+ final String expectedMessage,
+ final RuntimeExecutionMode mode) {
+ final TableEnvironment env = createEnvironment(mode);
+ assertThatThrownBy(() -> query.apply(env).explain())
+ .satisfies(anyCauseMatches(ValidationException.class,
expectedMessage));
+ }
+
+ private static Arguments tableApi(
+ final String name,
+ final Function<TableEnvironment, Table> query,
+ final String expectedMessage) {
+ return Arguments.of(Named.of(name, query), expectedMessage);
+ }
+
+ private static Stream<Arguments> inBothModes(final Arguments... cases) {
+ return Stream.concat(inMode(STREAMING, cases), inMode(BATCH, cases));
+ }
+
+ private static Stream<Arguments> inMode(
+ final RuntimeExecutionMode mode, final Arguments... cases) {
+ return Stream.of(cases).map(c -> withMode(c, mode));
+ }
+
+ private static Arguments withMode(final Arguments arguments, final
RuntimeExecutionMode mode) {
+ final Object[] args = Arrays.copyOf(arguments.get(),
arguments.get().length + 1);
+ args[args.length - 1] = mode;
+ return Arguments.of(args);
+ }
+
+ private static TableEnvironment createEnvironment(final
RuntimeExecutionMode mode) {
+ final TableEnvironment env =
+ TableEnvironment.create(
+ mode == STREAMING
+ ? EnvironmentSettings.inStreamingMode()
+ : EnvironmentSettings.inBatchMode());
+ for (final String table : List.of("t", "u")) {
+ env.executeSql(
+ String.format(
+ "CREATE TABLE %s ("
+ + "id INT, "
+ + "v VARIANT, "
+ + "r ROW<x VARIANT>, "
+ + "a ARRAY<VARIANT>, "
+ + "ts TIMESTAMP(3), "
+ + "WATERMARK FOR ts AS ts"
+ + ") WITH ('connector' = 'values',
'bounded' = '%s')",
+ table, mode == BATCH));
+ }
+ env.executeSql(
+ String.format(
+ "CREATE TABLE w ("
+ + "id INT, "
+ + "v VARIANT, "
+ + "ts TIMESTAMP(3), "
+ + "WATERMARK FOR ts AS "
+ + "CASE WHEN v = PARSE_JSON('1') THEN ts ELSE
ts - INTERVAL '1' SECOND END"
+ + ") WITH ('connector' = 'values', 'bounded' =
'%s')",
+ mode == BATCH));
+ env.createTemporarySystemFunction("f", PartitionedFunction.class);
+ return env;
+ }
+
+ /** A function with a set semantic table argument. */
+ public static class PartitionedFunction extends
ProcessTableFunction<Integer> {
+ public void eval(@ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) final
Row r) {}
+ }
+}
