(paimon) branch master updated: [flink] Do not shuffle to a negative channel when the partition hash is Integer.MIN_VALUE (#9408)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 1611369de9 [flink] Do not shuffle to a negative channel when the 
partition hash is Integer.MIN_VALUE (#9408)
1611369de9 is described below

commit 1611369de95828d0f71bfa816dccf5ae638fe9b1
Author: ZIHAN DAI <99155080+pd...@users.noreply.github.com>
AuthorDate: Wed Aug 26 19:30:16 2026 +1000

    [flink] Do not shuffle to a negative channel when the partition hash is 
Integer.MIN_VALUE (#9408)
---
 .../StatisticsOrRecordChannelComputer.java         |  2 +-
 .../StatisticsOrRecordChannelComputerTest.java     | 23 ++++++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputer.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputer.java
index 8a19f95e0e..0cb5f7c950 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputer.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputer.java
@@ -186,7 +186,7 @@ public class StatisticsOrRecordChannelComputer implements 
ChannelComputer<Statis
             if (assignment == null) {
                 int defaultSubtaskCount =
                         Math.min(numChannels, 
DEFAULT_SUBTASK_COUNT_FOR_UNKNOWN_PARTITION);
-                int startChannel = Math.abs(partitionKey.hashCode()) % 
numChannels;
+                int startChannel = ChannelComputer.startChannel(partitionKey, 
numChannels);
                 List<Integer> subtasks = new ArrayList<>(defaultSubtaskCount);
                 List<Long> weights = new ArrayList<>(defaultSubtaskCount);
                 for (int i = 0; i < defaultSubtaskCount; i++) {
diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputerTest.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputerTest.java
index deedae4c29..ef4f617b82 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputerTest.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/partition/StatisticsOrRecordChannelComputerTest.java
@@ -226,6 +226,29 @@ class StatisticsOrRecordChannelComputerTest {
         assertThat(assignment).containsKey(p2);
     }
 
+    @Test
+    void testUnknownPartitionWhoseKeyHashesToMinValue() {
+        // Math.abs leaves Integer.MIN_VALUE negative. Pin the fixture: if the 
row hash ever
+        // changes this stops testing anything, and this assertion says so 
instead of passing.
+        InternalRow row =
+                GenericRow.of(
+                        0, BinaryString.fromString("p3588823253"), 
BinaryString.fromString("d"));
+        
assertThat(getPartitionKey(row).hashCode()).isEqualTo(Integer.MIN_VALUE);
+
+        // Integer.MIN_VALUE % n is only zero when n divides 2^31, so a 
parallelism that is not a
+        // power of two is what turns the negative hash into a negative 
channel.
+        for (int downstreamParallelism = 1; downstreamParallelism <= 16; 
downstreamParallelism++) {
+            StatisticsOrRecordChannelComputer channelComputer =
+                    new StatisticsOrRecordChannelComputer(schema);
+            channelComputer.setup(downstreamParallelism);
+            for (int i = 0; i < 50; i++) {
+                
assertThat(channelComputer.channel(StatisticsOrRecord.fromRecord(row)))
+                        .as("parallelism %d", downstreamParallelism)
+                        .isBetween(0, downstreamParallelism - 1);
+            }
+        }
+    }
+
     private BinaryRow getPartitionKey(InternalRow row) {
         RowPartitionKeyExtractor extractor = new 
RowPartitionKeyExtractor(schema);
         return extractor.partition(row).copy();