[core] Introduce hive bucket function to align with spark hive bucket

docs/layouts/shortcodes/generated/core_configuration.html

@@ -108,7 +108,7 @@
             <td><h5>bucket-function.type</h5></td>
             <td style="word-wrap: break-word;">default</td>
             <td><p>Enum</p></td>
-            <td>The bucket function for paimon bucket.<br /><br />Possible values:<ul><li>"default": The default bucket function which will use arithmetic: bucket_id = Math.abs(hash_bucket_binary_row % numBuckets) to get bucket.</li><li>"mod": The modulus bucket function which will use modulus arithmetic: bucket_id = Math.floorMod(bucket_key_value, numBuckets) to get bucket. Note: the bucket key must be a single field of INT or BIGINT datatype.</li></ul></td>
+            <td>The bucket function for paimon bucket.<br /><br />Possible values:<ul><li>"default": The default bucket function which will use arithmetic: bucket_id = Math.abs(hash_bucket_binary_row % numBuckets) to get bucket.</li><li>"mod": The modulus bucket function which will use modulus arithmetic: bucket_id = Math.floorMod(bucket_key_value, numBuckets) to get bucket. Note: the bucket key must be a single field of INT or BIGINT datatype.</li><li>"hive": The hive bucket function which will use hive-compatible hash arithmetic to get bucket.</li></ul></td>
         </tr>
         <tr>
             <td><h5>bucket-key</h5></td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -152,7 +152,10 @@ public enum BucketFunctionType implements DescribedEnum {
         MOD(
                 "mod",
                 "The modulus bucket function which will use modulus arithmetic: bucket_id = Math.floorMod(bucket_key_value, numBuckets) to get bucket. "
-                        + "Note: the bucket key must be a single field of INT or BIGINT datatype.");
+                        + "Note: the bucket key must be a single field of INT or BIGINT datatype."),
+        HIVE(
+                "hive",
+                "The hive bucket function which will use hive-compatible hash arithmetic to get bucket.");
 
         private final String value;
         private final String description;
@@ -172,6 +175,8 @@ public static BucketFunctionType of(String bucketType) {
                 return DEFAULT;
             } else if (MOD.value.equalsIgnoreCase(bucketType)) {
                 return MOD;
+            } else if (HIVE.value.equalsIgnoreCase(bucketType)) {
+                return HIVE;
             }
             throw new IllegalArgumentException(
                     "cannot match type: " + bucketType + " for bucket function");

paimon-core/src/main/java/org/apache/paimon/bucket/BucketFunction.java

@@ -41,6 +41,8 @@ static BucketFunction create(
                 return new DefaultBucketFunction();
             case MOD:
                 return new ModBucketFunction(bucketKeyType);
+            case HIVE:
+                return new HiveBucketFunction(bucketKeyType);
             default:
                 throw new IllegalArgumentException(
                         "Unsupported bucket type: " + bucketFunctionType);

paimon-core/src/main/java/org/apache/paimon/bucket/HiveBucketFunction.java

@@ -0,0 +1,98 @@
+/*
+ * 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.paimon.bucket;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InternalRowUtils;
+
+/** Hive-compatible bucket function. */
+public class HiveBucketFunction implements BucketFunction {
+
+    private static final long serialVersionUID = 1L;
+
+    private static final int SEED = 0;
+
+    private final InternalRow.FieldGetter[] fieldGetters;
+
+    public HiveBucketFunction(RowType rowType) {
+        this.fieldGetters = InternalRowUtils.createFieldGetters(rowType.getFieldTypes());
+    }
+
+    @Override
+    public int bucket(BinaryRow row, int numBuckets) {
+        assert numBuckets > 0 && row.getRowKind() == RowKind.INSERT : "Num bucket is illegal";
+
+        int hash = SEED;
+        for (int i = 0; i < row.getFieldCount(); i++) {
+            hash = (31 * hash) + computeHash(fieldGetters[i].getFieldOrNull(row));
+        }
+        return mod(hash & Integer.MAX_VALUE, numBuckets);
+    }
+
+    static int mod(int value, int divisor) {
+        int remainder = value % divisor;
+        if (remainder < 0) {
+            return (remainder + divisor) % divisor;
+        }
+        return remainder;
+    }
+
+    private int computeHash(Object value) {
+        if (value == null) {
+            return 0;
+        }
+
+        if (value instanceof Boolean) {
+            return HiveHasher.hashInt((Boolean) value ? 1 : 0);
+        } else if (value instanceof Byte) {
+            return HiveHasher.hashInt(((Byte) value).intValue());
+        } else if (value instanceof Short) {
+            return HiveHasher.hashInt(((Short) value).intValue());
+        } else if (value instanceof Integer) {
+            return HiveHasher.hashInt((Integer) value);
+        } else if (value instanceof Long) {
+            return HiveHasher.hashLong((Long) value);
+        } else if (value instanceof Float) {
+            float floatValue = (Float) value;
+            return HiveHasher.hashInt(floatValue == -0.0f ? 0 : Float.floatToIntBits(floatValue));
+        } else if (value instanceof Double) {
+            double doubleValue = (Double) value;
+            return HiveHasher.hashLong(
+                    doubleValue == -0.0d ? 0L : Double.doubleToLongBits(doubleValue));
+        } else if (value instanceof BinaryString) {
+            BinaryString stringValue = (BinaryString) value;
+            return HiveHasher.hashUnsafeBytes(
+                    stringValue.getSegments(),
+                    stringValue.getOffset(),
+                    stringValue.getSizeInBytes());
+        } else if (value instanceof byte[]) {
+            return HiveHasher.hashBytes((byte[]) value);
+        } else if (value instanceof Decimal) {
+            return HiveHasher.normalizeDecimal(((Decimal) value).toBigDecimal()).hashCode();
+        }
+
+        throw new UnsupportedOperationException(
+                "Unsupported type as bucket key type " + value.getClass());
+    }
+}

paimon-core/src/main/java/org/apache/paimon/bucket/HiveHasher.java

@@ -0,0 +1,114 @@
+/*
+ * 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.paimon.bucket;
+
+import org.apache.paimon.memory.MemorySegment;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+
+/** Hive hash util. */
+public class HiveHasher {
+
+    private static final int HIVE_DECIMAL_MAX_PRECISION = 38;
+    private static final int HIVE_DECIMAL_MAX_SCALE = 38;
+
+    @Override
+    public String toString() {
+        return HiveHasher.class.getSimpleName();
+    }
+
+    public static int hashInt(int input) {
+        return input;
+    }
+
+    public static int hashLong(long input) {
+        return Long.hashCode(input);
+    }
+
+    public static int hashBytes(byte[] bytes) {
+        int result = 0;
+        for (byte value : bytes) {
+            result = (result * 31) + value;
+        }
+        return result;
+    }
+
+    public static int hashUnsafeBytes(MemorySegment[] segments, int offset, int length) {
+        int result = 0;
+        for (MemorySegment segment : segments) {
+            int remaining = segment.size() - offset;
+            if (remaining > 0) {
+                int bytesToRead = Math.min(remaining, length);
+                for (int i = 0; i < bytesToRead; i++) {
+                    result = (result * 31) + segment.get(offset + i);
+                }
+                length -= bytesToRead;
+                offset = 0;
+            } else {
+                offset -= segment.size();
+            }
+
+            if (length == 0) {
+                break;
+            }
+        }
+        return result;
+    }
+
+    public static BigDecimal normalizeDecimal(BigDecimal input) {
+        if (input == null) {
+            return null;
+        }
+
+        BigDecimal result = trimDecimal(input);
+        int intDigits = result.precision() - result.scale();
+        if (intDigits > HIVE_DECIMAL_MAX_PRECISION) {
+            return null;
+        }
+
+        int maxScale =
+                Math.min(
+                        HIVE_DECIMAL_MAX_SCALE,
+                        Math.min(HIVE_DECIMAL_MAX_PRECISION - intDigits, result.scale()));
+        if (result.scale() > maxScale) {
+            result = result.setScale(maxScale, RoundingMode.HALF_UP);
+            result = trimDecimal(result);
+        }
+
+        return result;
+    }
+
+    private static BigDecimal trimDecimal(BigDecimal input) {
+        if (input.compareTo(BigDecimal.ZERO) == 0) {
+            return BigDecimal.ZERO;
+        }
+
+        BigDecimal result = input.stripTrailingZeros();
+        if (result.compareTo(BigDecimal.ZERO) == 0) {
+            return BigDecimal.ZERO;
+        }
+
+        if (result.scale() < 0) {
+            result = result.setScale(0);
+        }
+
+        return result;
+    }
+}

paimon-core/src/test/java/org/apache/paimon/bucket/HiveBucketFunctionTest.java

@@ -0,0 +1,105 @@
+/*
+ * 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.paimon.bucket;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HiveBucketFunction}. */
+class HiveBucketFunctionTest {
+
+    @Test
+    void testHiveBucketFunction() {
+        RowType rowType =
+                RowType.of(
+                        DataTypes.INT(),
+                        DataTypes.STRING(),
+                        DataTypes.BYTES(),
+                        DataTypes.DECIMAL(10, 4));
+        HiveBucketFunction hiveBucketFunction = new HiveBucketFunction(rowType);
+
+        BinaryRow row =
+                toBinaryRow(
+                        rowType,
+                        7,
+                        BinaryString.fromString("hello"),
+                        new byte[] {1, 2, 3},
+                        Decimal.fromBigDecimal(new BigDecimal("12.3400"), 10, 4));
+
+        int expectedHash =
+                31
+                                * (31
+                                                * (31 * 7
+                                                        + HiveHasher.hashBytes(
+                                                                "hello"
+                                                                        .getBytes(
+                                                                                StandardCharsets
+                                                                                        .UTF_8)))
+                                        + HiveHasher.hashBytes(new byte[] {1, 2, 3}))
+                        + new BigDecimal("12.34").hashCode();
+        assertThat(hiveBucketFunction.bucket(row, 8))
+                .isEqualTo((expectedHash & Integer.MAX_VALUE) % 8);
+    }
+
+    @Test
+    void testHiveBucketFunctionWithNulls() {
+        RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
+        HiveBucketFunction hiveBucketFunction = new HiveBucketFunction(rowType);
+
+        BinaryRow row = toBinaryRow(rowType, null, null);
+
+        assertThat(hiveBucketFunction.bucket(row, 4)).isZero();
+    }
+
+    @Test
+    void testHiveBucketFunctionUnsupportedType() {
+        RowType rowType = RowType.of(DataTypes.TIMESTAMP());
+        HiveBucketFunction hiveBucketFunction = new HiveBucketFunction(rowType);
+
+        assertThat(hiveBucketFunction.bucket(toBinaryRow(rowType, (Object) null), 4)).isZero();
+
+        assertThatThrownBy(
+                        () ->
+                                hiveBucketFunction.bucket(
+                                        toBinaryRow(
+                                                rowType,
+                                                org.apache.paimon.data.Timestamp.fromEpochMillis(
+                                                        1L)),
+                                        4))
+                .isInstanceOf(UnsupportedOperationException.class)
+                .hasMessageContaining("Unsupported type as bucket key type");
+    }
+
+    private BinaryRow toBinaryRow(RowType rowType, Object... values) {
+        return new InternalRowSerializer(rowType).toBinaryRow(GenericRow.of(values));
+    }
+}