Hash partitioning not falling back to Spark for high precision decimal partition keys #3079

Activity

1. andygrove

   added

   bugSomething isn't workingSomething isn't working

   on Jan 13, 2026
2. andygrove

   self-assigned this

   on Jan 13, 2026
3. andygrove

   added this to the 0.13.0 milestone on Jan 13, 2026
4. andygrove commented on Jan 13, 2026

   

   andygrove

   on Jan 13, 2026

   MemberAuthor

   More actions

   This is trivial to fix but requires updating some golden files.
5. andygrove commented on Jan 13, 2026

   

   andygrove

   on Jan 13, 2026

   MemberAuthor

   More actions

   Acutally, this is not a bug. We do not need to guarantee that we allocate the same partitions as Spark.
6. andygrove

   closed this as completedon Jan 13, 2026
7. andygrove

   reopened this on Sep 22, 2026
8. andygrove commented on Sep 22, 2026

   

   andygrove

   on Sep 22, 2026

   MemberAuthor

   More actions

   Reopening. I closed this saying we don't need to allocate the same partitions as Spark, and that was wrong.

   We have two shuffle implementations and both can appear in the same plan. applyCometShuffle picks native or columnar per exchange with no harmonization across the plan, so the two sides of a join can end up with different ones. The columnar path uses Spark's HashPartitioning.partitionIdExpression and the native path uses the Rust murmur3. On a decimal key with precision above 18 those disagree. So the native path does have to match Spark, not for its own sake, but to stay consistent with its sibling.

   On main at 9a4d5f283, joining a Parquet table to a JSON view on a DECIMAL(38,0) key:

   CometSortMergeJoin [k#37], [k#40], Inner
   :- CometSort [k#37, a#38], [k#37 ASC NULLS FIRST]
   :  +- CometExchange hashpartitioning(k#37, 7), ENSURE_REQUIREMENTS, CometNativeShuffle
   :     +- CometNativeScan parquet spark_catalog.default.wd_parquet[k#37,a#38]
   +- CometSort [k#40, b#41], [k#40 ASC NULLS FIRST]
      +- CometColumnarExchange hashpartitioning(k#40, 7), ENSURE_REQUIREMENTS, CometColumnarShuffle
         +- FileScan json [k#40,b#41]
   

   Comet returns zero rows where Spark returns four. Default spark.comet.shuffle.mode=auto, nothing exotic set. It reproduces with AQE enabled too, as long as coalescing doesn't collapse the fixture into a single partition, which is what you'd get at any real data volume. That's most likely why the suites have never caught it.

   The mismatch itself is in hash_array_decimal!, which hashes value.to_le_bytes(), a fixed sixteen-byte little-endian i128. Spark hashes toJavaBigDecimal.unscaledValue.toByteArray(), which is minimal-length big-endian two's complement. Byte order and length both differ, and murmur3 mixes length into fmix, so every value diverges. Unscaled 1 in DECIMAL(38,0) at seed 42 hashes to -386724586 in Spark and -680163996 in Comet, which lands in partition 4 against partition 6 out of 7.

   #6005 implements the guard described in the original report. #5994 tracks making the native encoding match Spark, which would remove the need for the guard and also unblock hash and xxhash64 for these types.
9. andygrove

   modified the milestones: 0.13.0, 1.1.0 on Sep 22, 2026
10. andygrove

    added

    priority:criticalData corruption, silent wrong results, security issuesData corruption, silent wrong results, security issues

    on Sep 22, 2026
11. andygrove commented on Sep 24, 2026

    

    andygrove

    on Sep 24, 2026

    MemberAuthor

    More actions

    A cached relation makes this reachable from a query that has Comet turned off. The cache keeps the partitions native shuffle produced, and the cache scan reports the cached plan's hashpartitioning(k, 200) to every later query. A later join on k then shuffles only the other side, so if that side goes through Spark's hash, the join silently drops rows. That happens with Comet off for the query, or with Comet on when the other side's shuffle is columnar.

    On main at 67803a7, default Spark 4.1 profile, with spark.sql.autoBroadcastJoinThreshold=-1:

    spark.range(0, 10000, 1, 4)
      .selectExpr("cast(id as decimal(38,0)) as k", "id as v")
      .write.parquet(pq)
    spark.range(0, 10000, 1, 3)
      .selectExpr("cast(id as decimal(38,0)) as k2", "id as w")
      .write.parquet(pq2)
    
    val cached = spark.read.parquet(pq).repartition($"k").cache()
    cached.count() // the cached plan shuffles with CometNativeShuffle into 200 partitions
    
    spark.conf.set("spark.comet.enabled", "false")
    cached.join(spark.read.parquet(pq2), $"k" === $"k2").count() // 57, expected 10000

    Other side of the join	Rows
    Comet off, Parquet (Spark shuffle)	57 / 10000
    Comet on, JSON (CometColumnarShuffle)	57 / 10000
    Comet on, Parquet (CometNativeShuffle)	10000 / 10000

    It is the same with AQE on or off, and with spark.comet.exec.inMemoryCache.enabled on or off, since the partitioning comes from the cached plan rather than the cache format. decimal(20,2) keys fail too. Decimal keys of precision 18 or less, and integer, string, binary, date, timestamp, timestamp_ntz, boolean, float and double keys, all join correctly after a native-shuffled cache.

    The first row is the new part: without a cache, a query with Comet off never sees a native shuffle. #6005 should fix this too, since it keeps wide-decimal keys off native shuffle, and a cached case might be worth adding to its tests.
12. msafonov

    added a commit that references this issue on Oct 1, 2026

    fix: do not hash decimals above precision 18 in native shuffle (port of

    ...

    88cfb29
13. andygrove

    modified the milestones: 1.1.0, 1.2.0 on Oct 11, 2026