[
https://issues.apache.org/jira/browse/SPARK-58486?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58486:
-----------------------------------
Labels: correctness pull-request-available (was: correctness)
> InjectRuntimeFilter builds runtime bloom filters with collation-agnostic
> XxHash64, silently dropping join rows when a collated equi-join key is a
> non-attribute expression
> --------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-58486
> URL: https://issues.apache.org/jira/browse/SPARK-58486
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.1.0, 4.0.1, 4.2.0
> Reporter: Josh Rosen
> Priority: Major
> Labels: correctness, pull-request-available
>
> {{InjectRuntimeFilter}} builds and probes its runtime bloom filter with the
> plain {{XxHash64}}
> ([InjectRuntimeFilter.scala#L62-L79|https://github.com/apache/spark/blob/c7b2f1a865cce621d2c702bdfdff5132e251ca74/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala#L62-L79]).
> Since SPARK-52828 (4.0.1 / 4.1.0), {{XxHash64}} hashes collated strings by
> raw UTF-8 bytes while the join's equality remains collation-aware, so for a
> collated equi-join whose key is a non-attribute expression, two values that
> are equal under the join's collation but byte-different (e.g. {{'abc'}} vs
> {{'ABC'}} under {{{}UTF8_LCASE{}}}) hash differently,
> {{BloomFilterMightContain}} returns false, and matching probe-side rows are
> filtered out before the join runs: silently missing output rows under default
> configs (runtime bloom filters are on by default).
> h3. Reproduction
> Reproduced on current master (c7b2f1a865c). Default configs except the two
> thresholds below, which only make the tiny tables eligible for injection
> (they gate sizes, not semantics):
> {code:sql}
> CREATE TABLE dim(key STRING COLLATE UTF8_LCASE, cat STRING) USING parquet;
> INSERT INTO dim VALUES ('ABC', 'x');
> CREATE TABLE fact USING parquet AS
> SELECT concat('abc', CAST(id AS STRING)) COLLATE UTF8_LCASE AS k FROM
> range(5000);
> SET spark.sql.autoBroadcastJoinThreshold=-1;
> SET
> spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1;
> -- Every fact row satisfies substr(k, 1, 3) = 'abc', and 'abc' = 'ABC' under
> UTF8_LCASE,
> -- so this should return 5000:
> SELECT count(*) FROM fact f JOIN dim d ON substr(f.k, 1, 3) = d.key WHERE
> d.cat = 'x';
> {code}
> {noformat}
> spark.sql.optimizer.runtime.bloomFilter.enabled=true (default) -> 0
> (WRONG)
> spark.sql.optimizer.runtime.bloomFilter.enabled=false -> 5000
> (correct)
> {noformat}
> Toggling only {{spark.sql.optimizer.runtime.bloomFilter.enabled}} (which
> gates nothing but {{InjectRuntimeFilter)}} flips the result, and the
> optimized plan of the failing run contains the injected {{might_contain(...,
> xxhash64(...))}} probe (the {{BloomFilterAggregate}} build side sits inside a
> {{{}ScalarSubquery{}}}).
> h3. Root cause
> SPARK-52828
> ([f62724d2d8d|https://github.com/apache/spark/commit/f62724d2d8dc0f405c9d0dedcd0136bf4a4aa3b7],
> first released in 4.1.0; backported to branch-4.0 as
> [542faa70cef|https://github.com/apache/spark/commit/542faa70cefe6526387f2bfaebde546e4e562ace],
> first released in 4.0.1) made {{Murmur3Hash}} / {{XxHash64}}
> collation-agnostic by default (raw-byte hashing for collated strings unless
> {{{}spark.sql.legacy.collationAwareHashFunctions=true{}}},
> [SQLConf.scala#L1263-L1270|https://github.com/apache/spark/blob/c7b2f1a865cce621d2c702bdfdff5132e251ca74/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L1263-L1270])
> and migrated the internal consumers to collation-aware variants – shuffle
> partitioning to {{{}CollationAwareMurmur3Hash{}}}, hash-join keys via
> {{CollationKey}} – but did not update {{{}InjectRuntimeFilter{}}}, which
> still emits plain {{XxHash64}} on both the build side (L64/L66) and the probe
> side (L79).
> This is masked in the common case: {{RewriteCollationJoin}}
> ([RewriteCollationJoin.scala#L38-L41|https://github.com/apache/spark/blob/c7b2f1a865cce621d2c702bdfdff5132e251ca74/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteCollationJoin.scala#L38-L41])
> rewrites join equalities to {{CollationKey(l) = CollationKey(r)}} before the
> optimizer runs, so for those keys the bloom filter hashes collation-key bytes
> on both sides and stays correct. But its pattern matches only {{{}Equality(l:
> AttributeReference, r: AttributeReference){}}}: a non-attribute collated key
> ({{{}substr(k, 1, 3){}}}, {{{}concat(a, b){}}}, {{{}upper(k){}}}, ...)
> escapes the wrap. The join itself still compares correctly
> ({{{}HashJoin.injectCollationKey{}}} normalizes keys at planning,
> [HashJoin.scala#L745-L752|https://github.com/apache/spark/blob/c7b2f1a865cce621d2c702bdfdff5132e251ca74/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala#L745-L752]),
> which is why the raw-byte pre-filter produces wrong results.
> h3. Related tickets
> * SPARK-52828 is the introducing change; on 4.0.0 and earlier {{XxHash64}}
> was collation-aware and this path was correct.
> * SPARK-57055 originally reported this same defect and was retracted by its
> reporter on the grounds that {{RewriteCollationJoin}} pre-wraps join keys in
> {{{}CollationKey{}}}, making the bloom filter symmetric. That reasoning holds
> only for bare attribute-to-attribute equalities (the rule's pattern); the
> reproduction above – a non-attribute join key – is the case it does not
> cover. That ticket has since been rescoped to documenting the (by-design)
> byte-wise behavior of {{{}DataFrameStatFunctions.bloomFilter{}}}, an adjacent
> but distinct path.
> * SPARK-48000 introduced {{RewriteCollationJoin}} (whose attribute-only
> pattern bounds the exposure); SPARK-50228 moved it ahead of the optimizer,
> which is what protects the bare-attribute case.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
