This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new c89ba106a [lake/hudi] Introduce HudiBucketingFunction for bucket
strategy (#3316)
c89ba106a is described below
commit c89ba106a45e721c8def7e2344da9f35f5ea043a
Author: fhan <[email protected]>
AuthorDate: Fri May 22 16:33:36 2026 +0800
[lake/hudi] Introduce HudiBucketingFunction for bucket strategy (#3316)
---
.../apache/fluss/bucketing/BucketingFunction.java | 2 +
.../fluss/bucketing/HudiBucketingFunction.java | 61 +++
.../org/apache/fluss/row/encode/KeyEncoder.java | 11 +
.../fluss/row/encode/hudi/HudiKeyEncoder.java | 238 +++++++++
fluss-lake/fluss-lake-hudi/pom.xml | 37 ++
.../fluss/bucketing/HudiBucketingFunctionTest.java | 586 +++++++++++++++++++++
6 files changed, 935 insertions(+)
diff --git
a/fluss-common/src/main/java/org/apache/fluss/bucketing/BucketingFunction.java
b/fluss-common/src/main/java/org/apache/fluss/bucketing/BucketingFunction.java
index 915fbd172..5291ccdc3 100644
---
a/fluss-common/src/main/java/org/apache/fluss/bucketing/BucketingFunction.java
+++
b/fluss-common/src/main/java/org/apache/fluss/bucketing/BucketingFunction.java
@@ -43,6 +43,8 @@ public interface BucketingFunction {
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);
}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/bucketing/HudiBucketingFunction.java
b/fluss-common/src/main/java/org/apache/fluss/bucketing/HudiBucketingFunction.java
new file mode 100644
index 000000000..0799d64b5
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/row/encode/KeyEncoder.java
b/fluss-common/src/main/java/org/apache/fluss/row/encode/KeyEncoder.java
index 4d59be33b..fe52e926c 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/encode/KeyEncoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/encode/KeyEncoder.java
@@ -20,6 +20,7 @@ package org.apache.fluss.row.encode;
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 @@ public interface KeyEncoder {
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 @@ public interface KeyEncoder {
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);
}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/row/encode/hudi/HudiKeyEncoder.java
b/fluss-common/src/main/java/org/apache/fluss/row/encode/hudi/HudiKeyEncoder.java
new file mode 100644
index 000000000..e0518b548
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/row/encode/hudi/HudiKeyEncoder.java
@@ -0,0 +1,238 @@
+/*
+ * 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.DataTypeRoot;
+import org.apache.fluss.types.RowType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * An implementation of {@link KeyEncoder} that follows Hudi's bucket-key
encoding strategy.
+ *
+ * <h3>What this encoder must reproduce</h3>
+ *
+ * <p>The bucket id Hudi computes in production is:
+ *
+ * <pre>{@code
+ * BucketIdentifier.getBucketId(HoodieKey, indexKeyFields, numBuckets)
+ * = (KeyGenUtils.extractRecordKeysByFields(recordKey, fields).hashCode()
+ * & Integer.MAX_VALUE) % numBuckets
+ * }</pre>
+ *
+ * <p>In other words, Hudi parses a {@code "f1:v1,f2:v2,..."} record-key
string back into a {@code
+ * List<String>} (with {@code "__null__"} → {@code null} and {@code
"__empty__"} → {@code ""}) and
+ * then takes {@link java.util.List#hashCode()}.
+ *
+ * <p>This encoder produces a list <em>element-wise equivalent</em> to the one
Hudi reconstructs,
+ * and hashes it with the same {@code List#hashCode()} contract. The encoded
bytes are a 4-byte
+ * big-endian representation of that hash, decoded symmetrically by {@code
HudiBucketingFunction}.
+ *
+ * <h3>Why a list of nullable {@code String}s (and not stringified
placeholders)</h3>
+ *
+ * <p>Hudi's parser turns the literal {@code "__null__"}/{@code "__empty__"}
substrings back into
+ * {@code null}/{@code ""} <em>before</em> hashing. {@code List#hashCode}
treats a {@code null}
+ * element as contributing {@code 0} and a non-null element as contributing
its {@code String} hash.
+ * Hashing the placeholder string directly would therefore give a different
bucket id than Hudi for
+ * any row with a null bucket-key field.
+ *
+ * <h3>Why we reject values containing {@code ','}</h3>
+ *
+ * <p>{@code ','} is Hudi's record-part separator and is <strong>not
escaped</strong> in Hudi's
+ * record-key serialization. Any bucket-key value containing {@code ','} is
therefore ambiguous on
+ * Hudi's parse path and would produce a different reconstructed list than the
one we hash here. We
+ * refuse such values up front with {@link IllegalArgumentException} so
callers cannot silently
+ * desync from Hudi's bucket layout. ({@code ':'} on the other hand is safe —
Hudi's parser has a
+ * dedicated look-ahead loop that handles values containing {@code ':'}, e.g.
timestamps like {@code
+ * "2023-10-25T10:01:13.182Z"}.)
+ *
+ * <h3>Supported key field types</h3>
+ *
+ * <p>See {@link #SUPPORTED_BUCKET_KEY_TYPE_ROOTS}. Composite or binary types
(ARRAY/MAP/ROW/
+ * BINARY/VARBINARY) would fall back to {@code Object#toString()} and produce
instance-bound,
+ * non-reproducible bucket ids; they are rejected at construction time.
+ */
+public class HudiKeyEncoder implements KeyEncoder {
+
+ /**
+ * Placeholder used by Hudi's serialization for a {@code null} bucket-key
field. Kept here for
+ * cross-validation in tests — the encoder itself never hashes this
literal because Hudi's
+ * parser turns it back into {@code null} before hashing.
+ */
+ public static final String NULL_RECORDKEY_PLACEHOLDER = "__null__";
+
+ /** Hudi's record-part separator. Values containing it are rejected. */
+ private static final char RECORD_KEY_PARTS_SEPARATOR = ',';
+
+ /**
+ * The set of {@link DataTypeRoot}s allowed as Hudi bucket key fields.
These are exactly the
+ * scalar types for which {@code toString()} (or a deterministic
equivalent) round-trips through
+ * Hudi's record-key serialization.
+ */
+ public static final Set<DataTypeRoot> SUPPORTED_BUCKET_KEY_TYPE_ROOTS =
+ Collections.unmodifiableSet(
+ EnumSet.of(
+ DataTypeRoot.CHAR,
+ DataTypeRoot.STRING,
+ DataTypeRoot.BOOLEAN,
+ DataTypeRoot.DECIMAL,
+ DataTypeRoot.TINYINT,
+ DataTypeRoot.SMALLINT,
+ DataTypeRoot.INTEGER,
+ DataTypeRoot.BIGINT,
+ DataTypeRoot.FLOAT,
+ DataTypeRoot.DOUBLE,
+ DataTypeRoot.DATE,
+ DataTypeRoot.TIME_WITHOUT_TIME_ZONE,
+ DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE,
+ DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE));
+
+ private final InternalRow.FieldGetter[] fieldGetters;
+ private final String[] keyFieldNames;
+
+ public HudiKeyEncoder(RowType rowType, List<String> keys) {
+ // for getting key fields out of fluss internal row
+ fieldGetters = new InternalRow.FieldGetter[keys.size()];
+ keyFieldNames = new String[keys.size()];
+ for (int i = 0; i < keys.size(); i++) {
+ String keyField = keys.get(i);
+ int keyIndex = rowType.getFieldIndex(keyField);
+ if (keyIndex < 0) {
+ throw new IllegalArgumentException(
+ "Bucket key field '" + keyField + "' does not exist in
row type.");
+ }
+ DataType keyDataType = rowType.getTypeAt(keyIndex);
+ validateSupportedBucketKeyType(keyField, keyDataType);
+ fieldGetters[i] = InternalRow.createFieldGetter(keyDataType,
keyIndex);
+ keyFieldNames[i] = keyField;
+ }
+ }
+
+ private static void validateSupportedBucketKeyType(String keyField,
DataType dataType) {
+ DataTypeRoot typeRoot = dataType.getTypeRoot();
+ if (!SUPPORTED_BUCKET_KEY_TYPE_ROOTS.contains(typeRoot)) {
+ throw new IllegalArgumentException(
+ "Unsupported Hudi bucket key type for field '"
+ + keyField
+ + "': "
+ + dataType
+ + ". A bucket key field must be a scalar type with
a deterministic "
+ + "string form; composite or binary types
(ARRAY/MAP/ROW/BINARY/VARBINARY) "
+ + "would otherwise fall back to Object#toString()
and produce "
+ + "non-reproducible bucket ids. Supported type
roots are: "
+ + SUPPORTED_BUCKET_KEY_TYPE_ROOTS);
+ }
+ }
+
+ @Override
+ public byte[] encodeKey(InternalRow row) {
+ // Build the same List<String> that Hudi's
KeyGenUtils.extractRecordKeysByFields would
+ // hand to List#hashCode() in BucketIdentifier#getBucketId.
+ //
+ // Hudi has two parse paths:
+ // * single-field record key (no ',' AND no ':' in the serialized
form): the parser
+ // returns the raw recordKey verbatim — placeholders are NOT
round-tripped, so
+ // "__null__"/"__empty__" stay as literal strings.
+ // * composite record key ("f1:v1,f2:v2,..."): the parser splits on
':'/',' and
+ // additionally turns "__null__" → null and "__empty__" → ""
before hashing.
+ //
+ // We mirror the same two-mode behaviour so the resulting
List#hashCode matches Hudi's.
+ boolean composite = fieldGetters.length > 1;
+ List<String> values = new ArrayList<>(fieldGetters.length);
+ for (int i = 0; i < fieldGetters.length; i++) {
+ Object value = fieldGetters[i].getFieldOrNull(row);
+ values.add(toHudiHashElement(keyFieldNames[i], value, composite));
+ }
+ 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
+ };
+ }
+
+ /**
+ * Produces the same {@code String} element Hudi's parser would have
placed in the list it
+ * hashes. The behaviour is mode-dependent because Hudi itself has two
parse modes:
+ *
+ * <ul>
+ * <li><b>single-field</b> ({@code composite == false}): Hudi returns
the raw recordKey
+ * verbatim, so {@code "__null__"} stays as the literal string. We
therefore emit the
+ * placeholder string for a null field, and reject {@code
','}-containing values only for
+ * safety (a single-field recordKey containing {@code ','} would
actually trip the
+ * composite quick-path check on Hudi's side too).
+ * <li><b>composite</b> ({@code composite == true}): Hudi parses {@code
"__null__"} → {@code
+ * null} and {@code "__empty__"} → {@code ""} before hashing. We
emit {@code null}/{@code
+ * ""} directly. Values literally equal to a placeholder, or
containing {@code ','}, are
+ * rejected because they would either collide with Hudi's reserved
sentinels or break
+ * Hudi's reverse parsing.
+ * </ul>
+ */
+ private static String toHudiHashElement(String fieldName, Object value,
boolean composite) {
+ if (value == null) {
+ // Composite mode: Hudi parses "__null__" back to null before
hashing.
+ // Single-field mode: Hudi keeps the recordKey verbatim, so we
keep the placeholder.
+ return composite ? null : NULL_RECORDKEY_PLACEHOLDER;
+ }
+ String stringValue =
+ (value instanceof BinaryString) ? value.toString() :
String.valueOf(value);
+ if (composite && stringValue.isEmpty()) {
+ // Composite mode: Hudi parses "__empty__" back to "" before
hashing.
+ return "";
+ }
+ if (composite) {
+ rejectIfContainsRecordSeparator(fieldName, stringValue);
+ // Guard against literal placeholder collisions — if a user
literally stores
+ // "__null__"/"__empty__" as the value, Hudi would round-trip it
to null/"" and
+ // we'd silently disagree.
+ if (NULL_RECORDKEY_PLACEHOLDER.equals(stringValue) ||
"__empty__".equals(stringValue)) {
+ throw new IllegalArgumentException(
+ "Bucket key field '"
+ + fieldName
+ + "' has value "
+ + stringValue
+ + " which collides with Hudi's reserved
record-key placeholder; "
+ + "this value cannot be safely used as a
composite Hudi bucket key.");
+ }
+ }
+ return stringValue;
+ }
+
+ private static void rejectIfContainsRecordSeparator(String fieldName,
String value) {
+ if (value.indexOf(RECORD_KEY_PARTS_SEPARATOR) >= 0) {
+ throw new IllegalArgumentException(
+ "Bucket key field '"
+ + fieldName
+ + "' has value containing ',', which is Hudi's
record-part "
+ + "separator and is not escaped by Hudi's
record-key serialization. "
+ + "Such a value would produce a bucket id
divergent from Hudi's "
+ + "BucketIdentifier and is rejected to keep the
bucket layout in sync.");
+ }
+ }
+}
diff --git a/fluss-lake/fluss-lake-hudi/pom.xml
b/fluss-lake/fluss-lake-hudi/pom.xml
index ddf6b1329..9fbce449f 100644
--- a/fluss-lake/fluss-lake-hudi/pom.xml
+++ b/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>
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/bucketing/HudiBucketingFunctionTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/bucketing/HudiBucketingFunctionTest.java
new file mode 100644
index 000000000..7038f87f6
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/bucketing/HudiBucketingFunctionTest.java
@@ -0,0 +1,586 @@
+/*
+ * 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;
+
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.row.encode.hudi.HudiKeyEncoder;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.apache.hudi.index.bucket.BucketIdentifier;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.fluss.row.encode.hudi.HudiKeyEncoder.NULL_RECORDKEY_PLACEHOLDER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit test for {@link HudiBucketingFunction}. */
+class HudiBucketingFunctionTest {
+
+ @Test
+ void testIntegerHash() {
+ int testValue = 42;
+ int bucketNum = 10;
+ String key = "id";
+
+ RowType rowType = RowType.of(new DataType[] {DataTypes.INT()}, new
String[] {key});
+
+ GenericRow row = GenericRow.of(testValue);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ String recordKey = String.valueOf(testValue);
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes = toBytes(new String[] {recordKey});
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, key,
bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testLongHash() {
+ long testValue = 1234567890123456789L;
+ int bucketNum = 10;
+ String key = "id";
+
+ RowType rowType = RowType.of(new DataType[] {DataTypes.BIGINT()}, new
String[] {key});
+
+ GenericRow row = GenericRow.of(testValue);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ String recordKey = String.valueOf(testValue);
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes = toBytes(new String[] {recordKey});
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, key,
bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testStringHash() {
+ // NOTE: ',' is Hudi's record-part separator and is rejected by the
encoder — use a
+ // delimiter-free message here. (See HudiKeyEncoder Javadoc for
details.)
+ String testValue = "Hello Hudi - Fluss this side!";
+ int bucketNum = 10;
+ String key = "name";
+
+ RowType rowType = RowType.of(new DataType[] {DataTypes.STRING()}, new
String[] {key});
+
+ GenericRow row = GenericRow.of(BinaryString.fromString(testValue));
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes = toBytes(new String[] {testValue});
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket = BucketIdentifier.getBucketId(testValue, key,
bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testDecimalHash() {
+ BigDecimal testValue = new BigDecimal("123.45");
+ Decimal decimal = Decimal.fromBigDecimal(testValue, 10, 2);
+ int bucketNum = 10;
+ String key = "amount";
+
+ RowType rowType = RowType.of(new DataType[] {DataTypes.DECIMAL(10,
2)}, new String[] {key});
+
+ GenericRow row = GenericRow.of(decimal);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes = toBytes(new String[]
{testValue.toPlainString()});
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket =
BucketIdentifier.getBucketId(testValue.toPlainString(), key, bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testTimestampEncodingHash() throws IOException {
+ long millis = 1698235273182L;
+ int nanos = 123456;
+ long micros = millis * 1000 + (nanos / 1000);
+ TimestampNtz testValue = TimestampNtz.fromMillis(millis, nanos);
+ int bucketNum = 10;
+ String key = "event_time";
+
+ RowType rowType = RowType.of(new DataType[] {DataTypes.TIMESTAMP(6)},
new String[] {key});
+
+ GenericRow row = GenericRow.of(testValue);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes = toBytes(new String[] {testValue.toString()});
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket = BucketIdentifier.getBucketId(testValue.toString(),
key, bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testDateHash() {
+ int dateValue = 19655;
+ int bucketNum = 10;
+ String key = "date";
+
+ RowType rowType = RowType.of(new DataType[] {DataTypes.DATE()}, new
String[] {key});
+ GenericRow row = GenericRow.of(dateValue);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes = toBytes(new String[]
{String.valueOf(dateValue)});
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket =
BucketIdentifier.getBucketId(String.valueOf(dateValue), key, bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testMultiKeysHashing() {
+ int testIntValue = 42;
+ long testLongValue = 1234567890123456789L;
+ String testStringValue = "Hello Hudi - Fluss this side!";
+ BigDecimal testValue = new BigDecimal("123.45");
+ Decimal decimal = Decimal.fromBigDecimal(testValue, 10, 2);
+ int bucketNum = 10;
+ String key = "age,id,name,amount";
+ String recordKey =
+ "age:"
+ + testIntValue
+ + ",id:"
+ + testLongValue
+ + ",name:"
+ + testStringValue
+ + ",amount:"
+ + testValue;
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING(),
+ DataTypes.DECIMAL(10, 2)
+ },
+ new String[] {"age", "id", "name", "amount"});
+ GenericRow row =
+ GenericRow.of(
+ testIntValue,
+ testLongValue,
+ BinaryString.fromString(testStringValue),
+ decimal);
+ HudiKeyEncoder encoder =
+ new HudiKeyEncoder(rowType, Arrays.asList("age", "id", "name",
"amount"));
+
+ // Encode with our implementation
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ // This is the equivalent bytes array which the key should be encoded
to.
+ byte[] equivalentBytes =
+ toBytes(
+ new String[] {
+ String.valueOf(testIntValue),
+ String.valueOf(testLongValue),
+ testStringValue,
+ String.valueOf(decimal)
+ });
+ assertThat(ourEncodedKey).isEqualTo(equivalentBytes);
+
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, key,
bucketNum);
+
+ HudiBucketingFunction hudiBucketingFunction = new
HudiBucketingFunction();
+ int ourBucket = hudiBucketingFunction.bucketing(ourEncodedKey,
bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testSingleFieldNull_keepsPlaceholderString_matchesHudi() {
+ // SINGLE-field mode: Hudi's KeyGenUtils.extractRecordKeysByFields
takes the
+ // fast path (no ',' AND no ':' in the recordKey) and returns the raw
string
+ // verbatim — i.e. the literal "__null__" — without round-tripping it
back to
+ // a real null. The strong-aligned encoder must mirror that and hash
the
+ // placeholder string, NOT a null element.
+ int bucketNum = 10;
+ String key = "name";
+
+ RowType rowType =
+ RowType.of(new DataType[] {DataTypes.STRING().copy(true)}, new
String[] {key});
+
+ GenericRow row = GenericRow.of((Object) null);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ byte[] placeholderStringBytes = toBytes(new String[]
{NULL_RECORDKEY_PLACEHOLDER});
+ byte[] literalNullStringBytes = toBytes(new String[] {"null"});
+
+ assertThat(ourEncodedKey).isEqualTo(placeholderStringBytes);
+ assertThat(ourEncodedKey).isNotEqualTo(literalNullStringBytes);
+
+ // Cross-validate against Hudi's recordKey-string overload (production
path).
+ int hudiBucket =
BucketIdentifier.getBucketId(NULL_RECORDKEY_PLACEHOLDER, key, bucketNum);
+ int ourBucket = new HudiBucketingFunction().bucketing(ourEncodedKey,
bucketNum);
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testBucketingRejectsInvalidBucketKey() {
+ HudiBucketingFunction function = new HudiBucketingFunction();
+
+ assertThatThrownBy(() -> function.bucketing(null, 10))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must not be null");
+
+ // length 0 — wrong length
+ assertThatThrownBy(() -> function.bucketing(new byte[0], 10))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("exactly 4 bytes");
+
+ // length < 4 — used to throw BufferUnderflowException, now must be IAE
+ assertThatThrownBy(() -> function.bucketing(new byte[] {1, 2, 3}, 10))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("exactly 4 bytes");
+
+ // length > 4 — used to silently truncate, now must be IAE
+ assertThatThrownBy(() -> function.bucketing(new byte[] {1, 2, 3, 4,
5}, 10))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("exactly 4 bytes");
+ }
+
+ @Test
+ void testBucketingRejectsNonPositiveNumBuckets() {
+ HudiBucketingFunction function = new HudiBucketingFunction();
+ byte[] anyKey = new byte[] {0, 0, 0, 1};
+
+ assertThatThrownBy(() -> function.bucketing(anyKey, 0))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be positive");
+ assertThatThrownBy(() -> function.bucketing(anyKey, -1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be positive");
+ }
+
+ @Test
+ void testCompositeBucketKeyMatchesHudiRecordKeyOverload() {
+ // Strong alignment: go through Hudi's production parse path
+ // (BucketIdentifier#getBucketId(String recordKey, ...)) instead of the
+ // pre-parsed List<String> overload, so any future divergence in
parsing
+ // (including ':'-handling for timestamps) is caught.
+ int bucketNum = 16;
+ String f1 = "user_id";
+ String f2 = "region";
+ int idValue = 12345;
+ String regionValue = "cn-north";
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING()},
+ new String[] {f1, f2});
+ GenericRow row = GenericRow.of(idValue,
BinaryString.fromString(regionValue));
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType, Arrays.asList(f1,
f2));
+
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ int ourBucket = new HudiBucketingFunction().bucketing(ourEncodedKey,
bucketNum);
+
+ String recordKey = f1 + ":" + idValue + "," + f2 + ":" + regionValue;
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, f1 + "," +
f2, bucketNum);
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testCompositeBucketKeyWithNullFieldMatchesHudiRecordKeyOverload() {
+ // Hudi serializes a null bucket-key field as "__null__" in the record
key, then
+ // parses it back to a `null` element before hashing. The
strong-aligned encoder
+ // must produce a bucket id identical to Hudi's recordKey-string parse
path.
+ int bucketNum = 8;
+ String f1 = "user_id";
+ String f2 = "region";
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT().copy(true),
DataTypes.STRING().copy(true)},
+ new String[] {f1, f2});
+ // f2 is null on purpose
+ GenericRow row = GenericRow.of(42, null);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType, Arrays.asList(f1,
f2));
+
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ int ourBucket = new HudiBucketingFunction().bucketing(ourEncodedKey,
bucketNum);
+
+ // The on-wire record key Hudi would have built for this row.
+ String recordKey = f1 + ":42," + f2 + ":" + NULL_RECORDKEY_PLACEHOLDER;
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, f1 + "," +
f2, bucketNum);
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testBooleanAndIntegralTypes() {
+ int bucketNum = 10;
+ // BOOLEAN
+ assertSingleFieldRoundTrip(
+ DataTypes.BOOLEAN(), true, String.valueOf(true), "flag",
bucketNum);
+ assertSingleFieldRoundTrip(
+ DataTypes.BOOLEAN(), false, String.valueOf(false), "flag",
bucketNum);
+ // TINYINT / SMALLINT
+ assertSingleFieldRoundTrip(DataTypes.TINYINT(), (byte) 7, "7", "b",
bucketNum);
+ assertSingleFieldRoundTrip(DataTypes.SMALLINT(), (short) 12345,
"12345", "s", bucketNum);
+ // FLOAT (use a value whose Float.toString is stable across JVMs)
+ assertSingleFieldRoundTrip(DataTypes.FLOAT(), 3.5f,
Float.toString(3.5f), "f", bucketNum);
+ }
+
+ @Test
+ void testDateAndTimeTypes() {
+ int bucketNum = 10;
+ // DATE is stored as days-since-epoch int internally; record key is
its String value.
+ int dateDays = 19852;
+ assertSingleFieldRoundTrip(
+ DataTypes.DATE(), dateDays, String.valueOf(dateDays), "d",
bucketNum);
+ // TIME is stored as ms-of-day int.
+ int timeMillis = 12345678;
+ assertSingleFieldRoundTrip(
+ DataTypes.TIME(), timeMillis, String.valueOf(timeMillis), "t",
bucketNum);
+ }
+
+ @Test
+ void testTimestampLtzType() throws IOException {
+ int bucketNum = 10;
+ TimestampLtz ts =
TimestampLtz.fromInstant(Instant.ofEpochMilli(1700000000000L));
+ RowType rowType =
+ RowType.of(new DataType[] {DataTypes.TIMESTAMP_LTZ(6)}, new
String[] {"ts"});
+ GenericRow row = GenericRow.of(ts);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList("ts"));
+ byte[] enc = encoder.encodeKey(row);
+ byte[] expected = toBytes(new String[] {ts.toString()});
+ assertThat(enc).isEqualTo(expected);
+
+ int hudiBucket =
+
BucketIdentifier.getBucketId(Collections.singletonList(ts.toString()),
bucketNum);
+ int ourBucket = new HudiBucketingFunction().bucketing(enc, bucketNum);
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testBucketingNumBucketsBoundaryValues() {
+ HudiBucketingFunction f = new HudiBucketingFunction();
+ // numBuckets == 1 => bucket id always 0
+ for (int sample : new int[] {0, 1, -1, Integer.MAX_VALUE,
Integer.MIN_VALUE}) {
+ byte[] key = intToBytes(sample);
+ assertThat(f.bucketing(key, 1)).isEqualTo(0);
+ }
+ // numBuckets == Integer.MAX_VALUE: bucket id == hash & MAX_VALUE
+ int hash = Integer.MIN_VALUE; // stress sign-bit handling
+ byte[] key = intToBytes(hash);
+ int expected = (hash & Integer.MAX_VALUE) % Integer.MAX_VALUE;
+ assertThat(f.bucketing(key, Integer.MAX_VALUE)).isEqualTo(expected);
+ // bucket id must always be in [0, numBuckets)
+ assertThat(f.bucketing(key,
7)).isGreaterThanOrEqualTo(0).isLessThan(7);
+ }
+
+ private void assertSingleFieldRoundTrip(
+ DataType dataType, Object value, String stringified, String key,
int bucketNum) {
+ RowType rowType = RowType.of(new DataType[] {dataType}, new String[]
{key});
+ GenericRow row = GenericRow.of(value);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Collections.singletonList(key));
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ byte[] expected = toBytes(new String[] {stringified});
+ assertThat(ourEncodedKey).isEqualTo(expected);
+
+ int hudiBucket = BucketIdentifier.getBucketId(stringified, key,
bucketNum);
+ int ourBucket = new HudiBucketingFunction().bucketing(ourEncodedKey,
bucketNum);
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ private static byte[] intToBytes(int v) {
+ return new byte[] {(byte) (v >>> 24), (byte) (v >>> 16), (byte) (v >>>
8), (byte) v};
+ }
+
+ private byte[] toBytes(String[] value) {
+ List<String> values = Arrays.asList(value);
+ return ByteBuffer.allocate(4).putInt(values.hashCode()).array();
+ }
+
+ // ------------------------------------------------------------------
+ // End-to-end tests against Hudi's recordKey-string overloads — these
+ // exercise the same parsing path Hudi takes in production (HoodieKey
+ // -> getHashKeys via ':' / ',' split) and therefore catch divergence
+ // that the List<String>-based overload would silently hide.
+ // ------------------------------------------------------------------
+
+ @Test
+ void testCompositeBucketKeyMatchesHudiRecordKeyOverload_safeValues() {
+ // Values intentionally free of ':' and ',' — the encoder must agree
with Hudi's
+ // record-key-parsing path under the documented contract.
+ int bucketNum = 16;
+ String f1 = "user_id";
+ String f2 = "region";
+ int idValue = 12345;
+ String regionValue = "cn-north";
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING()},
+ new String[] {f1, f2});
+ GenericRow row = GenericRow.of(idValue,
BinaryString.fromString(regionValue));
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType, Arrays.asList(f1,
f2));
+
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ int ourBucket = new HudiBucketingFunction().bucketing(ourEncodedKey,
bucketNum);
+
+ // Build Hudi's wire-format record key: "f1:v1,f2:v2", then go through
the
+ // recordKey-string overload that splits on ':' / ',' before hashing.
+ String recordKey = f1 + ":" + idValue + "," + f2 + ":" + regionValue;
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, f1 + "," +
f2, bucketNum);
+
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testCompositeKeyWithColonInValue_matchesHudiRecordKeyOverload() {
+ // Hudi's KeyGenUtils.extractRecordKeysByFields has a dedicated
look-ahead loop
+ // that correctly handles values containing ':' (e.g. timestamps like
+ // "2023-10-25T10:01:13.182Z"). The strong-aligned encoder must match
Hudi
+ // bit-for-bit on this case — not merely "document the divergence".
+ int bucketNum = 8;
+ String f1 = "tenant";
+ String f2 = "ts";
+ String tenant = "acme";
+ TimestampLtz ts =
TimestampLtz.fromInstant(Instant.ofEpochMilli(1700000000000L));
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.STRING(),
DataTypes.TIMESTAMP_LTZ(6)},
+ new String[] {f1, f2});
+ GenericRow row = GenericRow.of(BinaryString.fromString(tenant), ts);
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType, Arrays.asList(f1,
f2));
+
+ byte[] ourEncodedKey = encoder.encodeKey(row);
+ int ourBucket = new HudiBucketingFunction().bucketing(ourEncodedKey,
bucketNum);
+
+ String recordKey = f1 + ":" + tenant + "," + f2 + ":" + ts.toString();
+ int hudiBucket = BucketIdentifier.getBucketId(recordKey, f1 + "," +
f2, bucketNum);
+
+ // Strong alignment: bucket ids MUST match.
+ assertThat(ourBucket).isEqualTo(hudiBucket);
+ }
+
+ @Test
+ void testRejectsValueContainingRecordSeparator() {
+ // ',' is Hudi's record-part separator and is NOT escaped by Hudi's
record-key
+ // serialization. The encoder must refuse such values rather than
silently
+ // diverge from Hudi's bucket layout.
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.STRING(),
DataTypes.STRING()},
+ new String[] {"tenant", "name"});
+ GenericRow row =
+ GenericRow.of(
+ BinaryString.fromString("acme"),
BinaryString.fromString("smith, john"));
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Arrays.asList("tenant", "name"));
+
+ assertThatThrownBy(() -> encoder.encodeKey(row))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("'name'")
+ .hasMessageContaining("','");
+ }
+
+ @Test
+ void testRejectsLiteralPlaceholderValueInCompositeKey() {
+ // Composite-key mode: a user-supplied value equal to Hudi's reserved
placeholder
+ // would be round-tripped to null by Hudi's parser, silently changing
the bucket id.
+ // (In single-field mode the placeholder is kept verbatim by Hudi, so
it's safe.)
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.STRING(),
DataTypes.STRING()},
+ new String[] {"tenant", "name"});
+ GenericRow row =
+ GenericRow.of(
+ BinaryString.fromString("acme"),
+ BinaryString.fromString(NULL_RECORDKEY_PLACEHOLDER));
+ HudiKeyEncoder encoder = new HudiKeyEncoder(rowType,
Arrays.asList("tenant", "name"));
+
+ assertThatThrownBy(() -> encoder.encodeKey(row))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("reserved record-key placeholder");
+ }
+
+ @Test
+ void testRejectsUnsupportedBucketKeyType() {
+ // BYTES / BINARY would otherwise fall through to Object#toString()
and produce
+ // instance-bound bucket ids (e.g. "[B@1a2b3c"). The encoder must
refuse them up
+ // front rather than silently corrupt the bucket layout.
+ RowType rowType = RowType.of(new DataType[] {DataTypes.BYTES()}, new
String[] {"payload"});
+
+ assertThatThrownBy(() -> new HudiKeyEncoder(rowType,
Collections.singletonList("payload")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Unsupported Hudi bucket key type")
+ .hasMessageContaining("payload");
+ }
+}
