[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 1 commit 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.
| return Math.abs(key.hashCode()) % numPartitions; | ||
| // Math.abs leaves Integer.MIN_VALUE negative, and a Partitioner must answer in | ||
| // [0, numPartitions). floorMod is non-negative for every input. | ||
| return Math.floorMod(key.hashCode(), numPartitions); |
There was a problem hiding this comment.
The code is right, but the PR body and commit message claim floorMod "agrees with the old expression on every hash the old one already handled correctly" and that other keys "route exactly as before at power-of-two parallelism". Every negative hash reroutes at every parallelism: Math.abs(-1) % 3 == 1 vs Math.floorMod(-1, 3) == 2; Math.abs(-1) % 4 == 1 vs Math.floorMod(-1, 4) == 3.
Harmless here: the only caller (SparkStreamingMetadataWriteHandler:63) drops the partitioner with .map(entry -> entry._2), and floorMod is exactly Spark's HashPartitioner (Utils.nonNegativeMod), already used by UpsertPartitioner:358 and BucketizedBloomCheckPartitioner:176.
Please reword those two sentences to: negative hashes now route to a different but valid partition, matching Spark's HashPartitioner; positive hashes are unchanged. Keep floorMod.
| return Math.abs(Objects.hash(partitionPathExtractor.apply(o))) % numPartitions; | ||
| // Math.abs leaves Integer.MIN_VALUE negative, and a Partitioner must answer in | ||
| // [0, numPartitions). floorMod is non-negative for every input. | ||
| return Math.floorMod(Objects.hash(partitionPathExtractor.apply(o)), numPartitions); |
There was a problem hiding this comment.
Same routing change applies here for every negative Objects.hash, not just MIN_VALUE. Also fine: PartitionPathRepartitionPartitioner:66, PartitionPathRepartitionAndSortPartitioner:67 and LSMPartitionPathRepartitionAndSortPartitioner:75 all end in .values() and only rely on one partition path landing in one Spark partition, which any deterministic function keeps. Worth one sentence in the Impact section so the next reader does not have to re-derive it.
| for (int numPartitions : new int[] {1, 2, 3, 4, 5, 6, 7, 8, 16}) { | ||
| int partition = new CoalescingPartitioner(numPartitions).getPartition(key); | ||
| assertTrue(partition >= 0 && partition < numPartitions, | ||
| "partition " + partition + " out of range for numPartitions " + numPartitions); | ||
| } |
There was a problem hiding this comment.
Range-only leaves the routing unpinned: simpleCoalescingPartitionerTest uses Integer keys 0..100, so no test sees a negative hash, which is exactly where abs-mod and floorMod differ. Asserting equality with Spark's own HashPartitioner is a real oracle (it is Utils.nonNegativeMod, i.e. floorMod) and makes the "matches Spark" statement testable. Also, 1/2/4/8/16 cannot fail on the old code (Integer.MIN_VALUE % 2^k == 0, and 1 short-circuits before the modulo), so a comment keeps someone from trimming the list to powers of two.
Needs import org.apache.spark.HashPartitioner;.
| for (int numPartitions : new int[] {1, 2, 3, 4, 5, 6, 7, 8, 16}) { | |
| int partition = new CoalescingPartitioner(numPartitions).getPartition(key); | |
| assertTrue(partition >= 0 && partition < numPartitions, | |
| "partition " + partition + " out of range for numPartitions " + numPartitions); | |
| } | |
| // Integer.MIN_VALUE % 2^k == 0, so only 3, 5, 6 and 7 fail on the old Math.abs expression. | |
| for (int numPartitions : new int[] {1, 2, 3, 4, 5, 6, 7, 8, 16}) { | |
| int partition = new CoalescingPartitioner(numPartitions).getPartition(key); | |
| assertTrue(partition >= 0 && partition < numPartitions, | |
| "partition " + partition + " out of range for numPartitions " + numPartitions); | |
| assertEquals(new HashPartitioner(numPartitions).getPartition(key), partition); | |
| } |
| int partition = partitioner.getPartition(new Object()); | ||
| assertTrue(partition >= 0 && partition < numPartitions, | ||
| "partition " + partition + " out of range for numPartitions " + numPartitions); |
There was a problem hiding this comment.
Same as the Coalescing test: pin the exact index against Spark's HashPartitioner rather than only the range. This partitioner hashes Objects.hash(path) (31 + path.hashCode()), not the string itself, so feed Spark the boxed int. Needs import org.apache.spark.HashPartitioner;.
| int partition = partitioner.getPartition(new Object()); | |
| assertTrue(partition >= 0 && partition < numPartitions, | |
| "partition " + partition + " out of range for numPartitions " + numPartitions); | |
| int partition = partitioner.getPartition(new Object()); | |
| assertTrue(partition >= 0 && partition < numPartitions, | |
| "partition " + partition + " out of range for numPartitions " + numPartitions); | |
| assertEquals(new HashPartitioner(numPartitions).getPartition(Objects.hash(MIN_VALUE_HASH_PATH)), partition); |
| @Test | ||
| void assertFixtureStillOverflowsToMinValue() { | ||
| assertEquals(Integer.MIN_VALUE, Objects.hash(MIN_VALUE_HASH_PATH)); | ||
| } | ||
|
|
||
| @ParameterizedTest | ||
| @ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 16}) | ||
| void assertPartitionIsInRangeForMinValueHash(int numPartitions) { | ||
| PartitionPathRDDPartitioner partitioner = |
There was a problem hiding this comment.
nit, feel free to ignore: TestCoalescingPartitioner asserts the fixture inline as the first line of the test. Doing the same here drops a method and the now-unused org.junit.jupiter.api.Test import (checkstyle will flag it if left behind).
| @Test | |
| void assertFixtureStillOverflowsToMinValue() { | |
| assertEquals(Integer.MIN_VALUE, Objects.hash(MIN_VALUE_HASH_PATH)); | |
| } | |
| @ParameterizedTest | |
| @ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 16}) | |
| void assertPartitionIsInRangeForMinValueHash(int numPartitions) { | |
| PartitionPathRDDPartitioner partitioner = | |
| @ParameterizedTest | |
| @ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 16}) | |
| void assertPartitionIsInRangeForMinValueHash(int numPartitions) { | |
| assertEquals(Integer.MIN_VALUE, Objects.hash(MIN_VALUE_HASH_PATH)); | |
| PartitionPathRDDPartitioner partitioner = |
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| class TestPartitionPathRDDPartitioner { |
There was a problem hiding this comment.
Optional: every end-to-end harness for these sort modes uses power-of-two parallelism (TestBulkInsertInternalPartitioner:146 hardcodes 2, TestLSMBulkInsertPartitioner uses 1 and 4), and Integer.MIN_VALUE % 2^k == 0, so none of them could ever have caught this. If you want proof that Spark actually throws on the old code, one mapToPair(...).partitionBy(new PartitionPathRDDPartitioner(o -> "xfjfxsf", 3)) over a small RDD in a HoodieClientTestBase-derived test does it: BypassMergeSortShuffleWriter indexes partitionWriters with the result, unguarded. The getPartition asserts at 3/5/6/7 already discriminate, so not blocking on this.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19776 +/- ##
============================================
- Coverage 78.11% 75.44% -2.67%
+ Complexity 33673 32526 -1147
============================================
Files 2540 2540
Lines 141413 141413
Branches 17123 17123
============================================
- Hits 110467 106695 -3772
- Misses 23250 26853 +3603
- Partials 7696 7865 +169
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
4 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 and agrees with the old expression on every hash the old one already handled correctly — only theInteger.MIN_VALUEcase changes, and there the old answer was not a usable partition index.BucketIndexUtil((partition.hashCode() & Integer.MAX_VALUE) % parallelism) andJavaUpsertPartitioner(Math.floorMod) already avoidMath.absfor the same reason.Tests:
TestCoalescingPartitioner#testPartitionIsInRangeForMinValueHash— added to the existing class; asserts the fixture still hashes toInteger.MIN_VALUEfirst, so it cannot silently stop exercising the case, then checks the index is in range for 1..16 partitions.TestPartitionPathRDDPartitioner— new, same shape, assertingObjects.hashstill overflows for the fixture.Reverting the change turns them red with
partition -2 out of range for numPartitions 3and the equivalent for 5, 6 and 7; the power-of-two parallelisms stay green either way, which is why this has not been hit before.mvn test -pl hudi-client/hudi-spark-client -Dtest='*Partitioner*'— 59 tests, all passing.checkstyle:checkclean.Impact
No public API or config change. Records whose key hashes to
Integer.MIN_VALUEnow land on a valid partition instead of failing the write; every other key routes exactly as before at power-of-two parallelism, and to a different but valid partition otherwise.Risk Level
low
Documentation Update
none
Contributor's checklist