(iceberg) branch main updated: Flink: Fix range distribution with truncate on binary columns (#18696)

[pvary.apache.org via commits]

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

pvary 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 04dd1750b9 Flink: Fix range distribution with truncate on binary 
columns (#18696)
04dd1750b9 is described below

commit 04dd1750b9fed9f91baf6cd60bf798475b5a9dbb
Author: chenjian2664 <chenjian2...@gmail.com>
AuthorDate: Tue Nov 3 19:33:32 2026 +0800

    Flink: Fix range distribution with truncate on binary columns (#18696)
---
 .../java/org/apache/iceberg/flink/sink/shuffle/SortKeySerializer.java | 3 ++-
 .../iceberg/flink/sink/shuffle/TestSortKeySerializerPrimitives.java   | 4 +---
 2 files changed, 3 insertions(+), 4 deletions(-)

diff --git 
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SortKeySerializer.java
 
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SortKeySerializer.java
index ec099fbfe8..79af27dac9 100644
--- 
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SortKeySerializer.java
+++ 
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SortKeySerializer.java
@@ -45,6 +45,7 @@ import org.apache.iceberg.types.CheckCompatibility;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ByteBuffers;
 
 class SortKeySerializer extends TypeSerializer<SortKey> {
   private final Schema schema;
@@ -178,7 +179,7 @@ class SortKeySerializer extends TypeSerializer<SortKey> {
           break;
         case FIXED:
         case BINARY:
-          byte[] bytes = record.get(i, ByteBuffer.class).array();
+          byte[] bytes = ByteBuffers.toByteArray(record.get(i, 
ByteBuffer.class));
           target.writeInt(bytes.length);
           target.write(bytes);
           break;
diff --git 
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSortKeySerializerPrimitives.java
 
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSortKeySerializerPrimitives.java
index 44791094de..56d56ce400 100644
--- 
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSortKeySerializerPrimitives.java
+++ 
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestSortKeySerializerPrimitives.java
@@ -57,9 +57,7 @@ public class TestSortKeySerializerPrimitives extends 
TestSortKeySerializerBase {
         .asc("ts_ns_with_zone_field")
         .sortBy(
             Expressions.hour("ts_ns_without_zone_field"), SortDirection.ASC, 
NullOrder.NULLS_FIRST)
-        // can not test HeapByteBuffer due to equality test inside 
SerializerTestBase
-        // .sortBy(Expressions.truncate("binary_field", 2), SortDirection.ASC,
-        // NullOrder.NULLS_FIRST)
+        .sortBy(Expressions.truncate("binary_field", 2), SortDirection.ASC, 
NullOrder.NULLS_FIRST)
         .build();
   }