[
https://issues.apache.org/jira/browse/FLINK-41013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127781#comment-18127781
]
Lorand Bendig commented on FLINK-41013:
---------------------------------------
We ran into the same problem independently while working on FLINK-31207. Making
{{0.0}} and {{-0.0}} hash the same affects everything that goes through
{{{}CodeGenUtils#hashCodeForType{}}}, not only the three collection functions.
h3. Impact on batch jobs
In batch jobs, a hash shuffle distributes rows by the hash of the partition
key. Since {{0.0}} and {{-0.0}} have different hash codes, they may be sent to
different subtasks, depending on the parallelism. Operators that use SQL
equality, such as {{{}OVER (PARTITION BY ...){}}}, treat the two values as
equal, but they partition the data locally on each subtask. If both values land
on the same subtask, they form one partition; otherwise, they form two.
{code:sql}
SET 'execution.runtime-mode' = 'batch';
SET 'parallelism.default' = '4';
SET 'table.exec.resource.default-parallelism' = '4';
SET 'execution.batch.adaptive.auto-parallelism.enabled' = 'false';
SELECT
CAST(s AS DOUBLE) AS d,
COUNT(*) OVER (PARTITION BY CAST(s AS DOUBLE))
FROM (
VALUES
('-0.0'),
('0.0')
) AS T(s);
{code}
With parallelism of 4:
{noformat}
+------+--------+
| d | EXPR$1 |
+------+--------+
| 0.0 | 1 |
| -0.0 | 1 |
+------+--------+
{noformat}
With parallelism of 1:
{noformat}
+------+--------+
| d | EXPR$1 |
+------+--------+
| -0.0 | 2 |
| 0.0 | 2 |
+------+--------+
{noformat}
With the fix, both values always reach the same subtask, so the result is 2 for
both rows at any parallelism.
h3. Split distinct aggregation: wrong counts after a restore
When a streaming job sets {{table.optimizer.distinct-agg.split.enabled}} to
{{{}true{}}}, {{COUNT(DISTINCT ...)}} is computed in two steps. Records are
first distributed across buckets by the hash of the distinct key, aggregated
per bucket, and then merged.
If all these conditions are met:
* {{table.optimizer.distinct-agg.split.enabled}} is {{{}true{}}},
* the bucket count ({{{}table.optimizer.distinct-agg.split.bucket-num{}}}) is
not a power of 2 (the default, 1024, is not affected),
* the distinct column is FLOAT or DOUBLE and contains {{{}-0.0{}}},
* the job is restored from state written before the fix,
then after the restore, the job looks for {{-0.0}} in a different bucket than
the one where its old state is stored. This can lead to wrong counts.
h3. Sink upsert materializer: stale rows after a restore
If all these conditions are met:
* {{table.exec.sink.upsert-materialize-strategy.type}} is {{MAP}} or
{{ADAPTIVE}},
* the heap state backend ({{HashMapStateBackend}}) is used,
* rows containing {{-0.0}} are written to an upsert sink,
then the following can happen:
Flink keeps the current rows for each primary key in a MapState (HashMap) in
front of an upsert sink. On restore, the heap backend rebuilds the HashMap
using the old hash function stored in the checkpoint. If the materializer's
hash function also changes, the restored job would look up a row with -0.0
using a different
hash code and would miss the stored row. The DELETE would be silently dropped,
and the sink would retain a row that should have been deleted.
To prevent this, the materializer must continue using the old hash function.
Spark also fixed the same inconsistency in SPARK-26021 .
> Use a hash consistent with SQL equality to match MAP keys and array elements
> ----------------------------------------------------------------------------
>
> Key: FLINK-41013
> URL: https://issues.apache.org/jira/browse/FLINK-41013
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Planner, Table SQL / Runtime
> Reporter: Ramin Gharib
> Priority: Major
>
> h3. Problem
> Since FLINK-40989, {{MAP}} and {{MULTISET}} equality match keys with SQL
> {{=}}. Each left key is compared with every right key, which is quadratic in
> the map size. {{ARRAY_DISTINCT}}, {{ARRAY_UNION}} and {{MAP_UNION}} use the
> same nested loop.
> {{ARRAY_EXCEPT}}, {{ARRAY_INTERSECT}} and {{MAP_FROM_ENTRIES}} already hash
> elements through {{ObjectContainer}}, with a generated {{=}} and
> {{$HASHCODE$}}. Hashing only works when equal elements get equal hashes.
> {{$HASHCODE$}} hashes DOUBLE with {{Double.hashCode}}, so 0.0 and -0.0 get
> different hashes although {{0.0 = -0.0}} is TRUE. These three functions then
> give wrong results. With {{f0 = 0.0}} as a DOUBLE column:
> ||Query||Expected||Actual||
> |{{CARDINALITY(ARRAY_EXCEPT(ARRAY[f0], ARRAY[f0 * -1E0]))}}|0|1|
> |{{CARDINALITY(ARRAY_INTERSECT(ARRAY[f0], ARRAY[f0 * -1E0]))}}|1|0|
> |{{CARDINALITY(MAP_FROM_ENTRIES(ARRAY[ROW(f0, 1), ROW(f0 * -1E0, 2)]))}}|1|2|
> {{f0 * -1E0}} is -0.0 at runtime, and {{CAST(f0 * -1E0 AS STRING)}} returns
> {{-0.0}}. {{ARRAY_DISTINCT(ARRAY[f0, f0 * -1E0])}} and
> {{ARRAY_UNION(ARRAY[f0], ARRAY[f0 * -1E0])}} correctly return one element,
> because their nested loop uses {{=}} directly.
> h3. Proposal
> * Make the element hash agree with SQL {{=}}. For example, map -0.0 to 0.0
> before hashing FLOAT and DOUBLE, also inside ROW, ARRAY and MAP elements.
> * Use hash-based matching with that hash in the generated MAP and MULTISET
> equality, {{ARRAY_DISTINCT}}, {{ARRAY_UNION}} and {{MAP_UNION}}, so that they
> run in linear time.
> * A MAP used as a key or element can hold a key more than once, and then its
> last entry counts. Its hash must only count that last entry.
> h3. Tests
> * The queries above, for FLOAT and DOUBLE.
> * BINARY elements, and ROW elements where one side is generic data and the
> other binary data.
> * A large map or array, to check the linear time.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
