(iceberg) branch main updated: Flink: Fix RANGE distribution ignoring user-specified equality fields (#17276)

[mxm]

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

mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new d8c10a1608 Flink: Fix RANGE distribution ignoring user-specified 
equality fields (#17276)
d8c10a1608 is described below

commit d8c10a1608170f0ba83be740d6ab0b6a3757cb3e
Author: Eunbin Son <58901024+thswls...@users.noreply.github.com>
AuthorDate: Fri Jul 31 14:20:28 2026 +0900

    Flink: Fix RANGE distribution ignoring user-specified equality fields 
(#17276)
---
 .../flink/sink/dynamic/HashKeyGenerator.java       |  2 +-
 .../flink/sink/dynamic/TestHashKeyGenerator.java   | 71 ++++++++++++++++++++++
 2 files changed, 72 insertions(+), 1 deletion(-)

diff --git 
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
 
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
index 03541ed596..0fd6bed9e1 100644
--- 
a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
+++ 
b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java
@@ -173,7 +173,7 @@ class HashKeyGenerator {
         }
 
       case RANGE:
-        if (schema.identifierFieldIds().isEmpty()) {
+        if (equalityFields.isEmpty()) {
           LOG.warn(
               "{}: Fallback to use 'none' distribution mode, because there are 
no equality fields set "
                   + "and {}='range' is not supported yet in flink",
diff --git 
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
 
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
index 4181f00d0a..7a50ca4356 100644
--- 
a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
+++ 
b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestHashKeyGenerator.java
@@ -164,6 +164,77 @@ class TestHashKeyGenerator {
     assertThat(getSubTaskId(writeKey3, writeParallelism, 
maxWriteParallelism)).isEqualTo(0);
   }
 
+  @Test
+  void testRangeModeWithEqualityFields() throws Exception {
+    int writeParallelism = 2;
+    int maxWriteParallelism = 8;
+    HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+    PartitionSpec unpartitioned = PartitionSpec.unpartitioned();
+
+    GenericRowData row1 = GenericRowData.of(1, StringData.fromString("foo"));
+    GenericRowData row2 = GenericRowData.of(1, StringData.fromString("bar"));
+    GenericRowData row3 = GenericRowData.of(2, StringData.fromString("baz"));
+    // SCHEMA has no identifier fields, so the equality fields only come from 
the record
+    Set<String> equalityColumns = Collections.singleton("id");
+
+    int writeKey1 =
+        getWriteKey(
+            generator,
+            unpartitioned,
+            DistributionMode.RANGE,
+            writeParallelism,
+            equalityColumns,
+            row1);
+    int writeKey2 =
+        getWriteKey(
+            generator,
+            unpartitioned,
+            DistributionMode.RANGE,
+            writeParallelism,
+            equalityColumns,
+            row2);
+    int writeKey3 =
+        getWriteKey(
+            generator,
+            unpartitioned,
+            DistributionMode.RANGE,
+            writeParallelism,
+            equalityColumns,
+            row3);
+
+    assertThat(writeKey1).isEqualTo(writeKey2);
+    assertThat(writeKey2).isNotEqualTo(writeKey3);
+  }
+
+  @Test
+  void testRangeModeFallsBackToDistributionModeNone() throws Exception {
+    int writeParallelism = 2;
+    int maxWriteParallelism = 8;
+    HashKeyGenerator generator = new HashKeyGenerator(16, maxWriteParallelism);
+    Schema noIdSchema = new Schema(Types.NestedField.required(1, "x", 
Types.StringType.get()));
+    PartitionSpec unpartitioned = PartitionSpec.unpartitioned();
+
+    DynamicRecord record =
+        new DynamicRecord(
+            TABLE_IDENTIFIER,
+            BRANCH,
+            noIdSchema,
+            GenericRowData.of(StringData.fromString("v")),
+            unpartitioned,
+            DistributionMode.RANGE,
+            writeParallelism);
+
+    int writeKey1 = generator.generateKey(record);
+    int writeKey2 = generator.generateKey(record);
+    int writeKey3 = generator.generateKey(record);
+    assertThat(writeKey1).isNotEqualTo(writeKey2);
+    assertThat(writeKey3).isEqualTo(writeKey1);
+
+    assertThat(getSubTaskId(writeKey1, writeParallelism, 
maxWriteParallelism)).isEqualTo(1);
+    assertThat(getSubTaskId(writeKey2, writeParallelism, 
maxWriteParallelism)).isEqualTo(0);
+    assertThat(getSubTaskId(writeKey3, writeParallelism, 
maxWriteParallelism)).isEqualTo(1);
+  }
+
   @Test
   void testHashModeWithPartitionFieldAndEqualityField() throws Exception {
     int writeParallelism = 2;