Varun Bhandary created SPARK-60085:
--------------------------------------
Summary: Reduce hash clustering for Float/Double keys in constant
maps
Key: SPARK-60085
URL: https://issues.apache.org/jira/browse/SPARK-60085
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 4.2.0, 5.0.0
Reporter: Varun Bhandary
h3. Problem and relevance
A generated lookup into a constant map containing 1,000 Float/Double category
IDs can check hundreds of buckets per row. This affects SQL map expressions and
TargetEncoder's category mappings under default lookup and code-generation
settings. Results remain correct; the problem is avoidable work repeated across
input rows.
Spark precomputes a hash table for constant (foldable) maps with at least 1,000
entries. These maps stay fixed while lookup keys vary per row. [TargetEncoder
uses this
pattern|https://github.com/apache/spark/blob/0c14861e323174452d42ffb7947d1c1158c39286/mllib/src/main/scala/org/apache/spark/ml/feature/TargetEncoder.scala#L381-L385]
to map category IDs to learned values.
h3. Why it happens
For 1,000 entries, the table contains 2,048 buckets, selected using the hash's
lowest 11 bits. The current Float hash uses its bit representation; Double
XOR-folds its 64 bits into 32. For both types, keys {{0.0}} through {{999.0}}
all initially select bucket zero. Collision handling stores them in successive
slots, creating one long cluster. [Pinned
implementation|https://github.com/apache/spark/blob/0c14861e323174452d42ffb7947d1c1158c39286/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/complexTypeExtractors.scala#L558-L610].
Spark's actual tables require *500.5 bucket checks on average* for uniformly
distributed successful lookups. This structural count is independent of timing.
h3. Evidence and tradeoff
A local reference candidate using Spark's existing Murmur3 utility reduces that
average to *1.442 for Float and 1.418 for Double*. Against master
{{0c14861e323174452d42ffb7947d1c1158c39286}}, repeated execution shows a
benefit and a cost:
||Workload||Rows||Baseline (ms)||Candidate (ms)||Change across paired runs||
| TargetEncoder, 1,000 categories, successful lookups | 100,000 | 127.32 |
12.01 | 9.69–13.29× faster |
| Random fractional Double map, 10,000 keys; all misses using keys −1 through
−10,000 | 1,000,000 | 29.62 | 38.75 | 25.2–30.8% slower |
Each variant ran in three fresh JVMs; times are medians of their per-case
medians. Each case used five samples after at least two seconds of warmup for
TargetEncoder and five for the map, reversing variant and case order in round
two. Preparation and fitting were excluded; aggregates were independently
checked. Plans and generated code confirmed the path. These are constructed
workloads on one M3 Pro/JDK 21 machine, with one worker and a 4 GiB heap.
The slower miss case does not perform more average probing (1.5332 → 1.5294
checks). Extra hash arithmetic and a wider set of visited buckets accompany it;
their individual timing contributions are unprofiled. Dense-map unseen keys
improve structurally too; this random-map result does not establish a slowdown
for unseen TargetEncoder categories. Existing hashing already distributes
random map keys well.
h3. Proposed direction
Improve Float/Double hash distribution consistently in table construction and
generated lookup. Use Murmur3 as a reference and evaluate a cheaper mixer using
all 64 Double bits before selecting the implementation. Preserve lookup
semantics, signed-zero equality, NaN handling, other key types and strategy
selection. No public APIs, settings or dependencies change.
This addresses a distribution problem in [SPARK-55959's hash
lookup|https://issues.apache.org/jira/browse/SPARK-55959]. The adjacent
[null-key work in PR #58875|https://github.com/apache/spark/pull/58875] and
TargetEncoder's repeated lookup are separate concerns.
h3. Reproduction
With default lookup/codegen settings:
{code:sql}
SET spark.sql.adaptive.enabled=false;
SET spark.sql.shuffle.partitions=1;
SELECT sum(element_at(
map_from_arrays(
CAST(sequence(0, 999) AS ARRAY<DOUBLE>),
sequence(0, 999)),
CAST(id % 1000 AS DOUBLE))) AS checksum
FROM range(100000);
{code}
Expected checksum: {{49950000}}. Replace both {{DOUBLE}} casts with {{FLOAT}}
for that case. Prepend {{EXPLAIN CODEGEN}} to the {{SELECT}} to inspect the
bucket loop. This demonstrates the affected path; the timings above require the
separate benchmark harness.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
