voonhous commented on code in PR #19776:
URL: https://github.com/apache/hudi/pull/19776#discussion_r3879230503
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/CoalescingPartitioner.java:
##########
@@ -41,7 +41,9 @@ public int getPartition(Object key) {
if (numPartitions == 1) {
return 0;
} else {
- 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);
Review 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.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/execution/bulkinsert/PartitionPathRDDPartitioner.java:
##########
@@ -47,6 +47,8 @@ public int numPartitions() {
@SuppressWarnings("unchecked")
@Override
public int getPartition(Object o) {
- 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);
Review 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.
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestPartitionPathRDDPartitioner.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Objects;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestPartitionPathRDDPartitioner {
Review 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.
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestCoalescingPartitioner.java:
##########
@@ -184,4 +184,20 @@ public WriteStatus call(Tuple2<String, WriteStatus>
booleanIntegerTuple2) throws
return booleanIntegerTuple2._2;
}
}
+
+ /**
+ * "polygenelubricants" hashes to Integer.MIN_VALUE, which Math.abs leaves
negative, so a
+ * partitioner deriving its answer that way returns an index Spark cannot
use. Asserting the
+ * fixture first, so this stops silently passing if String.hashCode ever
changes.
+ */
+ @Test
+ public void testPartitionIsInRangeForMinValueHash() {
+ String key = "polygenelubricants";
+ assertEquals(Integer.MIN_VALUE, key.hashCode());
+ 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);
+ }
Review 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;`.
```suggestion
// 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);
}
```
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestPartitionPathRDDPartitioner.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Objects;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestPartitionPathRDDPartitioner {
+
+ /**
+ * Objects.hash(x) is 31 + x.hashCode(), so this partition path overflows it
to
+ * Integer.MIN_VALUE, which Math.abs leaves negative.
+ */
+ private static final String MIN_VALUE_HASH_PATH = "xfjfxsf";
+
+ @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 =
+ new PartitionPathRDDPartitioner(o -> MIN_VALUE_HASH_PATH,
numPartitions);
+ int partition = partitioner.getPartition(new Object());
+ assertTrue(partition >= 0 && partition < numPartitions,
+ "partition " + partition + " out of range for numPartitions " +
numPartitions);
Review 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;`.
```suggestion
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);
```
##########
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestPartitionPathRDDPartitioner.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.execution.bulkinsert;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Objects;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestPartitionPathRDDPartitioner {
+
+ /**
+ * Objects.hash(x) is 31 + x.hashCode(), so this partition path overflows it
to
+ * Integer.MIN_VALUE, which Math.abs leaves negative.
+ */
+ private static final String MIN_VALUE_HASH_PATH = "xfjfxsf";
+
+ @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 =
Review 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).
```suggestion
@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 =
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
