[iceberg] Introduce iceberg bucketing function (#1409)

Author: gyang94

fluss-common/src/main/java/com/alibaba/fluss/bucketing/BucketingFunction.java

@@ -41,6 +41,8 @@ static BucketingFunction of(@Nullable DataLakeFormat lakeFormat) {
             return new PaimonBucketingFunction();
         } else if (lakeFormat == DataLakeFormat.LANCE) {
             return new FlussBucketingFunction();
+        } else if (lakeFormat == DataLakeFormat.ICEBERG) {
+            return new IcebergBucketingFunction();
         } else {
             throw new UnsupportedOperationException("Unsupported lake format: " + lakeFormat);
         }

fluss-common/src/main/java/com/alibaba/fluss/bucketing/IcebergBucketingFunction.java

@@ -0,0 +1,38 @@
+/*
+ * 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 com.alibaba.fluss.bucketing;
+
+import com.alibaba.fluss.shaded.guava32.com.google.common.hash.HashFunction;
+import com.alibaba.fluss.shaded.guava32.com.google.common.hash.Hashing;
+
+/** An implementation of {@link BucketingFunction} to follow Iceberg's bucketing strategy. */
+public class IcebergBucketingFunction implements BucketingFunction {
+
+    private static final HashFunction MURMUR3 = Hashing.murmur3_32_fixed();
+
+    @Override
+    public int bucketing(byte[] bucketKey, int numBuckets) {
+        if (bucketKey == null || bucketKey.length == 0) {
+            throw new IllegalArgumentException("bucketKey must not be null or empty");
+        }
+        if (numBuckets <= 0) {
+            throw new IllegalArgumentException("numBuckets must be positive");
+        }
+        return (MURMUR3.hashBytes(bucketKey).asInt() & Integer.MAX_VALUE) % numBuckets;
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/row/encode/iceberg/IcebergBinaryRowWriter.java

@@ -27,48 +27,43 @@
 import java.io.Serializable;
 import java.util.Arrays;
 
-import static com.alibaba.fluss.types.DataTypeChecks.getPrecision;
-
 /**
  * A writer to encode Fluss's {@link com.alibaba.fluss.row.InternalRow} using Iceberg's binary
  * encoding format.
  *
- * <p>The encoding logic is based on Iceberg's Conversions.toByteBuffer() implementation:
- * https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/types/Conversions.java
- *
- * <p>Key encoding principles from Iceberg's Conversions class:
+ * <p>The encoding logic is to encode different types of keys into bytes array, which will have the
+ * same hash value with iceberg's bucketing hash function.
  *
  * <ul>
- *   <li>All numeric types (int, long, float, double, timestamps) use LITTLE-ENDIAN byte order
+ *   <li>For int type, it is treated as a long value for encoding
+ *   <li>All numeric types (long, float, double, timestamps) use LITTLE-ENDIAN byte order
  *   <li>Decimal types use BIG-ENDIAN byte order
  *   <li>Strings are encoded as UTF-8 bytes
  *   <li>Timestamps are stored as long values (microseconds since epoch)
  * </ul>
  *
  * <p>Note: This implementation uses Fluss's MemorySegment instead of ByteBuffer for performance,
- * but maintains byte-level compatibility with Iceberg's encoding.
+ * but maintains byte-level compatibility.
  */
 class IcebergBinaryRowWriter {
 
-    private final int arity;
     private byte[] buffer;
     private MemorySegment segment;
     private int cursor;
 
     public IcebergBinaryRowWriter(int arity) {
-        this.arity = arity;
         // Conservative initial size to avoid frequent resizing
         int initialSize = 8 + (arity * 8);
         setBuffer(new byte[initialSize]);
         reset();
     }
 
     public void reset() {
-        this.cursor = 0;
         // Clear only the used portion for efficiency
         if (cursor > 0) {
             Arrays.fill(buffer, 0, Math.min(cursor, buffer.length), (byte) 0);
         }
+        this.cursor = 0;
     }
 
     public byte[] toBytes() {
@@ -77,13 +72,6 @@ public byte[] toBytes() {
         return result;
     }
 
-    public void setNullAt(int pos) {
-        // For Iceberg key encoding, null values should not occur
-        // This is validated at the encoder level
-        throw new UnsupportedOperationException(
-                "Null values are not supported in Iceberg key encoding");
-    }
-
     public void writeBoolean(boolean value) {
         ensureCapacity(1);
         UnsafeUtils.putBoolean(buffer, cursor, value);
@@ -108,6 +96,10 @@ public void writeInt(int value) {
         cursor += 4;
     }
 
+    public void writeIntAsLong(int value) {
+        writeLong(value);
+    }
+
     public void writeLong(long value) {
         ensureCapacity(8);
         UnsafeUtils.putLong(buffer, cursor, value);
@@ -127,26 +119,40 @@ public void writeDouble(double value) {
     }
 
     public void writeString(BinaryString value) {
+        writeString(value, false);
+    }
+
+    public void writeString(BinaryString value, boolean skipEncodeLength) {
         // Convert to UTF-8 byte array
         byte[] bytes = BinaryString.encodeUTF8(value.toString());
-        // Write length prefix followed by UTF-8 bytes
-        writeInt(bytes.length); // 4-byte length prefix
-        ensureCapacity(bytes.length); // Ensure space for actual string bytes
-        segment.put(cursor, bytes, 0, bytes.length);
-        cursor += bytes.length;
+        writeByteArray(bytes, skipEncodeLength);
     }
 
     public void writeBytes(byte[] bytes) {
-        // Write length prefix followed by binary data
-        writeInt(bytes.length); // 4-byte length prefix
+        writeBytes(bytes, false);
+    }
+
+    public void writeBytes(byte[] bytes, boolean skipEncodeLength) {
+        writeByteArray(bytes, skipEncodeLength);
+    }
+
+    private void writeByteArray(byte[] bytes, boolean skipEncodeLength) {
+        if (!skipEncodeLength) {
+            // Write length prefix followed by binary data
+            writeInt(bytes.length); // 4-byte length prefix
+        }
         ensureCapacity(bytes.length); // Ensure space for actual binary bytes
         segment.put(cursor, bytes, 0, bytes.length);
         cursor += bytes.length;
     }
 
-    public void writeDecimal(Decimal value, int precision) {
+    public void writeDecimal(Decimal value) {
+        writeDecimal(value, false);
+    }
+
+    public void writeDecimal(Decimal value, boolean skipEncodeLength) {
         byte[] unscaled = value.toUnscaledBytes();
-        writeBytes(unscaled); // Adds 4-byte length prefix before the actual bytes
+        writeBytes(unscaled, skipEncodeLength); // Adds 4-byte length prefix before the actual bytes
     }
 
     private void ensureCapacity(int neededSize) {
@@ -178,7 +184,7 @@ public static FieldWriter createFieldWriter(DataType fieldType) {
         switch (fieldType.getTypeRoot()) {
             case INTEGER:
             case DATE:
-                return (writer, value) -> writer.writeInt((int) value);
+                return (writer, value) -> writer.writeIntAsLong((int) value);
 
             case TIME_WITHOUT_TIME_ZONE:
                 // Write time as microseconds long (milliseconds * 1000)
@@ -199,16 +205,15 @@ public static FieldWriter createFieldWriter(DataType fieldType) {
                 };
 
             case DECIMAL:
-                final int decimalPrecision = getPrecision(fieldType);
-                return (writer, value) -> writer.writeDecimal((Decimal) value, decimalPrecision);
+                return (writer, value) -> writer.writeDecimal((Decimal) value, true);
 
             case STRING:
             case CHAR:
-                return (writer, value) -> writer.writeString((BinaryString) value);
+                return (writer, value) -> writer.writeString((BinaryString) value, true);
 
             case BINARY:
             case BYTES:
-                return (writer, value) -> writer.writeBytes((byte[]) value);
+                return (writer, value) -> writer.writeBytes((byte[]) value, true);
 
             default:
                 throw new IllegalArgumentException(