[MINOR] Do not return a negative Spark partition when a hash is Integer.MIN_VALUE - #19776
[MINOR] Do not return a negative Spark partition when a hash is Integer.MIN_VALUE#19776PDGGK wants to merge 2 commits into
Conversation
…er.MIN_VALUE CoalescingPartitioner and PartitionPathRDDPartitioner both derive the partition as Math.abs(hash) % numPartitions. Math.abs leaves Integer.MIN_VALUE negative, so the expression is negative whenever numPartitions does not divide 2^31, and Partitioner#getPartition has to answer inside [0, numPartitions). Use Math.floorMod, which agrees with the old expression for every hash the old one handled correctly. BucketIndexUtil and JavaUpsertPartitioner already avoid Math.abs the same way.
voonhous
left a comment
voonhous
left a comment
There was a problem hiding this comment.
Fix looks correct and complete: the repo-wide sweep of extends Partitioner (Spark) and the Flink partitioners finds no other Math.abs(hash) % n site (HoodieTableMetadataUtil:960 uses the double-abs form, which is MIN_VALUE-safe). Inline comments cover the PR text and test strength.
Two things with no line to anchor on:
- nit, optional: 197 of the last 200 master commits use conventional-commit titles, and
pr_title_validation.ymllabels[MINOR]the legacy format (still accepted).fix(spark): do not return a negative Spark partition when a hash is Integer.MIN_VALUEwould match. - Unrelated, spotted while checking the floorMod precedent:
TestBucketizedBloomCheckPartitioner.java:190asserts0 <= partition && partition <= 1000for a 1000-partition partitioner; the upper bound should be< 1000. Worth a separate one-line PR rather than here.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19776 +/- ##
============================================
+ Coverage 78.11% 78.30% +0.19%
- Complexity 33673 33873 +200
============================================
Files 2540 2541 +1
Lines 141413 141708 +295
Branches 17123 17327 +204
============================================
+ Hits 110467 110970 +503
+ Misses 23250 23029 -221
- Partials 7696 7709 +13
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR replaces Math.abs(hash) % numPartitions with Math.floorMod(...) in CoalescingPartitioner and PartitionPathRDDPartitioner to prevent a negative Spark partition index when a hash is Integer.MIN_VALUE, and adds tests exercising that case. The fix looks correct; the substantive points around test coverage and sibling partitioners are already well covered in the existing inline comments from @voonhous. Please take a look at those inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. One minor naming nit on the new test class, otherwise the fix and tests are clean.
cc @yihua
| @Test | ||
| void assertFixtureStillOverflowsToMinValue() { | ||
| assertEquals(Integer.MIN_VALUE, Objects.hash(MIN_VALUE_HASH_PATH)); | ||
| } |
There was a problem hiding this comment.
🤖 nit: the assert prefix reads like a shared assertion helper rather than a test case — could you rename it to something like fixtureHashOverflowsToMinValue or testFixtureOverflowsToMinValue to make its role as a @Test clear at a glance? Same thought applies to assertPartitionIsInRangeForMinValueHash on line 47.
34ffce9 to
7f64828
Compare
Asserting only that the index is in range passes for abs-mod too. Spark's HashPartitioner routes with Utils.nonNegativeMod, which is floorMod, so comparing against it pins the exact index and is a real oracle. The range-only assertions also let a wrong claim into the PR description: abs-mod and floorMod differ for roughly half of all negative hashes at every parallelism, not only for Integer.MIN_VALUE.
7f64828 to
aabf146
Compare
|
You're right, sorry — I had the impact wrong, not just the wording. Reworded, and I moved the measured numbers into the Impact section rather than only fixing the two sentences, since the mistake was in what I thought the change did. Both tests now assert equality with Spark's I did take the optional one. |
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR replaces Math.abs(hash) % numPartitions with Math.floorMod in CoalescingPartitioner and PartitionPathRDDPartitioner so getPartition always returns an index in [0, numPartitions), fixing the Integer.MIN_VALUE edge case, and adds tests that pin routing against Spark's HashPartitioner as the oracle. No new issues flagged from this automated pass beyond what prior rounds already raised — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
5 participants
Describe the issue this Pull Request addresses
Two Spark
Partitionerimplementations derive the partition index asMath.abs(hash) % numPartitions:Math.abs(Integer.MIN_VALUE)isInteger.MIN_VALUE, so the expression stays negative whenevernumPartitionsdoes not divide 2^31 — that is, for every parallelism that is not a power of two.Partitioner#getPartitionhas to answer inside[0, numPartitions).Both are reachable from ordinary data:
CoalescingPartitioner"polygenelubricants"(hashCodeisInteger.MIN_VALUE)PartitionPathRDDPartitioner"xfjfxsf"Objects.hash(x)is31 + x.hashCode(), so the second one needs a partition path hashing to2147483617for the sum to overflow toInteger.MIN_VALUE;"xfjfxsf"is such a value.Summary and Changelog
Both now use
Math.floorMod, which is non-negative for every input. This is the same expression Spark's ownHashPartitioneruses (Utils.nonNegativeMod), and the oneJavaUpsertPartitioneralready uses;BucketIndexUtilavoidsMath.absthe same way with(hash & Integer.MAX_VALUE) % parallelism.Tests, both asserting against
org.apache.spark.HashPartitioneras the oracle rather than only checking the index is in range — a range check passes for the old expression too on most inputs, which is not a strong enough assertion here:TestCoalescingPartitioner#testPartitionMatchesSparkHashPartitioner— added to the existing class; asserts the fixture still hashes toInteger.MIN_VALUEfirst, so it cannot silently stop exercising that case, then compares against the oracle for negative and non-negative keys at 1..16 partitions.TestPartitionPathRDDPartitioner— new,@ParameterizedTestover the same parallelisms, assertingObjects.hashstill overflows for the fixture.TestCoalescingPartitioner#testShuffleSucceedsForMinValueHashKey— drives a realpartitionByoverjscso the failure is visible end to end rather than only throughgetPartition.Reverting the source change turns them red with the exact index, e.g.
key polygenelubricants at numPartitions 3 ==> expected: <1> but was: <-2>, and the shuffle test withArrayIndexOutOfBoundsException: Index -2 out of bounds for length 3fromBypassMergeSortShuffleWriter, which indexespartitionWriterswith the answer unguarded.mvn test -pl hudi-client/hudi-spark-client -Dtest='*Partitioner*'— 59 tests, all passing.checkstyle:checkclean.Impact
No public API, config, or on-disk change.
An earlier version of this description claimed the two expressions agree except at
Integer.MIN_VALUE, and that power-of-two parallelism routes unchanged. That is wrong, and thanks to @voonhous for catching it.Math.abs(h) % nandMath.floorMod(h, n)agree for every non-negativeh, but for negativehthey agree only whenndividesh(or, for evenn, when the remainder is exactlyn/2). Measured over negative hashes:So roughly half of all keys move to a different partition, at every parallelism above two —
Math.abs(-1) % 8is1whereMath.floorMod(-1, 8)is7.That is a shuffle-placement change, not a correctness or compatibility one. Both partitioners are used only to route records within a single write job (
SparkStreamingMetadataWriteHandlerand the three bulk-insert repartition partitioners); the index is never persisted or compared across runs, and both expressions are deterministic functions of the hash, so records sharing a key or partition path still land together. Both expressions spread the fullintdomain evenly (measured: exactly one value in 2^32 comes out illegal under the old one), so the only behavioural difference is that a key hashing toInteger.MIN_VALUEnow lands on a valid partition instead of failing the write.Risk Level
low
Documentation Update
none
Contributor's checklist