[MINOR] Do not return a negative Spark partition when a hash is Integer.MIN_VALUE

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);
     }
   }
 }

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);
   }
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestCoalescingPartitioner.java

@@ -24,6 +24,7 @@
 import org.apache.hudi.data.HoodieJavaRDD;
 import org.apache.hudi.testutils.HoodieClientTestBase;
 
+import org.apache.spark.HashPartitioner;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.function.FlatMapFunction;
 import org.apache.spark.api.java.function.Function;
@@ -184,4 +185,41 @@ public WriteStatus call(Tuple2<String, WriteStatus> booleanIntegerTuple2) throws
       return booleanIntegerTuple2._2;
     }
   }
+
+  /**
+   * Spark's own HashPartitioner routes with Utils.nonNegativeMod, which is floorMod. Pinning the
+   * exact index against it is a real oracle: asserting only that the index is in range would pass
+   * for abs-mod too, and abs-mod differs from floorMod for roughly half of all negative hashes.
+   */
+  @Test
+  public void testPartitionMatchesSparkHashPartitioner() {
+    String minValueHashKey = "polygenelubricants";
+    assertEquals(Integer.MIN_VALUE, minValueHashKey.hashCode());
+    // Integer.MIN_VALUE % 2^k == 0, so only 3, 5, 6 and 7 fail on the old Math.abs expression;
+    // trimming this list to powers of two would stop it exercising the fix.
+    for (int numPartitions : new int[] {1, 2, 3, 4, 5, 6, 7, 8, 16}) {
+      HashPartitioner oracle = new HashPartitioner(numPartitions);
+      for (Object key : new Object[] {minValueHashKey, -1, -2, -3, -5, -100, 0, 1, 100}) {
+        int partition = new CoalescingPartitioner(numPartitions).getPartition(key);
+        assertTrue(partition >= 0 && partition < numPartitions,
+            "partition " + partition + " out of range for numPartitions " + numPartitions);
+        assertEquals(oracle.getPartition(key), partition,
+            "key " + key + " at numPartitions " + numPartitions);
+      }
+    }
+  }
+
+  /**
+   * The assertions above only call getPartition. This drives a real shuffle so the failure the old
+   * expression produced is visible end to end: BypassMergeSortShuffleWriter indexes its
+   * partitionWriters array with whatever getPartition answers, unguarded.
+   */
+  @Test
+  public void testShuffleSucceedsForMinValueHashKey() {
+    JavaRDD<String> keys = jsc.parallelize(Collections.singletonList("polygenelubricants"), 1);
+    assertEquals(1, keys.mapToPair(key -> new Tuple2<>(key, key))
+        .partitionBy(new CoalescingPartitioner(3))
+        .count());
+  }
+
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/execution/bulkinsert/TestPartitionPathRDDPartitioner.java

@@ -0,0 +1,54 @@
+/*
+ * 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.apache.spark.HashPartitioner;
+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. The partitioner hashes that sum rather than the string, so the oracle is
+   * fed the boxed int. Spark's HashPartitioner routes with Utils.nonNegativeMod, i.e. floorMod.
+   */
+  // Integer.MIN_VALUE % 2^k == 0, so only 3, 5, 6 and 7 fail on the old Math.abs expression;
+  // trimming this list to powers of two would stop it exercising the fix.
+  @ParameterizedTest
+  @ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 16})
+  void testPartitionMatchesSparkHashPartitioner(int numPartitions) {
+    String minValueHashPath = "xfjfxsf";
+    assertEquals(Integer.MIN_VALUE, Objects.hash(minValueHashPath));
+
+    PartitionPathRDDPartitioner partitioner =
+        new PartitionPathRDDPartitioner(o -> minValueHashPath, 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(minValueHashPath)),
+        partition, "numPartitions " + numPartitions);
+  }
+
+}