[
https://issues.apache.org/jira/browse/FLINK-40910?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40910:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
