Ramin Gharib created FLINK-40910:
------------------------------------
Summary: Reject VARIANT where SQL needs equality. Byte comparison
gives wrong results in GROUP BY, DISTINCT, joins and =
Key: FLINK-40910
URL: https://issues.apache.org/jira/browse/FLINK-40910
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner, Table SQL / Runtime
Reporter: Ramin Gharib
Assignee: Ramin Gharib
h2. Problem
Flink compares VARIANT values by their binary encoding. One logical value has
many encodings. Queries that group, deduplicate, join or compare on VARIANT
therefore return wrong results, without any error.
{code:sql}
-- t(s STRING) has three rows:
-- row 1: \{"a":1,"b":2}
-- row 2: \{"b":2,"a":1}
-- row 3: \{"a":1,"c":3}
SELECT a, COUNT(*) FROM (SELECT PARSE_JSON(s)['a'] AS a FROM t) GROUP BY a;
{code}
{noformat}
Query Expected
Actual
GROUP BY PARSE_JSON(s)['a'] one group, count 3
three groups, count 1 each
COUNT(DISTINCT PARSE_JSON(s)) over row 1 and row 2 1
2
WHERE PARSE_JSON(s)['a'] = PARSE_JSON('1') 3 rows
0 rows
l JOIN r ON PARSE_JSON(l.s)['id'] = PARSE_JSON(r.s)['id'] 1 row
0 rows
with l = \{"id":7,"x":1} and r = \{"y":2,"id":7}
{noformat}
The actual results come from VariantSemanticTest on master 5afe863af62. Example
output for the GROUP BY:
{noformat}
Expecting actual:
["+I[1, 1]", "+I[1, 1]", "+I[1, 1]"]
to contain exactly in any order:
["+I[1, 1]", "-U[1, 1]", "+U[1, 2]", "-U[1, 2]", "+U[1, 3]"]
{noformat}
h2. Cause
A VARIANT is two byte arrays. The metadata is a dictionary of key names. The
value refers to keys by their position in that dictionary. Flink treats two
VARIANTs as equal only if both arrays are byte-equal:
* Grouping, DISTINCT and join keys are BinaryRowData.
\{{AbstractBinaryWriter#writeVariant}} writes the dictionary and the value
bytes into the key, and BinaryRowData hashes and compares those bytes.
* \{{=}} calls \{{BinaryVariant#equals}}, which compares the value, the
dictionary and the position.
One value has many valid encodings:
* The dictionary lists keys in the order they first appear. Row 1 and row 2 get
different dictionaries.
* A field taken by field access keeps the full dictionary of its parent object.
The number 1 from row 1 differs from the number 1 from row 3 and from
PARSE_JSON('1').
* Numbers keep their storage width. PARSE_JSON('1') is a TINYINT, CAST(CAST(1
AS BIGINT) AS VARIANT) is a BIGINT, and PARSE_JSON('1.0') is a DECIMAL. All
three print as 1.
The variant encoding spec allows all of these. For the same reason, the Iceberg
spec defines no hash for variant.
h2. Proposal
Reject VARIANT where SQL needs equality, like ORDER BY on VARIANT is rejected
today (\{{TypeCheckUtils#isComparable}}). Users cast to a concrete type first.
That already works:
{code:sql}
SELECT a, COUNT(*) FROM (SELECT CAST(PARSE_JSON(s)['a'] AS INT) AS a FROM t)
GROUP BY a;
-- one group, count 3
{code}
Reject VARIANT, also when nested in ROW, ARRAY or MAP, in:
* GROUP BY keys, including GROUPING SETS, ROLLUP, CUBE and window aggregations
* DISTINCT aggregates and SELECT DISTINCT
* equi-join keys
* =, <>, IN, IS DISTINCT FROM and IS NOT DISTINCT FROM between VARIANT operands
* UNION, INTERSECT and EXCEPT without ALL
* PARTITION BY of OVER windows, Top-N, deduplication, process table functions
and MATCH_RECOGNIZE
Still allowed: VARIANT as an aggregate argument such as COUNT(v),
FIRST_VALUE(v) and LAST_VALUE(v), IS NULL, IS NOT NULL, field access and casts.
The error should name the clause and suggest a cast, for example:
{noformat}
VARIANT cannot be used as a grouping key because VARIANT values have no
equality. Cast the value to a concrete type first, for example CAST(v['a'] AS
INT).
{noformat}
Open question: VARIANT as a MAP key. MAP lookups use the same byte equality.
h2. Other systems
||System||Behavior||
|Spark / Databricks|VARIANT is not comparable since SPARK-47569. GROUP BY fails
with GROUP_EXPRESSION_TYPE_IS_NOT_ORDERABLE. DISTINCT and set operations fail.
PARTITION BY fails with PARTITION_BY_VARIANT. Databricks docs: "The VARIANT
data type cannot be used for comparisons, grouping, ordering, and set
operations."|
|BigQuery|JSON is neither groupable nor comparable, and cannot be a partition
or cluster key. Users extract with JSON_VALUE first.|
|Apache Iceberg|The spec defines no hash for variant, because equivalent values
have several representations.|
Sources:
* https://docs.databricks.com/aws/en/semi-structured/variant
*
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
(checkValidGroupingExprs)
*
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
(variantColumnInSetOperation, PARTITION_BY_VARIANT)
* https://cloud.google.com/bigquery/docs/reference/standard-sql/data-types
* https://github.com/apache/iceberg/blob/main/format/spec.md
h2. Compatibility
* Flink 2.1 to 2.3 accept these queries. After the change they fail at
validation. This needs a release note.
* Existing tests assert the old behavior and must change: VariantSemanticTest
VARIANT_AS_AGG_KEY, and the COUNT(DISTINCT v) in BUILTIN_AGG and
BUILTIN_AGG_WITH_RETRACTION.
* Open question: a job restored from a compiled plan skips SQL validation.
Should it fail too, or keep running?
* Supporting equality later stays compatible, because it only accepts more
queries.
h2. Alternatives considered
* Write a field value with only the keys it uses. This fixes scalar fields, but
not objects or number widths.
* A canonical encoding: sorted dictionary, no unused keys, smallest number
width. This changes the bytes of existing state, and data from other writers
still differs because the spec allows several encodings.
* Semantic equality and hashing on decoded values. Correct, but a larger
design: how 1 compares to 1.0, NaN, and the timestamp kinds. It can follow
later.
h2. Tests
Four VariantSemanticTest programs reproduce the bug and fail on master as shown
above: variant-field-as-agg-key, variant-object-as-distinct-key,
variant-field-equals-variant and variant-field-as-join-key. With the fix they
become failing-SQL checks for the new error. A fifth program,
variant-field-cast-as-agg-key, checks the CAST workaround and passes today.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
