[raminqaf]

What is the purpose of the change

Flink compares VARIANT values by their binary encoding, but one value has many encodings. Object keys can be stored in any order, a field read from an object keeps the key dictionary of its parent, and a number keeps its storage width. So queries that group, deduplicate, join or compare VARIANT values return wrong results in streaming, without an error.

-- t(s STRING) has three rows: {"a":1,"b":2}, {"b":2,"a":1}, {"a":1,"c":3}
SELECT a, COUNT(*) FROM (SELECT PARSE_JSON(s)['a'] AS a FROM t) GROUP BY a;

Query	Before, streaming	After
the GROUP BY above	three groups, +I[1, 1] each	Column 'a' of type VARIANT cannot be used as a grouping key, because the type has no equality. Cast the value to a comparable type first.
... WHERE PARSE_JSON(s)['a'] = PARSE_JSON('1')	0 rows	An expression of type VARIANT cannot be used in a comparison with '=', because the type has no equality. Cast the value to a comparable type first.
... GROUP BY CAST(PARSE_JSON(s)['a'] AS INT)	one group, count 3	unchanged

Batch already rejected most of these queries late, during code generation, with Type 'VARIANT' cannot be ordered, so it cannot be used as a key for sorting, grouping, or joining. A batch nested loop join still compared VARIANT keys by their bytes and returned wrong rows. Batch now gets the same error as streaming.

This follows Spark and Databricks, which also reject VARIANT where equality or ordering is needed. Users cast to a concrete type first, as in the last row.

Brief change log

• LogicalTypeChecks#isComparableKeyType checks that a type contains no VARIANT, which every grouping, join, partition and sort key and every operand of a comparison needs. This includes a VARIANT nested in a ROW, ARRAY, MAP or structured type.
• A new KeyTypeValidator checks the logical plan before optimization, so it covers SQL and the Table API. It rejects such types in grouping keys, DISTINCT aggregates, also inside OVER windows, set operations other than UNION ALL, ORDER BY, the PARTITION BY and ORDER BY of OVER windows, MATCH_RECOGNIZE and table arguments, comparison operators, also inside watermark expressions, and IN subqueries. Joins are covered through the comparisons in their conditions. A comparison with a NULL literal is allowed.
• The VARIANT section of the data types page lists where VARIANT cannot be compared.

The check runs on the logical plan rather than on exec nodes. It sees the keys the query asks for, and keys the optimizer adds later, for example during decorrelation, only match a value with copies of itself. A job restored from a compiled plan skips the check and keeps running.

Not in this PR:

• Built-in functions that compare VARIANT array elements or map keys, such as ARRAY_CONTAINS and MAP_CONTAINS_KEY, have the same bug. FLINK-40936 covers them in a follow-up PR, because rejecting them also changes how restored compiled plans behave.
• VARIANT as a PRIMARY KEY, and MAP<VARIANT, ...> and MULTISET as types.
• Grouping by MAP or MULTISET values has a related streaming bug and needs its own issue.

Verifying this change

This change added tests and can be verified as follows:

• KeyTypeValidatorTest runs rejected and allowed queries in streaming and batch mode. It covers grouping, DISTINCT aggregates, set operations, ORDER BY, OVER windows, window TVFs, MATCH_RECOGNIZE, table arguments, comparisons, IN lists and subqueries, joins and watermarks, plus Table API cases. It also runs VARIANT joins with each batch join strategy, SortMergeJoin, ShuffleHashJoin, BroadcastHashJoin and NestedLoopJoin.
• VariantSemanticTest turns the reproductions from the JIRA issue into failing-SQL checks, including IS [NOT] DISTINCT FROM, and keeps the CAST workaround as a passing program.
• LogicalTypeChecksTest covers isComparableKeyType.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no. Queries that used VARIANT in these places now fail, which needs a release note.
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code 2.1.292 (Claude Opus 5.5)

[flinkbot]

CI report:

• f793b4e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[snuyanzin]

can we have a query similar to #29285 (comment) (with variants)
and see that it fails if we turnoff all kind of joins except SortMergeJoin?

for instance in this test SortMergeJoinTest

[snuyanzin]

i'm not sure it is true
how about MAP, RAW, STRUCTURED, MULTISET?

[raminqaf]

Good point, VARIANT is not the only type with this problem. I checked on this branch. MAP['a', 1, 'b', 2] = MAP['b', 2, 'a', 1] returns true, so comparing two maps is correct. Grouping is not: streaming GROUP BY m returns two groups for these maps, and batch rejects MAP keys during code generation.
MULTISET is stored as a map, so it should behave the same.
Structured keys compare attribute by attribute, so they only break with a VARIANT or MAP attribute, and a VARIANT attribute is already rejected.
RAW keys compare the serialized bytes.
The MAP and MULTISET grouping bug exists without VARIANT, and Spark fixes it by sorting map entries instead of rejecting them.
So I'd keep it out of this PR and open a JIRA for it. I'll reword the Javadoc so it states what every key needs, without claiming the method decides every key type.

[snuyanzin]

why do we need one more comparable method instead of reuse?

[raminqaf]

areComparable answers a different question: whether a function can compare two values. It says MAP is comparable and even orderable, although batch can't sort MAP keys. It says a structured type with the default EQUALS comparison isn't orderable, although ORDER BY on such a type works today in batch. So using it for keys would reject queries that work today and still let MAP keys through. #29416 goes the other way: areComparable calls this method, so VARIANT has one rule in both places.

[snuyanzin]

can we also have tests with IS [NOT ] DISTINCT FROM?

[raminqaf]

added

[raminqaf]

can we have a query similar to #29285 (comment) (with variants) and see that it fails if we turnoff all kind of joins except SortMergeJoin?

for instance in this test SortMergeJoinTest

Thanks, good check. I ran a VARIANT version of your probe in batch with table.exec.disabled-operators = 'HashJoin, NestedLoopJoin', as in SortMergeJoinTest, and also with 'ShuffleHashJoin, BroadcastHashJoin'. Both SELECT a, d FROM Tiny, Huge WHERE m = n and ... ON m['id'] = n['id'] now fail with An expression of type VARIANT cannot be used in a comparison with '=', because the type has no equality. Without this PR they plan a SortMergeJoin on the VARIANT key. The check runs before the optimizer picks a join strategy, so the disabled operators make no difference. The CAST workaround still plans a SortMergeJoin.

I will also add an test to cover this