[jira] [Created] (SPARK-60085) Reduce hash clustering for Float/Double keys in constant maps

["Varun Bhandary (Jira)" <ji...@apache.org>]

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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org