[lake/hudi] Introduce HudiBucketingFunction for bucket strategy

fluss-common/src/main/java/org/apache/fluss/bucketing/BucketingFunction.java

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

fluss-common/src/main/java/org/apache/fluss/bucketing/HudiBucketingFunction.java

@@ -0,0 +1,61 @@
+/*
+ * 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.fluss.bucketing;
+
+/**
+ * An implementation of {@link BucketingFunction} to follow Hudi's bucketing strategy.
+ *
+ * <p>The bucket id is computed in the same way as Hudi's {@code
+ * org.apache.hudi.index.bucket.BucketIdentifier#getBucketId(String, String, int)}: take a 32-bit
+ * integer hash that is encoded as a fixed 4-byte big-endian array by {@code HudiKeyEncoder}, mask
+ * its sign bit and modulo by {@code numBuckets}.
+ */
+public class HudiBucketingFunction implements BucketingFunction {
+
+    /** Length of a Hudi-encoded bucket key, in bytes (a single big-endian {@code int}). */
+    private static final int ENCODED_KEY_LENGTH = 4;
+
+    @Override
+    public int bucketing(byte[] bucketKey, int numBuckets) {
+        if (bucketKey == null) {
+            throw new IllegalArgumentException("bucketKey must not be null");
+        }
+        if (bucketKey.length != ENCODED_KEY_LENGTH) {
+            throw new IllegalArgumentException(
+                    "bucketKey must be exactly "
+                            + ENCODED_KEY_LENGTH
+                            + " bytes for Hudi bucketing, but got "
+                            + bucketKey.length
+                            + " bytes. The bucket key bytes are expected to be produced by HudiKeyEncoder.");
+        }
+        if (numBuckets <= 0) {
+            throw new IllegalArgumentException(
+                    "numBuckets must be positive, but got " + numBuckets);
+        }
+
+        // Decode 4-byte big-endian int produced by HudiKeyEncoder via bit operations
+        // to avoid allocating a ByteBuffer instance on this hot path.
+        int restored =
+                ((bucketKey[0] & 0xFF) << 24)
+                        | ((bucketKey[1] & 0xFF) << 16)
+                        | ((bucketKey[2] & 0xFF) << 8)
+                        | (bucketKey[3] & 0xFF);
+
+        return (restored & Integer.MAX_VALUE) % numBuckets;
+    }
+}

fluss-common/src/main/java/org/apache/fluss/row/encode/KeyEncoder.java

@@ -20,6 +20,7 @@
 import org.apache.fluss.config.TableConfig;
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.hudi.HudiKeyEncoder;
 import org.apache.fluss.row.encode.iceberg.IcebergKeyEncoder;
 import org.apache.fluss.row.encode.paimon.PaimonKeyEncoder;
 import org.apache.fluss.types.RowType;
@@ -76,6 +77,14 @@ static KeyEncoder ofPrimaryKeyEncoder(
         Optional<Integer> optKvFormatVersion = tableConfig.getKvFormatVersion();
         DataLakeFormat dataLakeFormat = tableConfig.getDataLakeFormat().orElse(null);
         int kvFormatVersion = optKvFormatVersion.orElse(1);
+
+        // Hudi's HudiKeyEncoder is lossy (4-byte hash); it must NOT be used for
+        // primary key encoding because different keys with the same List#hashCode
+        // would collide. Use CompactedKeyEncoder instead.
+        if (dataLakeFormat == DataLakeFormat.HUDI) {
+            return CompactedKeyEncoder.createKeyEncoder(rowType, keyFields);
+        }
+
         if (kvFormatVersion == 1) {
             return of(rowType, keyFields, dataLakeFormat);
         }
@@ -129,6 +138,8 @@ static KeyEncoder of(
             return CompactedKeyEncoder.createKeyEncoder(rowType, keyFields);
         } else if (lakeFormat == DataLakeFormat.ICEBERG) {
             return new IcebergKeyEncoder(rowType, keyFields);
+        } else if (lakeFormat == DataLakeFormat.HUDI) {
+            return new HudiKeyEncoder(rowType, keyFields);
         } else {
             throw new UnsupportedOperationException("Unsupported datalake format: " + lakeFormat);
         }

fluss-common/src/main/java/org/apache/fluss/row/encode/hudi/HudiKeyEncoder.java

@@ -0,0 +1,88 @@
+/*
+ * 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.fluss.row.encode.hudi;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.KeyEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.RowType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * An implementation of {@link KeyEncoder} to follow Hudi's encoding strategy.
+ *
+ * <p>The encoded bytes are a 4-byte big-endian representation of {@code List<String>.hashCode()}
+ * over the stringified key fields, which matches the way Hudi's {@code BucketIdentifier} hashes a
+ * record key. Null fields are replaced by {@link #NULL_RECORDKEY_PLACEHOLDER} so that an explicit
+ * null and the literal string {@code "null"} no longer collide in the hash space.
+ */
+public class HudiKeyEncoder implements KeyEncoder {
+
+    /**
+     * Placeholder used to represent a {@code null} key field when computing the record-key hash. It
+     * is intentionally aligned with Hudi's {@code KeyGenUtils.NULL_RECORDKEY_PLACEHOLDER} so that
+     * the resulting bucket id stays identical to what Hudi would compute on its side.
+     */
+    public static final String NULL_RECORDKEY_PLACEHOLDER = "__null__";
+
+    private final InternalRow.FieldGetter[] fieldGetters;
+
+    public HudiKeyEncoder(RowType rowType, List<String> keys) {
+        // for getting key fields out of fluss internal row
+        fieldGetters = new InternalRow.FieldGetter[keys.size()];
+        for (int i = 0; i < keys.size(); i++) {
+            int keyIndex = rowType.getFieldIndex(keys.get(i));
+            DataType keyDataType = rowType.getTypeAt(keyIndex);
+            fieldGetters[i] = InternalRow.createFieldGetter(keyDataType, keyIndex);
+        }
+    }
+
+    @Override
+    public byte[] encodeKey(InternalRow row) {
+        // Build the same string list that Hudi would build out of a record key, so the
+        // resulting List#hashCode() — and therefore the bucket id — match Hudi's own
+        // BucketIdentifier#getBucketId.
+        List<String> values = new ArrayList<>(fieldGetters.length);
+        for (InternalRow.FieldGetter fieldGetter : fieldGetters) {
+            Object value = fieldGetter.getFieldOrNull(row);
+            values.add(stringifyForRecordKey(value));
+        }
+        int hashCode = values.hashCode();
+
+        // 4-byte big-endian, decoded symmetrically by HudiBucketingFunction.
+        return new byte[] {
+            (byte) (hashCode >>> 24),
+            (byte) (hashCode >>> 16),
+            (byte) (hashCode >>> 8),
+            (byte) hashCode
+        };
+    }
+
+    private static String stringifyForRecordKey(Object value) {
+        if (value == null) {
+            return NULL_RECORDKEY_PLACEHOLDER;
+        }
+        if (value instanceof BinaryString) {
+            return value.toString();
+        }
+        return String.valueOf(value);
+    }
+}

fluss-lake/fluss-lake-hudi/pom.xml

@@ -32,6 +32,43 @@
 
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.hudi</groupId>
+            <artifactId>hudi-flink${flink.major.version}-bundle</artifactId>
+            <version>${hudi.version}</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.commons</groupId>
+                    <artifactId>commons-lang3</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-framework</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-recipes</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.zookeeper</groupId>
+                    <artifactId>zookeeper</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.fluss</groupId>
             <artifactId>fluss-common</artifactId>