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 33bfd9ea5 [lake/hudi] Introduce Hudi LakeCommitter (#3509)
33bfd9ea5 is described below
commit 33bfd9ea5f24ced04d5f6f53ab20b2bb6768d9f1
Author: fhan <[email protected]>
AuthorDate: Tue Jun 23 22:32:11 2026 +0800
[lake/hudi] Introduce Hudi LakeCommitter (#3509)
---
.../fluss/client/lookup/PrimaryKeyLookuper.java | 10 +-
.../client/table/writer/UpsertWriterImpl.java | 12 +-
.../org/apache/fluss/row/decode/KeyDecoder.java | 6 +
.../org/apache/fluss/row/encode/KeyEncoder.java | 30 ++
.../fluss/row/decode/CompactedKeyDecoderTest.java | 19 +
.../fluss/row/encode/CompactedKeyEncoderTest.java | 45 ++
fluss-lake/fluss-lake-hudi/pom.xml | 105 ++++
.../apache/fluss/lake/hudi/HudiLakeStorage.java | 5 +-
.../fluss/lake/hudi/source/HudiRecordReader.java | 4 +-
.../fluss/lake/hudi/tiering/HudiCommittable.java | 122 +++++
.../hudi/tiering/HudiCommittableSerializer.java | 70 +++
.../fluss/lake/hudi/tiering/HudiLakeCommitter.java | 301 ++++++++++++
.../lake/hudi/tiering/HudiLakeTieringFactory.java | 16 +-
.../fluss/lake/hudi/tiering/HudiLakeWriter.java | 4 +-
.../fluss/lake/hudi/tiering/HudiWriteResult.java | 50 +-
.../hudi/tiering/HudiWriteResultSerializer.java | 50 +-
.../fluss/lake/hudi/tiering/HudiWriteStats.java | 119 +++++
.../lake/hudi/tiering/HudiWriteStatsSerde.java | 121 +++++
.../fluss/lake/hudi/tiering/RecordWriter.java | 4 +-
.../hudi/tiering/writer/RecordWriteBuffer.java | 11 +-
.../lake/hudi/utils/meta/CkpMetadataFactory.java | 7 +-
.../fluss/lake/hudi/HudiLakeStorageTest.java | 41 ++
.../hudi/testutils/FlinkHudiTieringTestBase.java | 539 +++++++++++++++++++++
.../tiering/HudiCommittableSerializerTest.java | 117 +++++
.../fluss/lake/hudi/tiering/HudiTieringITCase.java | 233 +++++++++
.../fluss/lake/hudi/tiering/HudiTieringTest.java | 274 +++++++++++
.../tiering/HudiWriteResultSerializerTest.java | 47 +-
.../src/test/resources/log4j2-test.properties | 2 +-
28 files changed, 2263 insertions(+), 101 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java
b/fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java
index 26a03c47c..c6c720ea7 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java
@@ -81,10 +81,12 @@ class PrimaryKeyLookuper extends AbstractLookuper
implements Lookuper {
tableInfo.getTableConfig(),
tableInfo.isDefaultBucketKey());
this.bucketKeyEncoder =
- tableInfo.isDefaultBucketKey()
- ? primaryKeyEncoder
- : KeyEncoder.ofBucketKeyEncoder(
- lookupRowType, tableInfo.getBucketKeys(),
lakeFormat);
+ KeyEncoder.ofBucketKeyEncoder(
+ lookupRowType,
+ tableInfo.getBucketKeys(),
+ tableInfo.getTableConfig(),
+ tableInfo.isDefaultBucketKey(),
+ primaryKeyEncoder);
this.bucketingFunction = BucketingFunction.of(lakeFormat);
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java
index 6b7f821a1..8417855cc 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/writer/UpsertWriterImpl.java
@@ -88,12 +88,12 @@ class UpsertWriterImpl extends AbstractTableWriter
implements UpsertWriter {
tableInfo.getTableConfig(),
tableInfo.isDefaultBucketKey());
this.bucketKeyEncoder =
- tableInfo.isDefaultBucketKey()
- ? primaryKeyEncoder
- : KeyEncoder.ofBucketKeyEncoder(
- tableInfo.getRowType(),
- tableInfo.getBucketKeys(),
-
tableInfo.getTableConfig().getDataLakeFormat().orElse(null));
+ KeyEncoder.ofBucketKeyEncoder(
+ tableInfo.getRowType(),
+ tableInfo.getBucketKeys(),
+ tableInfo.getTableConfig(),
+ tableInfo.isDefaultBucketKey(),
+ primaryKeyEncoder);
this.kvFormat = tableInfo.getTableConfig().getKvFormat();
this.writeFormat = WriteFormat.fromKvFormat(this.kvFormat);
diff --git
a/fluss-common/src/main/java/org/apache/fluss/row/decode/KeyDecoder.java
b/fluss-common/src/main/java/org/apache/fluss/row/decode/KeyDecoder.java
index 770526879..2c4dbdba7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/decode/KeyDecoder.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/decode/KeyDecoder.java
@@ -56,6 +56,12 @@ public interface KeyDecoder {
short kvFormatVersion,
@Nullable DataLakeFormat lakeFormat,
boolean isDefaultBucketKey) {
+ // Hudi's HudiKeyEncoder is lossy (4-byte hash); primary keys are
encoded by
+ // CompactedKeyEncoder, so decoding must use CompactedKeyDecoder as
well.
+ if (lakeFormat == DataLakeFormat.HUDI) {
+ return CompactedKeyDecoder.createKeyDecoder(rowType, keyFields);
+ }
+
if (kvFormatVersion == 1 || (kvFormatVersion == 2 &&
isDefaultBucketKey)) {
if (lakeFormat == null || lakeFormat == DataLakeFormat.LANCE) {
return CompactedKeyDecoder.createKeyDecoder(rowType,
keyFields);
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 fe52e926c..bea1a21c9 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
@@ -30,6 +30,8 @@ import javax.annotation.Nullable;
import java.util.List;
import java.util.Optional;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
/** An interface for encoding key of row into bytes. */
public interface KeyEncoder {
@@ -113,6 +115,34 @@ public interface KeyEncoder {
return of(rowType, keyFields, lakeFormat);
}
+ /**
+ * Creates a bucket key encoder for bucket calculation and reuses the
primary key encoder when
+ * the bucket key encoding is compatible with the primary key encoding.
+ *
+ * <p>When the bucket key is the same as the primary key, most formats can
reuse the primary key
+ * encoder and avoid encoding twice. Hudi is an exception: its primary key
encoder must be
+ * lossless, while bucket routing must keep using Hudi's hash-based bucket
key encoding.
+ *
+ * @param rowType the row type of the input row
+ * @param keyFields the bucket key fields to encode
+ * @param tableConfig the table configuration containing lake format
+ * @param isDefaultBucketKey true if bucket key equals primary key
+ * @param primaryKeyEncoder the primary key encoder to reuse when
compatible
+ * @return the bucket key encoder
+ */
+ static KeyEncoder ofBucketKeyEncoder(
+ RowType rowType,
+ List<String> keyFields,
+ TableConfig tableConfig,
+ boolean isDefaultBucketKey,
+ KeyEncoder primaryKeyEncoder) {
+ DataLakeFormat dataLakeFormat =
tableConfig.getDataLakeFormat().orElse(null);
+ if (isDefaultBucketKey && dataLakeFormat != DataLakeFormat.HUDI) {
+ return checkNotNull(primaryKeyEncoder, "Primary key encoder must
not be null.");
+ }
+ return ofBucketKeyEncoder(rowType, keyFields, dataLakeFormat);
+ }
+
/**
* Creates a key encoder based on the datalake format.
*
diff --git
a/fluss-common/src/test/java/org/apache/fluss/row/decode/CompactedKeyDecoderTest.java
b/fluss-common/src/test/java/org/apache/fluss/row/decode/CompactedKeyDecoderTest.java
index 529fb4c57..c6aa4cb53 100644
---
a/fluss-common/src/test/java/org/apache/fluss/row/decode/CompactedKeyDecoderTest.java
+++
b/fluss-common/src/test/java/org/apache/fluss/row/decode/CompactedKeyDecoderTest.java
@@ -18,6 +18,7 @@
package org.apache.fluss.row.decode;
+import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.row.BinaryString;
import org.apache.fluss.row.Decimal;
import org.apache.fluss.row.InternalRow;
@@ -175,6 +176,24 @@ class CompactedKeyDecoderTest {
}
}
+ @Test
+ void testHudiPrimaryKeyDecoderUsesCompactedDecoder() {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.STRING()},
+ new String[] {"id", "name"});
+ List<String> pk = Arrays.asList("id", "name");
+
+ assertThat(
+ KeyDecoder.ofPrimaryKeyDecoder(
+ rowType, pk, (short) 1, DataLakeFormat.HUDI,
true))
+ .isInstanceOf(CompactedKeyDecoder.class);
+ assertThat(
+ KeyDecoder.ofPrimaryKeyDecoder(
+ rowType, pk, (short) 2, DataLakeFormat.HUDI,
false))
+ .isInstanceOf(CompactedKeyDecoder.class);
+ }
+
@Test
void testDecodeInvalidKey() {
RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
diff --git
a/fluss-common/src/test/java/org/apache/fluss/row/encode/CompactedKeyEncoderTest.java
b/fluss-common/src/test/java/org/apache/fluss/row/encode/CompactedKeyEncoderTest.java
index d7691886b..ab8d5876c 100644
---
a/fluss-common/src/test/java/org/apache/fluss/row/encode/CompactedKeyEncoderTest.java
+++
b/fluss-common/src/test/java/org/apache/fluss/row/encode/CompactedKeyEncoderTest.java
@@ -17,12 +17,17 @@
package org.apache.fluss.row.encode;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.TableConfig;
import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.row.BinaryString;
import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.compacted.CompactedRowDeserializer;
import org.apache.fluss.row.compacted.CompactedRowReader;
+import org.apache.fluss.row.encode.hudi.HudiKeyEncoder;
import org.apache.fluss.row.indexed.IndexedRow;
import org.apache.fluss.row.indexed.IndexedRowTest;
import org.apache.fluss.row.indexed.IndexedRowWriter;
@@ -83,6 +88,36 @@ class CompactedKeyEncoderTest {
assertThat(encodedKey.getString(0).toString()).isEqualTo("a2");
}
+ @Test
+ void testBucketKeyEncoderReusesPrimaryKeyEncoderWhenCompatible() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.INT()}, new
String[] {"id"});
+ List<String> primaryKeys = Collections.singletonList("id");
+ KeyEncoder primaryKeyEncoder =
CompactedKeyEncoder.createKeyEncoder(rowType, primaryKeys);
+
+ assertThat(
+ KeyEncoder.ofBucketKeyEncoder(
+ rowType, primaryKeys, tableConfig(), true,
primaryKeyEncoder))
+ .isSameAs(primaryKeyEncoder);
+ }
+
+ @Test
+ void testHudiBucketKeyEncoderDoesNotReusePrimaryKeyEncoder() {
+ RowType rowType = RowType.of(new DataType[] {DataTypes.INT()}, new
String[] {"id"});
+ List<String> primaryKeys = Collections.singletonList("id");
+ KeyEncoder primaryKeyEncoder =
CompactedKeyEncoder.createKeyEncoder(rowType, primaryKeys);
+
+ KeyEncoder bucketKeyEncoder =
+ KeyEncoder.ofBucketKeyEncoder(
+ rowType,
+ primaryKeys,
+ tableConfig(DataLakeFormat.HUDI),
+ true,
+ primaryKeyEncoder);
+
+ assertThat(bucketKeyEncoder).isInstanceOf(HudiKeyEncoder.class);
+ assertThat(bucketKeyEncoder).isNotSameAs(primaryKeyEncoder);
+ }
+
@Test
void testGetKey() {
// test int, long as primary key
@@ -179,4 +214,14 @@ class CompactedKeyEncoderTest {
compactedRowDeserializer.deserialize(compactedRowReader, genericRow);
return genericRow;
}
+
+ private static TableConfig tableConfig() {
+ return new TableConfig(new Configuration());
+ }
+
+ private static TableConfig tableConfig(DataLakeFormat dataLakeFormat) {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.TABLE_DATALAKE_FORMAT, dataLakeFormat);
+ return new TableConfig(configuration);
+ }
}
diff --git a/fluss-lake/fluss-lake-hudi/pom.xml
b/fluss-lake/fluss-lake-hudi/pom.xml
index 64f1ca2c1..3006c73c9 100644
--- a/fluss-lake/fluss-lake-hudi/pom.xml
+++ b/fluss-lake/fluss-lake-hudi/pom.xml
@@ -69,6 +69,43 @@
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ <version>${fluss.hadoop.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <artifactId>commons-io</artifactId>
+ <groupId>commons-io</groupId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-lang</groupId>
+ <artifactId>commons-lang</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</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>
+ </exclusions>
+ </dependency>
+
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
@@ -137,6 +174,12 @@
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
@@ -163,6 +206,29 @@
</dependency>
<!-- test dependency -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
<dependency>
<groupId>org.apache.fluss</groupId>
<artifactId>fluss-common</artifactId>
@@ -175,6 +241,45 @@
<groupId>org.apache.fluss</groupId>
<artifactId>fluss-test-utils</artifactId>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-test</artifactId>
+ <version>${curator.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Flink test dependency -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-${flink.major.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-test-utils</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
index 55ce818a9..9c635d92b 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
@@ -20,6 +20,7 @@ package org.apache.fluss.lake.hudi;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.lake.hudi.source.HudiLakeSource;
import org.apache.fluss.lake.hudi.source.HudiSplit;
+import org.apache.fluss.lake.hudi.tiering.HudiLakeTieringFactory;
import org.apache.fluss.lake.lakestorage.LakeCatalog;
import org.apache.fluss.lake.lakestorage.LakeStorage;
import org.apache.fluss.lake.source.LakeSource;
@@ -37,9 +38,7 @@ public class HudiLakeStorage implements LakeStorage {
@Override
public LakeTieringFactory<?, ?> createLakeTieringFactory() {
- throw new UnsupportedOperationException(
- "Hudi lake tiering writer is not implemented yet, so
HudiLakeStorage does not "
- + "support creating a LakeTieringFactory.");
+ return new HudiLakeTieringFactory(hudiConfig);
}
@Override
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java
index 9cd52a15b..ea88aac94 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiRecordReader.java
@@ -103,10 +103,10 @@ public class HudiRecordReader implements RecordReader {
boolean emitDelete =
hudiTableInfo.getTableType() ==
HoodieTableType.MERGE_ON_READ
&& (flinkHudiOptions
- .getString(FlinkOptions.QUERY_TYPE)
+ .get(FlinkOptions.QUERY_TYPE)
.equals(FlinkOptions.QUERY_TYPE_SNAPSHOT)
|| flinkHudiOptions
- .getString(FlinkOptions.QUERY_TYPE)
+ .get(FlinkOptions.QUERY_TYPE)
.equals(FlinkOptions.QUERY_TYPE_INCREMENTAL));
UnifiedHudiTableReader unifiedHudiTableReader =
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittable.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittable.java
new file mode 100644
index 000000000..e1266d639
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittable.java
@@ -0,0 +1,122 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/** The committable aggregated from Hudi write results for one tiering round.
*/
+public class HudiCommittable implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Map<String, HudiWriteStats> writeStats;
+ private final Map<String, HudiWriteStats> compactionWriteStats;
+
+ public HudiCommittable(
+ Map<String, HudiWriteStats> writeStats,
+ @Nullable Map<String, HudiWriteStats> compactionWriteStats) {
+ this.writeStats = copyWriteStats(writeStats);
+ this.compactionWriteStats = copyWriteStats(compactionWriteStats);
+ }
+
+ public Map<String, HudiWriteStats> getWriteStats() {
+ return writeStats;
+ }
+
+ public Map<String, HudiWriteStats> getCompactionWriteStats() {
+ return compactionWriteStats;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ private static Map<String, HudiWriteStats> copyWriteStats(
+ @Nullable Map<String, HudiWriteStats> statsByInstant) {
+ if (statsByInstant == null || statsByInstant.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(new HashMap<>(statsByInstant));
+ }
+
+ /** Builder for {@link HudiCommittable}. */
+ public static class Builder {
+
+ private final Map<String, HudiWriteStats> writeStats = new HashMap<>();
+ private final Map<String, HudiWriteStats> compactionWriteStats = new
HashMap<>();
+
+ public Builder addWriteStats(Map<String, HudiWriteStats>
statsByInstant) {
+ addAll(writeStats, statsByInstant);
+ return this;
+ }
+
+ public Builder addCompactionWriteStats(
+ @Nullable Map<String, HudiWriteStats> statsByInstant) {
+ addAll(compactionWriteStats, statsByInstant);
+ return this;
+ }
+
+ public HudiCommittable build() {
+ return new HudiCommittable(writeStats, compactionWriteStats);
+ }
+
+ private static void addAll(
+ Map<String, HudiWriteStats> target, @Nullable Map<String,
HudiWriteStats> source) {
+ if (source == null || source.isEmpty()) {
+ return;
+ }
+ for (Map.Entry<String, HudiWriteStats> entry : source.entrySet()) {
+ target.merge(entry.getKey(), entry.getValue(),
HudiWriteStats::merge);
+ }
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof HudiCommittable)) {
+ return false;
+ }
+ HudiCommittable that = (HudiCommittable) o;
+ return Objects.equals(writeStats, that.writeStats)
+ && Objects.equals(compactionWriteStats,
that.compactionWriteStats);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(writeStats, compactionWriteStats);
+ }
+
+ @Override
+ public String toString() {
+ return "HudiCommittable{"
+ + "writeStats="
+ + writeStats
+ + ", compactionWriteStats="
+ + compactionWriteStats
+ + '}';
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializer.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializer.java
new file mode 100644
index 000000000..39eb3a1a9
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializer.java
@@ -0,0 +1,70 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Map;
+
+/** Serializer for {@link HudiCommittable}. */
+public class HudiCommittableSerializer implements
SimpleVersionedSerializer<HudiCommittable> {
+
+ private static final int CURRENT_VERSION = 1;
+
+ @Override
+ public int getVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public byte[] serialize(HudiCommittable hudiCommittable) throws
IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(baos)) {
+ HudiWriteStatsSerde.writeStatsMap(dos,
hudiCommittable.getWriteStats());
+ HudiWriteStatsSerde.writeStatsMap(dos,
hudiCommittable.getCompactionWriteStats());
+ return baos.toByteArray();
+ }
+ }
+
+ @Override
+ public HudiCommittable deserialize(int version, byte[] serialized) throws
IOException {
+ if (version != CURRENT_VERSION) {
+ throw new IOException(
+ "Unsupported HudiCommittable version "
+ + version
+ + ", expected "
+ + CURRENT_VERSION);
+ }
+
+ Map<String, HudiWriteStats> writeStats;
+ Map<String, HudiWriteStats> compactionWriteStats;
+ try (DataInputStream dis = new DataInputStream(new
ByteArrayInputStream(serialized))) {
+ writeStats = HudiWriteStatsSerde.readStatsMap(dis, "WriteStats");
+ compactionWriteStats = HudiWriteStatsSerde.readStatsMap(dis,
"CompactionWriteStats");
+ if (dis.available() > 0) {
+ throw new IOException("Corrupted serialization: trailing bytes
" + dis.available());
+ }
+ }
+ return new HudiCommittable(writeStats, compactionWriteStats);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java
new file mode 100644
index 000000000..44d898e29
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java
@@ -0,0 +1,301 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
+import org.apache.fluss.lake.committer.LakeCommitResult;
+import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadataProvider;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.exception.HoodieException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static
org.apache.fluss.lake.writer.LakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER;
+
+/** Hudi implementation of {@link LakeCommitter}. */
+public class HudiLakeCommitter implements LakeCommitter<HudiWriteResult,
HudiCommittable> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(HudiLakeCommitter.class);
+
+ private static final String COMMITTER_USER = "commit-user";
+
+ private final HudiWriteTableInfo hudiTableInfo;
+ private final HoodieFlinkWriteClient<?> writeClient;
+ private final CkpMetadata ckpMetadata;
+
+ public HudiLakeCommitter(
+ HudiCatalogProvider hudiCatalogProvider,
+ CkpMetadataProvider ckpMetadataProvider,
+ TablePath tablePath)
+ throws IOException {
+ this.hudiTableInfo = HudiWriteTableInfo.create(hudiCatalogProvider,
tablePath);
+ this.writeClient = hudiTableInfo.getWriteClient();
+ this.ckpMetadata = ckpMetadataProvider.get(tablePath, hudiTableInfo);
+ LOG.info(
+ "Created HudiLakeCommitter with configuration {}.",
hudiTableInfo.getFlinkConfig());
+ }
+
+ @Override
+ public HudiCommittable toCommittable(List<HudiWriteResult>
hudiWriteResults) {
+ HudiCommittable.Builder committableBuilder = HudiCommittable.builder();
+ for (HudiWriteResult hudiWriteResult : hudiWriteResults) {
+ committableBuilder.addWriteStats(hudiWriteResult.getWriteStats());
+
committableBuilder.addCompactionWriteStats(hudiWriteResult.getCompactionWriteStats());
+ }
+ return committableBuilder.build();
+ }
+
+ @Override
+ public LakeCommitResult commit(
+ HudiCommittable committable, Map<String, String>
snapshotProperties)
+ throws IOException {
+ ensureNoCompactionWriteStats(committable);
+
+ Map<String, HudiWriteStats> writeStatsByInstant =
committable.getWriteStats();
+ if (writeStatsByInstant.size() != 1) {
+ throw new IOException(
+ "Hudi write stats must contain exactly one instant, but
got "
+ + writeStatsByInstant.keySet()
+ + ".");
+ }
+
+ Map.Entry<String, HudiWriteStats> entry =
writeStatsByInstant.entrySet().iterator().next();
+ String instant = entry.getKey();
+ HudiWriteStats writeStats = entry.getValue();
+
+ Map<String, String> commitMetadata = new HashMap<>(snapshotProperties);
+ commitMetadata.put(COMMITTER_USER, FLUSS_LAKE_TIERING_COMMIT_USER);
+
+ try {
+ validateWriteStats(instant, writeStats);
+
+ LOG.info(
+ "Committing Hudi instant {} with {} write stat entries and
metadata {}.",
+ instant,
+ writeStats.getWriteStats().size(),
+ commitMetadata);
+ boolean committed =
+ writeClient.commitStats(
+ instant,
+ writeStats.getWriteStats(),
+ Option.of(commitMetadata),
+
hudiTableInfo.getMetaClient().getCommitActionType());
+ if (!committed) {
+ IOException failure =
+ new IOException("Failed to commit Hudi instant " +
instant + ".");
+ try {
+ abortInstant(instant);
+ } catch (IOException abortFailure) {
+ failure.addSuppressed(abortFailure);
+ }
+ throw failure;
+ }
+
+ ckpMetadata.commitInstant(instant);
+ LOG.info("Committed Hudi instant {} successfully.", instant);
+ return
LakeCommitResult.committedIsReadable(parseSnapshotId(instant));
+ } catch (Exception e) {
+ if (e instanceof IOException) {
+ throw (IOException) e;
+ }
+ throw new IOException("Failed to commit Hudi instant " + instant +
".", e);
+ }
+ }
+
+ @Override
+ public void abort(HudiCommittable committable) throws IOException {
+ Set<String> instants = new
LinkedHashSet<>(committable.getWriteStats().keySet());
+ instants.addAll(committable.getCompactionWriteStats().keySet());
+ IOException failure = null;
+ for (String instant : instants) {
+ try {
+ abortInstant(instant);
+ LOG.info("Aborted Hudi instant {}.", instant);
+ } catch (IOException e) {
+ failure =
+ addSuppressed(
+ failure,
+ new IOException(
+ "Failed to abort Hudi instant " +
instant + ".", e));
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private static void ensureNoCompactionWriteStats(HudiCommittable
committable)
+ throws IOException {
+ Map<String, HudiWriteStats> compactionWriteStats =
committable.getCompactionWriteStats();
+ if (!compactionWriteStats.isEmpty()) {
+ throw new IOException(
+ "Hudi compaction write stats are not supported yet, but
got instants "
+ + compactionWriteStats.keySet()
+ + ".");
+ }
+ }
+
+ @Nullable
+ @Override
+ public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long
latestLakeSnapshotIdOfFluss)
+ throws IOException {
+ HoodieTimeline latestLakeTimeline =
+
getCompletedTimelineCommittedBy(FLUSS_LAKE_TIERING_COMMIT_USER);
+ Optional<HoodieInstant> latestLakeInstant =
+ latestLakeTimeline.getReverseOrderedInstants().findFirst();
+ if (!latestLakeInstant.isPresent()) {
+ return null;
+ }
+
+ long latestLakeSnapshotId =
parseSnapshotId(latestLakeInstant.get().requestedTime());
+ if (latestLakeSnapshotIdOfFluss != null
+ && latestLakeSnapshotId <= latestLakeSnapshotIdOfFluss) {
+ return null;
+ }
+
+ HoodieCommitMetadata metadata =
+ latestLakeTimeline.readCommitMetadata(latestLakeInstant.get());
+ if (metadata == null) {
+ throw new IOException("Failed to load committed Hudi instant
metadata.");
+ }
+ Map<String, String> extraMetadata = metadata.getExtraMetadata();
+ if (extraMetadata == null) {
+ throw new IOException("Failed to load committed Hudi instant extra
metadata.");
+ }
+ return new CommittedLakeSnapshot(latestLakeSnapshotId, extraMetadata);
+ }
+
+ @Override
+ public void close() throws Exception {
+ IOUtils.closeQuietly(ckpMetadata, "hudi checkpoint metadata");
+ IOUtils.closeQuietly(hudiTableInfo, "hudi table info");
+ }
+
+ private void validateWriteStats(String instant, HudiWriteStats writeStats)
{
+ long totalErrorRecords = writeStats.getTotalErrorRecords();
+ if (totalErrorRecords > 0
+ &&
!hudiTableInfo.getFlinkConfig().get(FlinkOptions.IGNORE_FAILED)) {
+ throw new HoodieException(
+ String.format(
+ "Commit Hudi instant %s failed with %s error
records.",
+ instant, totalErrorRecords));
+ }
+ }
+
+ private void abortInstant(String instant) throws IOException {
+ IOException failure = null;
+ try {
+ boolean rolledBack = writeClient.rollback(instant);
+ if (!rolledBack) {
+ throw new IOException("Hudi rollback returned false for
instant " + instant + ".");
+ }
+ } catch (Exception e) {
+ failure =
+ addSuppressed(
+ failure,
+ new IOException("Failed to rollback Hudi instant "
+ instant + ".", e));
+ }
+
+ try {
+ ckpMetadata.abortInstant(instant);
+ } catch (Exception e) {
+ failure =
+ addSuppressed(
+ failure,
+ new IOException(
+ "Failed to abort Hudi checkpoint metadata
for instant "
+ + instant
+ + ".",
+ e));
+ }
+
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private HoodieTimeline getCompletedTimelineCommittedBy(String commitUser)
throws IOException {
+ hudiTableInfo.getMetaClient().reloadActiveTimeline();
+ HoodieTimeline timeline =
+ writeClient
+ .getHoodieTable()
+ .getMetaClient()
+ .getActiveTimeline()
+ .getCommitsAndCompactionTimeline()
+ .filterCompletedInstants();
+ try {
+ return timeline.filter(instant -> isCommittedBy(timeline, instant,
commitUser));
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
+ }
+ }
+
+ private static boolean isCommittedBy(
+ HoodieTimeline timeline, HoodieInstant instant, String commitUser)
{
+ try {
+ HoodieCommitMetadata metadata =
timeline.readCommitMetadata(instant);
+ if (metadata == null) {
+ throw new IOException("Failed to load committed Hudi instant
metadata.");
+ }
+ Map<String, String> extraMetadata = metadata.getExtraMetadata();
+ return extraMetadata != null &&
commitUser.equals(extraMetadata.get(COMMITTER_USER));
+ } catch (IOException e) {
+ // a read failure must not be silently treated as "not committed
by Fluss",
+ // otherwise we may miss an already-tiered snapshot and re-commit
duplicated data
+ throw new UncheckedIOException(
+ "Failed to read Hudi commit metadata for instant " +
instant + ".", e);
+ }
+ }
+
+ private static long parseSnapshotId(String instant) throws IOException {
+ try {
+ return Long.parseLong(instant);
+ } catch (NumberFormatException e) {
+ throw new IOException("Invalid Hudi instant time " + instant +
".", e);
+ }
+ }
+
+ private static IOException addSuppressed(IOException failure, IOException
current) {
+ if (failure == null) {
+ return current;
+ }
+ failure.addSuppressed(current);
+ return failure;
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeTieringFactory.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeTieringFactory.java
index a126cbd17..eaa792721 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeTieringFactory.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeTieringFactory.java
@@ -27,10 +27,10 @@ import org.apache.fluss.lake.writer.LakeWriter;
import org.apache.fluss.lake.writer.WriterInitContext;
import java.io.IOException;
-import java.io.Serializable;
/** Hudi implementation of {@link LakeTieringFactory}. */
-public class HudiLakeTieringFactory implements
LakeTieringFactory<HudiWriteResult, Serializable> {
+public class HudiLakeTieringFactory
+ implements LakeTieringFactory<HudiWriteResult, HudiCommittable> {
private static final long serialVersionUID = 1L;
@@ -54,14 +54,14 @@ public class HudiLakeTieringFactory implements
LakeTieringFactory<HudiWriteResul
}
@Override
- public LakeCommitter<HudiWriteResult, Serializable> createLakeCommitter(
- CommitterInitContext committerInitContext) {
- throw new UnsupportedOperationException("Hudi lake committer is not
implemented yet.");
+ public LakeCommitter<HudiWriteResult, HudiCommittable> createLakeCommitter(
+ CommitterInitContext committerInitContext) throws IOException {
+ return new HudiLakeCommitter(
+ hudiCatalogProvider, ckpMetadataProvider,
committerInitContext.tablePath());
}
@Override
- public SimpleVersionedSerializer<Serializable> getCommittableSerializer() {
- throw new UnsupportedOperationException(
- "Hudi lake committable serializer is not implemented yet.");
+ public SimpleVersionedSerializer<HudiCommittable>
getCommittableSerializer() {
+ return new HudiCommittableSerializer();
}
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeWriter.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeWriter.java
index f58bd2900..1b9c869a2 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeWriter.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeWriter.java
@@ -36,7 +36,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -91,7 +91,7 @@ public class HudiLakeWriter implements
LakeWriter<HudiWriteResult> {
public HudiWriteResult complete() throws IOException {
try {
Map<String, List<WriteStatus>> writeStatuses =
recordWriter.complete();
- return new HudiWriteResult(writeStatuses, new HashMap<>());
+ return HudiWriteResult.fromWriteStatuses(writeStatuses,
Collections.emptyMap());
} catch (Exception e) {
throw new IOException("Failed to complete Hudi write.", e);
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResult.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResult.java
index 973911554..d6436e2d7 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResult.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResult.java
@@ -19,7 +19,11 @@ package org.apache.fluss.lake.hudi.tiering;
import org.apache.hudi.client.WriteStatus;
+import javax.annotation.Nullable;
+
import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -28,21 +32,49 @@ public class HudiWriteResult implements Serializable {
private static final long serialVersionUID = 1L;
- private final Map<String, List<WriteStatus>> writeStatuses;
- private final Map<String, List<WriteStatus>> compactionWriteStatuses;
+ private final Map<String, HudiWriteStats> writeStats;
+ private final Map<String, HudiWriteStats> compactionWriteStats;
public HudiWriteResult(
+ Map<String, HudiWriteStats> writeStats,
+ @Nullable Map<String, HudiWriteStats> compactionWriteStats) {
+ this.writeStats = copyWriteStats(writeStats);
+ this.compactionWriteStats = copyWriteStats(compactionWriteStats);
+ }
+
+ public static HudiWriteResult fromWriteStatuses(
Map<String, List<WriteStatus>> writeStatuses,
- Map<String, List<WriteStatus>> compactionWriteStatuses) {
- this.writeStatuses = writeStatuses;
- this.compactionWriteStatuses = compactionWriteStatuses;
+ @Nullable Map<String, List<WriteStatus>> compactionWriteStatuses) {
+ return new HudiWriteResult(
+ toWriteStats(writeStatuses),
toWriteStats(compactionWriteStatuses));
+ }
+
+ public Map<String, HudiWriteStats> getWriteStats() {
+ return writeStats;
+ }
+
+ public Map<String, HudiWriteStats> getCompactionWriteStats() {
+ return compactionWriteStats;
}
- public Map<String, List<WriteStatus>> getWriteStatuses() {
- return writeStatuses;
+ private static Map<String, HudiWriteStats> copyWriteStats(
+ @Nullable Map<String, HudiWriteStats> statsByInstant) {
+ if (statsByInstant == null || statsByInstant.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(new HashMap<>(statsByInstant));
}
- public Map<String, List<WriteStatus>> getCompactionWriteStatuses() {
- return compactionWriteStatuses;
+ private static Map<String, HudiWriteStats> toWriteStats(
+ @Nullable Map<String, List<WriteStatus>> statusesByInstant) {
+ if (statusesByInstant == null || statusesByInstant.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ Map<String, HudiWriteStats> statsByInstant = new HashMap<>();
+ for (Map.Entry<String, List<WriteStatus>> entry :
statusesByInstant.entrySet()) {
+ statsByInstant.put(entry.getKey(),
HudiWriteStats.fromWriteStatuses(entry.getValue()));
+ }
+ return statsByInstant;
}
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializer.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializer.java
index 57d60fa9b..e80d398ab 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializer.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializer.java
@@ -18,16 +18,12 @@
package org.apache.fluss.lake.hudi.tiering;
import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
-import org.apache.fluss.utils.InstantiationUtils;
-
-import org.apache.hudi.client.WriteStatus;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.util.List;
import java.util.Map;
/** Serializer for {@link HudiWriteResult}. */
@@ -42,17 +38,10 @@ public class HudiWriteResultSerializer implements
SimpleVersionedSerializer<Hudi
@Override
public byte[] serialize(HudiWriteResult hudiWriteResult) throws
IOException {
- byte[] writeResultBytes =
-
InstantiationUtils.serializeObject(hudiWriteResult.getWriteStatuses());
- byte[] compactionWriteResultBytes =
-
InstantiationUtils.serializeObject(hudiWriteResult.getCompactionWriteStatuses());
-
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(baos)) {
- dos.writeInt(writeResultBytes.length);
- dos.write(writeResultBytes);
- dos.writeInt(compactionWriteResultBytes.length);
- dos.write(compactionWriteResultBytes);
+ HudiWriteStatsSerde.writeStatsMap(dos,
hudiWriteResult.getWriteStats());
+ HudiWriteStatsSerde.writeStatsMap(dos,
hudiWriteResult.getCompactionWriteStats());
return baos.toByteArray();
}
}
@@ -67,40 +56,15 @@ public class HudiWriteResultSerializer implements
SimpleVersionedSerializer<Hudi
+ CURRENT_VERSION);
}
- Map<String, List<WriteStatus>> writeResult;
- Map<String, List<WriteStatus>> compactionWriteResult;
+ Map<String, HudiWriteStats> writeStats;
+ Map<String, HudiWriteStats> compactionWriteStats;
try (DataInputStream dis = new DataInputStream(new
ByteArrayInputStream(serialized))) {
- byte[] writeResultBytes = readBytes(dis, "WriteResult");
- writeResult =
- InstantiationUtils.deserializeObject(
- writeResultBytes, getClass().getClassLoader());
-
- byte[] compactionWriteResultBytes = readBytes(dis,
"CompactionWriteResult");
- compactionWriteResult =
- InstantiationUtils.deserializeObject(
- compactionWriteResultBytes,
getClass().getClassLoader());
+ writeStats = HudiWriteStatsSerde.readStatsMap(dis, "WriteStats");
+ compactionWriteStats = HudiWriteStatsSerde.readStatsMap(dis,
"CompactionWriteStats");
if (dis.available() > 0) {
throw new IOException("Corrupted serialization: trailing bytes
" + dis.available());
}
- } catch (ClassNotFoundException e) {
- throw new IOException("Couldn't deserialize HudiWriteResult.", e);
- }
- return new HudiWriteResult(writeResult, compactionWriteResult);
- }
-
- private static byte[] readBytes(DataInputStream dis, String field) throws
IOException {
- int length = dis.readInt();
- validateLength(length, dis.available(), field);
- byte[] bytes = new byte[length];
- dis.readFully(bytes);
- return bytes;
- }
-
- private static void validateLength(int length, int remainingLength, String
field)
- throws IOException {
- if (length < 0 || length > remainingLength) {
- throw new IOException(
- "Corrupted serialization: invalid " + field + " length " +
length);
}
+ return new HudiWriteResult(writeStats, compactionWriteStats);
}
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteStats.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteStats.java
new file mode 100644
index 000000000..bcb53b204
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteStats.java
@@ -0,0 +1,119 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieWriteStat;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Hudi write stats that are sufficient for committing one Hudi instant. */
+public class HudiWriteStats implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<HoodieWriteStat> writeStats;
+ private final long totalErrorRecords;
+
+ public HudiWriteStats(List<HoodieWriteStat> writeStats, long
totalErrorRecords) {
+ checkArgument(
+ totalErrorRecords >= 0,
+ "Hudi total error records must not be negative, but is %s.",
+ totalErrorRecords);
+ this.writeStats = copyWriteStats(writeStats);
+ this.totalErrorRecords = totalErrorRecords;
+ }
+
+ public static HudiWriteStats fromWriteStatuses(List<WriteStatus>
writeStatuses) {
+ checkNotNull(writeStatuses, "Hudi write statuses must not be null.");
+
+ List<HoodieWriteStat> stats = new ArrayList<>(writeStatuses.size());
+ long totalErrorRecords = 0L;
+ for (WriteStatus writeStatus : writeStatuses) {
+ checkNotNull(writeStatus, "Hudi write status must not be null.");
+ stats.add(checkNotNull(writeStatus.getStat(), "Hudi write stat
must not be null."));
+ totalErrorRecords += writeStatus.getTotalErrorRecords();
+ }
+ return new HudiWriteStats(stats, totalErrorRecords);
+ }
+
+ public List<HoodieWriteStat> getWriteStats() {
+ return writeStats;
+ }
+
+ public long getTotalErrorRecords() {
+ return totalErrorRecords;
+ }
+
+ public HudiWriteStats merge(HudiWriteStats other) {
+ checkNotNull(other, "Hudi write stats to merge must not be null.");
+
+ List<HoodieWriteStat> mergedStats =
+ new ArrayList<>(writeStats.size() + other.writeStats.size());
+ mergedStats.addAll(writeStats);
+ mergedStats.addAll(other.writeStats);
+ return new HudiWriteStats(mergedStats, totalErrorRecords +
other.totalErrorRecords);
+ }
+
+ private static List<HoodieWriteStat> copyWriteStats(List<HoodieWriteStat>
writeStats) {
+ if (writeStats == null || writeStats.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<HoodieWriteStat> copiedStats = new ArrayList<>(writeStats.size());
+ for (HoodieWriteStat writeStat : writeStats) {
+ copiedStats.add(checkNotNull(writeStat, "Hudi write stat must not
be null."));
+ }
+ return Collections.unmodifiableList(copiedStats);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof HudiWriteStats)) {
+ return false;
+ }
+ HudiWriteStats that = (HudiWriteStats) o;
+ return totalErrorRecords == that.totalErrorRecords
+ && Objects.equals(writeStats, that.writeStats);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(writeStats, totalErrorRecords);
+ }
+
+ @Override
+ public String toString() {
+ return "HudiWriteStats{"
+ + "writeStats="
+ + writeStats
+ + ", totalErrorRecords="
+ + totalErrorRecords
+ + '}';
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteStatsSerde.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteStatsSerde.java
new file mode 100644
index 000000000..33059fc00
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteStatsSerde.java
@@ -0,0 +1,121 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Utilities to serialize Hudi write stats without Java object serialization.
*/
+final class HudiWriteStatsSerde {
+
+ private HudiWriteStatsSerde() {}
+
+ static void writeStatsMap(DataOutputStream dos, Map<String,
HudiWriteStats> statsByInstant)
+ throws IOException {
+ dos.writeInt(statsByInstant.size());
+ for (Map.Entry<String, HudiWriteStats> entry :
statsByInstant.entrySet()) {
+ writeString(dos, entry.getKey());
+ HudiWriteStats writeStats = entry.getValue();
+ dos.writeLong(writeStats.getTotalErrorRecords());
+ writeString(dos, toCommitMetadataJson(writeStats.getWriteStats()));
+ }
+ }
+
+ static Map<String, HudiWriteStats> readStatsMap(DataInputStream dis,
String field)
+ throws IOException {
+ int mapSize = dis.readInt();
+ if (mapSize < 0 || mapSize > dis.available()) {
+ throw new IOException("Corrupted serialization: invalid " + field
+ " size " + mapSize);
+ }
+
+ Map<String, HudiWriteStats> statsByInstant = new LinkedHashMap<>();
+ for (int i = 0; i < mapSize; i++) {
+ String instant = readString(dis, field + " instant");
+ long totalErrorRecords = dis.readLong();
+ if (totalErrorRecords < 0) {
+ throw new IOException(
+ "Corrupted serialization: invalid "
+ + field
+ + " total error records "
+ + totalErrorRecords);
+ }
+ String commitMetadataJson = readString(dis, field + " write
stats");
+ statsByInstant.put(
+ instant,
+ new HudiWriteStats(
+ fromCommitMetadataJson(commitMetadataJson),
totalErrorRecords));
+ }
+ return statsByInstant;
+ }
+
+ private static String toCommitMetadataJson(List<HoodieWriteStat>
writeStats)
+ throws IOException {
+ HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata();
+ for (HoodieWriteStat writeStat : writeStats) {
+ commitMetadata.addWriteStat(writeStat.getPartitionPath(),
writeStat);
+ }
+ return commitMetadata.toJsonString();
+ }
+
+ private static List<HoodieWriteStat> fromCommitMetadataJson(String
commitMetadataJson)
+ throws IOException {
+ try {
+ HoodieCommitMetadata commitMetadata =
+ HoodieCommitMetadata.fromJsonString(
+ commitMetadataJson, HoodieCommitMetadata.class);
+ if (commitMetadata == null) {
+ throw new IOException("Failed to deserialize Hudi commit
metadata.");
+ }
+ return commitMetadata.getWriteStats();
+ } catch (IOException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IOException("Failed to deserialize Hudi commit
metadata.", e);
+ }
+ }
+
+ private static void writeString(DataOutputStream dos, String value) throws
IOException {
+ byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+ dos.writeInt(bytes.length);
+ dos.write(bytes);
+ }
+
+ private static String readString(DataInputStream dis, String field) throws
IOException {
+ int length = dis.readInt();
+ validateLength(length, dis.available(), field);
+ byte[] bytes = new byte[length];
+ dis.readFully(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ private static void validateLength(int length, int remainingLength, String
field)
+ throws IOException {
+ if (length < 0 || length > remainingLength) {
+ throw new IOException(
+ "Corrupted serialization: invalid " + field + " length " +
length);
+ }
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java
index e4a0e2d06..c730ae937 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java
@@ -91,9 +91,7 @@ public abstract class RecordWriter implements AutoCloseable {
public void setRecordLocation(HoodieFlinkInternalRow internalRow,
Configuration configuration) {
String partition = internalRow.getPartitionPath();
- if (!configuration
- .getString(FlinkOptions.OPERATION)
- .equals(WriteOperationType.INSERT.value())) {
+ if
(!configuration.get(FlinkOptions.OPERATION).equals(WriteOperationType.INSERT.value()))
{
bootstrapIndexIfNeed(partition);
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBuffer.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBuffer.java
index 3a99ede2a..0f03dd782 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBuffer.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBuffer.java
@@ -179,8 +179,7 @@ public class RecordWriteBuffer {
ValidationUtils.checkState(!bucket.isEmpty(), "Data bucket to flush
has no records.");
List<WriteStatus> writeStatus =
writeStatuses.getOrDefault(instantTime, new ArrayList<>());
- writeStatus.addAll(
- writeRecords(instantTime,
config.getString(FlinkOptions.OPERATION), bucket));
+ writeStatus.addAll(writeRecords(instantTime,
config.get(FlinkOptions.OPERATION), bucket));
writeStatuses.put(instantTime, writeStatus);
return true;
}
@@ -191,7 +190,7 @@ public class RecordWriteBuffer {
hudiTableInfo.getRowType(),
MemorySegmentPoolFactory.createMemorySegmentPool(config)),
bucketInfo,
- config.getDouble(FlinkOptions.WRITE_BATCH_SIZE));
+ config.get(FlinkOptions.WRITE_BATCH_SIZE));
}
protected List<WriteStatus> writeRecords(
@@ -361,11 +360,9 @@ public class RecordWriteBuffer {
TotalSizeTracer(Configuration conf) {
long mergeReaderMem = 100;
- long mergeMapMaxMem =
conf.getInteger(FlinkOptions.WRITE_MERGE_MAX_MEMORY);
+ long mergeMapMaxMem =
conf.get(FlinkOptions.WRITE_MERGE_MAX_MEMORY);
this.maxBufferSize =
- (conf.getDouble(FlinkOptions.WRITE_TASK_MAX_SIZE)
- - mergeReaderMem
- - mergeMapMaxMem)
+ (conf.get(FlinkOptions.WRITE_TASK_MAX_SIZE) -
mergeReaderMem - mergeMapMaxMem)
* 1024
* 1024;
ValidationUtils.checkState(
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataFactory.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataFactory.java
index 40408b1cc..225cc7262 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataFactory.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataFactory.java
@@ -31,10 +31,9 @@ public class CkpMetadataFactory {
public static CkpMetadata getCkpMetadata(Configuration conf) {
FileSystem fs =
HadoopFSUtils.getFs(
- conf.getString(FlinkOptions.PATH),
- HadoopConfigurations.getHadoopConf(conf));
- String basePath = conf.getString(FlinkOptions.PATH);
- String uniqueId = conf.getString(FlinkOptions.WRITE_CLIENT_ID);
+ conf.get(FlinkOptions.PATH),
HadoopConfigurations.getHadoopConf(conf));
+ String basePath = conf.get(FlinkOptions.PATH);
+ String uniqueId = conf.get(FlinkOptions.WRITE_CLIENT_ID);
return new CkpMetadata(fs, basePath, uniqueId);
}
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeStorageTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeStorageTest.java
new file mode 100644
index 000000000..2514f2289
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeStorageTest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.lake.hudi;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.hudi.tiering.HudiLakeTieringFactory;
+import org.apache.fluss.lake.writer.LakeTieringFactory;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link HudiLakeStorage}. */
+class HudiLakeStorageTest {
+
+ @Test
+ void testCreateLakeTieringFactory() {
+ HudiLakeStorage lakeStorage = new HudiLakeStorage(new Configuration());
+
+ LakeTieringFactory<?, ?> lakeTieringFactory =
lakeStorage.createLakeTieringFactory();
+
+
assertThat(lakeTieringFactory).isInstanceOf(HudiLakeTieringFactory.class);
+ assertThat(lakeTieringFactory.getWriteResultSerializer()).isNotNull();
+ assertThat(lakeTieringFactory.getCommittableSerializer()).isNotNull();
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
new file mode 100644
index 000000000..14892fff0
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
@@ -0,0 +1,539 @@
+/*
+ * 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.lake.hudi.testutils;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.writer.AppendWriter;
+import org.apache.fluss.client.table.writer.TableWriter;
+import org.apache.fluss.client.table.writer.UpsertWriter;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.flink.tiering.LakeTieringJobBuilder;
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.lake.hudi.source.UnifiedHudiTableReader;
+import org.apache.fluss.lake.hudi.utils.HudiTableInfo;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.server.replica.Replica;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.data.lake.LakeTable;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.data.RowData;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.index.bucket.BucketIdentifier;
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.table.catalog.HoodieCatalog;
+import org.apache.hudi.table.format.InternalSchemaManager;
+import org.apache.hudi.util.StreamerUtil;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.io.Closeable;
+import java.nio.file.Files;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static
org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
+import static
org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test base for tiering to Hudi by Flink. */
+public abstract class FlinkHudiTieringTestBase {
+
+ protected static final String DEFAULT_DB = "fluss";
+ protected static final String CATALOG_NAME = "testcatalog";
+
+ protected StreamExecutionEnvironment execEnv;
+
+ protected static Connection conn;
+ protected static Admin admin;
+ protected static Configuration clientConf;
+ protected static String warehousePath;
+ protected static Catalog hudiCatalog;
+
+ protected static Configuration initConfig() {
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1))
+ .set(ConfigOptions.KV_MAX_RETAINED_SNAPSHOTS,
Integer.MAX_VALUE);
+ conf.set(ConfigOptions.SERVER_DATA_DISK_WRITE_LIMIT_RATIO, 1.0);
+ conf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.HUDI);
+ conf.setString("datalake.hudi.mode", "dfs");
+
+ try {
+ warehousePath =
Files.createTempDirectory("fluss-testing-hudi-tiered").toString();
+ } catch (Exception e) {
+ throw new FlussRuntimeException("Failed to create Hudi warehouse
path", e);
+ }
+ conf.setString("datalake.hudi.catalog.path", warehousePath);
+ return conf;
+ }
+
+ protected static void beforeAll(Configuration conf) {
+ clientConf = conf;
+ conn = ConnectionFactory.createConnection(clientConf);
+ admin = conn.getAdmin();
+ hudiCatalog = getHudiCatalog();
+ hudiCatalog.open();
+ }
+
+ @AfterAll
+ static void afterAll() throws Exception {
+ if (admin != null) {
+ admin.close();
+ admin = null;
+ }
+ if (conn != null) {
+ conn.close();
+ conn = null;
+ }
+ if (hudiCatalog instanceof Closeable) {
+ ((Closeable) hudiCatalog).close();
+ hudiCatalog = null;
+ }
+ }
+
+ @BeforeEach
+ public void beforeEach() {
+ execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ execEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
+ execEnv.setParallelism(2);
+ }
+
+ protected abstract FlussClusterExtension getFlussClusterExtension();
+
+ protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv)
throws Exception {
+ Configuration flussConfig = new Configuration(clientConf);
+ flussConfig.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500L));
+ return LakeTieringJobBuilder.newBuilder(
+ execEnv,
+ flussConfig,
+ Configuration.fromMap(getHudiCatalogConf()),
+ new Configuration(),
+ DataLakeFormat.HUDI.toString())
+ .build();
+ }
+
+ protected static Map<String, String> getHudiCatalogConf() {
+ Map<String, String> hudiConf = new HashMap<>();
+ hudiConf.put("mode", "dfs");
+ hudiConf.put("catalog.path", warehousePath);
+ return hudiConf;
+ }
+
+ protected static Catalog getHudiCatalog() {
+ return new HoodieCatalog(
+ CATALOG_NAME,
+
org.apache.flink.configuration.Configuration.fromMap(getHudiCatalogConf()));
+ }
+
+ protected long createPkTable(
+ TablePath tablePath, int bucketNum, boolean enableAutoCompaction,
Schema schema)
+ throws Exception {
+ TableDescriptor.Builder tableBuilder =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(),
"true")
+ .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS,
Duration.ofMillis(500))
+ .customProperty("hudi.precombine.field", "f_time");
+
+ if (enableAutoCompaction) {
+
tableBuilder.property(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
"true");
+ }
+ return createTable(tablePath, tableBuilder.build());
+ }
+
+ protected long createLogTable(
+ TablePath tablePath, int bucketNum, boolean enableAutoCompaction,
Schema schema)
+ throws Exception {
+ TableDescriptor.Builder tableBuilder =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum, "f_int")
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(),
"true")
+ .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS,
Duration.ofMillis(500))
+ .customProperty("hudi.precombine.field", "f_str")
+ .customProperty("hudi." +
FlinkOptions.RECORD_KEY_FIELD.key(), "f_int");
+
+ if (enableAutoCompaction) {
+
tableBuilder.property(ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
"true");
+ }
+ return createTable(tablePath, tableBuilder.build());
+ }
+
+ protected long createTable(TablePath tablePath, TableDescriptor
tableDescriptor)
+ throws Exception {
+ admin.createTable(tablePath, tableDescriptor, true).get();
+ return admin.getTableInfo(tablePath).get().getTableId();
+ }
+
+ protected void waitUntilSnapshot(long tableId, int bucketNum) {
+ List<TableBucket> buckets = new ArrayList<>();
+ for (int i = 0; i < bucketNum; i++) {
+ buckets.add(new TableBucket(tableId, i));
+ }
+ getFlussClusterExtension().triggerAndWaitSnapshots(buckets);
+ }
+
+ protected void writeRows(TablePath tablePath, List<InternalRow> rows,
boolean append)
+ throws Exception {
+ try (Table table = conn.getTable(tablePath)) {
+ TableWriter tableWriter;
+ if (append) {
+ tableWriter = table.newAppend().createWriter();
+ } else {
+ tableWriter = table.newUpsert().createWriter();
+ }
+ for (InternalRow row : rows) {
+ if (tableWriter instanceof AppendWriter) {
+ ((AppendWriter) tableWriter).append(row);
+ } else {
+ ((UpsertWriter) tableWriter).upsert(row);
+ }
+ }
+ tableWriter.flush();
+ }
+ }
+
+ protected Map<String, List<InternalRow>> writeRowsIntoPartitionedTable(
+ TablePath tablePath,
+ TableDescriptor tableDescriptor,
+ Map<Long, String> partitionNameByIds)
+ throws Exception {
+ List<InternalRow> rows = new ArrayList<>();
+ Map<String, List<InternalRow>> writtenRowsByPartition = new
HashMap<>();
+ for (String partitionName : partitionNameByIds.values()) {
+ List<InternalRow> partitionRows =
+ Arrays.asList(
+ row(11, "v1", partitionName),
+ row(12, "v2", partitionName),
+ row(13, "v3", partitionName));
+ rows.addAll(partitionRows);
+ writtenRowsByPartition.put(partitionName, partitionRows);
+ }
+
+ writeRows(tablePath, rows, !tableDescriptor.hasPrimaryKey());
+ return writtenRowsByPartition;
+ }
+
+ /**
+ * Waits until the default number of partitions is created, and returns
the partition id to
+ * partition name mapping.
+ */
+ public Map<Long, String> waitUntilPartitions(TablePath tablePath) {
+ return waitUntilPartitions(
+ getFlussClusterExtension().getZooKeeperClient(),
+ tablePath,
+
ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.defaultValue());
+ }
+
+ /**
+ * Waits until the default number of partitions is created, and returns
the partition id to
+ * partition name mapping.
+ */
+ public static Map<Long, String> waitUntilPartitions(
+ ZooKeeperClient zooKeeperClient, TablePath tablePath) {
+ return waitUntilPartitions(
+ zooKeeperClient,
+ tablePath,
+
ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE.defaultValue());
+ }
+
+ /**
+ * Waits until the expected number of partitions is created, and returns
the partition id to
+ * partition name mapping.
+ */
+ public static Map<Long, String> waitUntilPartitions(
+ ZooKeeperClient zooKeeperClient, TablePath tablePath, int
expectPartitions) {
+ return waitValue(
+ () -> {
+ Map<Long, String> gotPartitions =
+ zooKeeperClient.getPartitionIdAndNames(tablePath);
+ return expectPartitions == gotPartitions.size()
+ ? Optional.of(gotPartitions)
+ : Optional.empty();
+ },
+ Duration.ofMinutes(1),
+ String.format("expect %d table partition has not been
created", expectPartitions));
+ }
+
+ protected Replica getLeaderReplica(TableBucket tableBucket) {
+ return getFlussClusterExtension().waitAndGetLeaderReplica(tableBucket);
+ }
+
+ protected void assertReplicaStatus(TableBucket tb, long
expectedLogEndOffset) {
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ Replica replica = getLeaderReplica(tb);
+ assertThat(replica.getLogTablet().getLakeTableSnapshotId())
+ .isGreaterThanOrEqualTo(0);
+
assertThat(replica.getLakeLogEndOffset()).isEqualTo(expectedLogEndOffset);
+ });
+ }
+
+ protected void waitUntilBucketSynced(TableBucket tb) {
+ waitUntil(
+ () ->
getLeaderReplica(tb).getLogTablet().getLakeTableSnapshotId() >= 0,
+ Duration.ofMinutes(2),
+ "bucket " + tb + " not synced");
+ }
+
+ protected void checkDataInHudiMORTable(
+ TablePath tablePath, String partition, List<InternalRow>
expectedRows, int bucket)
+ throws Exception {
+ List<String> expectedRecords = new ArrayList<>();
+ for (InternalRow row : expectedRows) {
+ expectedRecords.add(formatMORRow(row));
+ }
+
+ List<String> actualRecords =
+ collectHudiRows(
+ tablePath,
+ partition,
+ bucket,
+ record ->
+ record.getBoolean(5)
+ + ","
+ + record.getInt(6)
+ + ","
+ + record.getLong(7)
+ + ","
+ + record.getFloat(8)
+ + ","
+ + record.getDouble(9)
+ + ","
+ + record.getString(10).toString()
+ + ","
+ + record.getDecimal(11, 5,
2).toBigDecimal().toPlainString()
+ + ","
+ + record.getDecimal(12, 20, 0)
+ .toBigDecimal()
+ .toPlainString());
+
+
assertThat(actualRecords).containsExactlyInAnyOrderElementsOf(expectedRecords);
+ }
+
+ protected void checkDataInHudiMORPartitionTable(
+ TablePath tablePath, String partition, List<InternalRow>
expectedRows, int bucket)
+ throws Exception {
+ List<String> expectedRecords = new ArrayList<>();
+ for (InternalRow row : expectedRows) {
+ expectedRecords.add(
+ row.getInt(0)
+ + ","
+ + row.getString(1).toString()
+ + ","
+ + row.getString(2).toString());
+ }
+
+ List<String> actualRecords =
+ collectHudiRows(
+ tablePath,
+ partition,
+ bucket,
+ record ->
+ record.getInt(5)
+ + ","
+ + record.getString(6).toString()
+ + ","
+ + record.getString(7).toString());
+
+
assertThat(actualRecords).containsExactlyInAnyOrderElementsOf(expectedRecords);
+ }
+
+ protected void checkDataInHudiCOWTable(
+ TablePath tablePath,
+ String partition,
+ List<InternalRow> expectedRows,
+ long startingOffset,
+ int bucket)
+ throws Exception {
+ List<String> expectedRecords = new ArrayList<>();
+ Iterator<InternalRow> flussRowIterator = expectedRows.iterator();
+ while (flussRowIterator.hasNext()) {
+ InternalRow flussRow = flussRowIterator.next();
+ expectedRecords.add(
+ flussRow.getInt(0)
+ + ","
+ + flussRow.getString(1).toString()
+ + ","
+ + startingOffset++);
+ }
+
+ List<String> actualRecords =
+ collectHudiRows(
+ tablePath,
+ partition,
+ bucket,
+ record ->
+ record.getInt(5)
+ + ","
+ + record.getString(6).toString()
+ + ","
+ + record.getLong(8));
+
+
assertThat(actualRecords).containsExactlyInAnyOrderElementsOf(expectedRecords);
+ assertThat(flussRowIterator.hasNext()).isFalse();
+ }
+
+ protected void checkFlussOffsetsInSnapshot(
+ TablePath tablePath, Map<TableBucket, Long> expectedOffsets)
throws Exception {
+ try (HudiTableInfo hudiTableInfo =
+ HudiTableInfo.create(tablePath,
Configuration.fromMap(getHudiCatalogConf()))) {
+ HoodieTableMetaClient metaClient = hudiTableInfo.getMetaClient();
+ metaClient.reloadActiveTimeline();
+ HoodieTimeline timeline =
+ metaClient
+ .getActiveTimeline()
+ .getCommitsAndCompactionTimeline()
+ .filterCompletedInstants();
+ Optional<HoodieInstant> latestInstant =
+
timeline.getReverseOrderedInstantsByCompletionTime().findFirst();
+ assertThat(latestInstant).isPresent();
+
+ HoodieCommitMetadata metadata =
timeline.readCommitMetadata(latestInstant.get());
+ Map<String, String> extraMetadata = metadata.getExtraMetadata();
+ assertThat(extraMetadata).isNotNull();
+ String offsetFile =
extraMetadata.get(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY);
+ assertThat(offsetFile).isNotNull();
+
+ Map<TableBucket, Long> recordedOffsets =
+ new LakeTable(
+ new LakeTable.LakeSnapshotMetadata(
+ -1, new FsPath(offsetFile), null))
+ .getOrReadLatestTableSnapshot()
+ .getBucketLogEndOffset();
+ assertThat(recordedOffsets).isEqualTo(expectedOffsets);
+ }
+ }
+
+ private static String formatMORRow(InternalRow row) {
+ return row.getBoolean(0)
+ + ","
+ + row.getInt(1)
+ + ","
+ + row.getLong(2)
+ + ","
+ + row.getFloat(3)
+ + ","
+ + row.getDouble(4)
+ + ","
+ + row.getString(5).toString()
+ + ","
+ + row.getDecimal(6, 5, 2).toBigDecimal().toPlainString()
+ + ","
+ + row.getDecimal(7, 20, 0).toBigDecimal().toPlainString();
+ }
+
+ private List<String> collectHudiRows(
+ TablePath tablePath, String partition, int bucket,
HudiRowFormatter formatter)
+ throws Exception {
+ try (HudiTableInfo hudiTableInfo =
+ HudiTableInfo.create(tablePath,
Configuration.fromMap(getHudiCatalogConf()))) {
+ org.apache.hudi.org.apache.avro.Schema avroSchema =
+
StreamerUtil.getTableAvroSchema(hudiTableInfo.getMetaClient(), true);
+ org.apache.flink.configuration.Configuration flinkHudiOptions =
+ buildFlinkHudiOptions(tablePath, hudiTableInfo,
avroSchema);
+
+ StorageConfiguration<org.apache.hadoop.conf.Configuration>
hadoopConf =
+ new HadoopStorageConfiguration(
+
HadoopConfigurations.getHadoopConf(flinkHudiOptions));
+ InternalSchemaManager internalSchemaManager =
+ InternalSchemaManager.get(hadoopConf,
hudiTableInfo.getMetaClient());
+ int columnCount = avroSchema.getFields().size();
+
+ List<FileSlice> fileSlices =
+ hudiTableInfo
+ .getFileSystemView()
+ .getLatestFileSlices(partition)
+ .collect(Collectors.toList());
+ List<String> records = new ArrayList<>();
+ for (FileSlice fileSlice : fileSlices) {
+ if
(!fileSlice.getFileId().contains(BucketIdentifier.bucketIdStr(bucket))) {
+ continue;
+ }
+ try (UnifiedHudiTableReader reader =
+ UnifiedHudiTableReader.newBuilder()
+
.withMetaClient(hudiTableInfo.getMetaClient())
+
.withInternalSchemaManager(internalSchemaManager)
+ .withProps(flinkHudiOptions)
+ .withTableSchema(avroSchema)
+ .withSelectedFields(
+ IntStream.range(0,
columnCount).toArray())
+
.withLatestCommitTime(fileSlice.getLatestInstantTime())
+ .build();
+ ClosableIterator<RowData> iterator =
reader.readFileSlice(fileSlice)) {
+ while (iterator.hasNext()) {
+ records.add(formatter.format(iterator.next()));
+ }
+ }
+ }
+ return records;
+ }
+ }
+
+ private org.apache.flink.configuration.Configuration buildFlinkHudiOptions(
+ TablePath tablePath,
+ HudiTableInfo hudiTableInfo,
+ org.apache.hudi.org.apache.avro.Schema avroSchema) {
+ Map<String, String> hudiOptions = new
HashMap<>(hudiTableInfo.getTableOptions());
+ hudiOptions.putAll(getHudiCatalogConf());
+ hudiOptions.put(FlinkOptions.PATH.key(), hudiTableInfo.getBasePath());
+ hudiOptions.put(FlinkOptions.TABLE_NAME.key(),
tablePath.getTableName());
+ hudiOptions.put(FlinkOptions.SOURCE_AVRO_SCHEMA.key(),
avroSchema.toString());
+ return
org.apache.flink.configuration.Configuration.fromMap(hudiOptions);
+ }
+
+ private interface HudiRowFormatter {
+ String format(RowData row);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializerTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializerTest.java
new file mode 100644
index 000000000..52dba1c7b
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiCommittableSerializerTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HudiCommittableSerializer}. */
+class HudiCommittableSerializerTest {
+
+ @Test
+ void testSerializeAndDeserializeCommittable() throws Exception {
+ HudiCommittableSerializer serializer = new HudiCommittableSerializer();
+ HudiCommittable committable =
+ new HudiCommittable(writeStats("20260622000100000"),
Collections.emptyMap());
+
+ HudiCommittable deserialized =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(committable));
+
+
assertThat(deserialized.getWriteStats()).containsOnlyKeys("20260622000100000").hasSize(1);
+ HudiWriteStats writeStats =
deserialized.getWriteStats().get("20260622000100000");
+ assertThat(writeStats.getWriteStats()).hasSize(2);
+ assertThat(writeStats.getWriteStats())
+ .extracting(HoodieWriteStat::getFileId)
+ .containsExactly("file-1", "file-2");
+ assertThat(writeStats.getTotalErrorRecords()).isEqualTo(3L);
+ assertThat(deserialized.getCompactionWriteStats()).isEmpty();
+ }
+
+ @Test
+ void testBuilderAggregatesStatsByInstant() {
+ HudiCommittable committable =
+ HudiCommittable.builder()
+ .addWriteStats(writeStats("20260622000100000"))
+ .addWriteStats(writeStats("20260622000100000"))
+
.addCompactionWriteStats(writeStats("20260622000200000"))
+ .build();
+
+ HudiWriteStats writeStats =
committable.getWriteStats().get("20260622000100000");
+ assertThat(writeStats.getWriteStats()).hasSize(4);
+ assertThat(writeStats.getTotalErrorRecords()).isEqualTo(6L);
+
assertThat(committable.getCompactionWriteStats().get("20260622000200000").getWriteStats())
+ .hasSize(2);
+ }
+
+ @Test
+ void testRejectUnsupportedVersion() {
+ HudiCommittableSerializer serializer = new HudiCommittableSerializer();
+
+ assertThatThrownBy(() ->
serializer.deserialize(serializer.getVersion() + 1, new byte[0]))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported HudiCommittable version");
+ }
+
+ @Test
+ void testRejectCorruptedLength() throws Exception {
+ HudiCommittableSerializer serializer = new HudiCommittableSerializer();
+
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (DataOutputStream dos = new DataOutputStream(baos)) {
+ dos.writeInt(0);
+ dos.writeInt(1);
+ dos.writeInt(1024);
+ }
+
+ assertThatThrownBy(
+ () -> serializer.deserialize(serializer.getVersion(),
baos.toByteArray()))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Corrupted serialization: invalid
CompactionWriteStats");
+ }
+
+ private static Map<String, HudiWriteStats> writeStats(String instant) {
+ Map<String, HudiWriteStats> writeStats = new HashMap<>();
+ writeStats.put(instant, new HudiWriteStats(writeStats(), 3L));
+ return writeStats;
+ }
+
+ private static List<HoodieWriteStat> writeStats() {
+ return Arrays.asList(writeStat("file-1"), writeStat("file-2"));
+ }
+
+ private static HoodieWriteStat writeStat(String fileId) {
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setFileId(fileId);
+ writeStat.setPartitionPath("partition");
+ writeStat.setPath("partition/" + fileId + ".parquet");
+ writeStat.setNumWrites(1L);
+ return writeStat;
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java
new file mode 100644
index 000000000..0f1acc0de
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringITCase.java
@@ -0,0 +1,233 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.fluss.config.AutoPartitionTimeUnit;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.lake.hudi.testutils.FlinkHudiTieringTestBase;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.TypeUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.apache.flink.core.execution.JobClient;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.math.BigDecimal;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT case for tiering tables to Hudi. */
+class HudiTieringITCase extends FlinkHudiTieringTestBase {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setClusterConf(initConfig())
+ .setNumOfTabletServers(3)
+ .build();
+
+ private static final Schema PK_SCHEMA =
+ Schema.newBuilder()
+ .column("f_boolean", DataTypes.BOOLEAN())
+ .column("f_int", DataTypes.INT())
+ .column("f_long", DataTypes.BIGINT())
+ .column("f_float", DataTypes.FLOAT())
+ .column("f_double", DataTypes.DOUBLE())
+ .column("f_string", DataTypes.STRING())
+ .column("f_decimal1", DataTypes.DECIMAL(5, 2))
+ .column("f_decimal2", DataTypes.DECIMAL(20, 0))
+ .column("f_timestamp_ltz1", DataTypes.TIMESTAMP_LTZ(3))
+ .column("f_timestamp_ltz2", DataTypes.TIMESTAMP_LTZ(6))
+ .column("f_timestamp_ntz1", DataTypes.TIMESTAMP(3))
+ .column("f_timestamp_ntz2", DataTypes.TIMESTAMP(6))
+ .column("f_binary", DataTypes.BINARY(4))
+ .column("f_date", DataTypes.DATE())
+ .column("f_time", DataTypes.TIME())
+ .column("f_char", DataTypes.CHAR(3))
+ .column("f_bytes", DataTypes.BYTES())
+ .primaryKey("f_int")
+ .build();
+
+ private static final Schema LOG_SCHEMA =
+ Schema.newBuilder()
+ .column("f_int", DataTypes.INT())
+ .column("f_str", DataTypes.STRING())
+ .build();
+
+ @BeforeAll
+ protected static void beforeAll() {
+
FlinkHudiTieringTestBase.beforeAll(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ }
+
+ @Test
+ void testTiering() throws Exception {
+ TablePath pkTablePath = TablePath.of(DEFAULT_DB, "pkTable");
+ long pkTableId = createPkTable(pkTablePath, 1, false, PK_SCHEMA);
+ TableBucket pkTableBucket = new TableBucket(pkTableId, 0);
+
+ List<InternalRow> rows = Arrays.asList(pkRow(1, "v1"), pkRow(2, "v2"),
pkRow(3, "v3"));
+ writeRows(pkTablePath, rows, false);
+ waitUntilSnapshot(pkTableId, 1);
+
+ execEnv.enableCheckpointing(1000);
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ assertReplicaStatus(pkTableBucket, 3);
+ checkDataInHudiMORTable(pkTablePath, "", rows, 0);
+ checkFlussOffsetsInSnapshot(pkTablePath,
Collections.singletonMap(pkTableBucket, 3L));
+
+ testLogTableTiering();
+
+ rows = Arrays.asList(pkRow(1, "v111"), pkRow(2, "v222"), pkRow(3,
"v333"));
+ writeRows(pkTablePath, rows, false);
+
+ // 3 initial records + 3 delete records + 3 insert records.
+ assertReplicaStatus(pkTableBucket, 9);
+ checkDataInHudiMORTable(pkTablePath, "", rows, 0);
+ checkFlussOffsetsInSnapshot(pkTablePath,
Collections.singletonMap(pkTableBucket, 9L));
+
+ testPartitionedTableTiering();
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ private void testLogTableTiering() throws Exception {
+ TablePath logTablePath = TablePath.of(DEFAULT_DB, "logTable");
+ long logTableId = createLogTable(logTablePath, 1, false, LOG_SCHEMA);
+ TableBucket logTableBucket = new TableBucket(logTableId, 0);
+
+ List<InternalRow> flussRows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ List<InternalRow> rows =
+ Arrays.asList(
+ row(i + 1, "v" + (i + 1)),
+ row(i + 11, "v" + (i + 11)),
+ row(i + 21, "v" + (i + 21)));
+ flussRows.addAll(rows);
+ writeRows(logTablePath, rows, true);
+ }
+
+ assertReplicaStatus(logTableBucket, 30);
+
assertThat(getLeaderReplica(logTableBucket).getLogTablet().getLakeMaxTimestamp())
+ .isGreaterThan(-1);
+ checkDataInHudiCOWTable(logTablePath, "", flussRows, 0, 0);
+ checkFlussOffsetsInSnapshot(logTablePath,
Collections.singletonMap(logTableBucket, 30L));
+ }
+
+ private void testPartitionedTableTiering() throws Exception {
+ TablePath partitionedTablePath = TablePath.of(DEFAULT_DB,
"partitionedTable");
+ Tuple2<Long, TableDescriptor> tableIdAndDescriptor =
+ createPartitionedTable(partitionedTablePath);
+ Map<Long, String> partitionNameByIds =
waitUntilPartitions(partitionedTablePath);
+
+ TableDescriptor partitionedTableDescriptor = tableIdAndDescriptor.f1;
+ Map<String, List<InternalRow>> writtenRowsByPartition =
+ writeRowsIntoPartitionedTable(
+ partitionedTablePath, partitionedTableDescriptor,
partitionNameByIds);
+ long tableId = tableIdAndDescriptor.f0;
+
+ Map<TableBucket, Long> expectedOffsets = new HashMap<>();
+ for (Long partitionId : partitionNameByIds.keySet()) {
+ TableBucket tableBucket = new TableBucket(tableId, partitionId, 0);
+ assertReplicaStatus(tableBucket, 3);
+ expectedOffsets.put(tableBucket, 3L);
+ }
+
+ for (String partitionName : partitionNameByIds.values()) {
+ checkDataInHudiMORPartitionTable(
+ partitionedTablePath,
+ partitionName,
+ writtenRowsByPartition.get(partitionName),
+ 0);
+ }
+ checkFlussOffsetsInSnapshot(partitionedTablePath, expectedOffsets);
+ }
+
+ private Tuple2<Long, TableDescriptor> createPartitionedTable(TablePath
partitionedTablePath)
+ throws Exception {
+ TableDescriptor partitionedTableDescriptor =
+ TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("date", DataTypes.STRING())
+ .build())
+ .distributedBy(1, "id")
+ .partitionedBy("date")
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED,
true)
+ .property(
+ ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT,
+ AutoPartitionTimeUnit.YEAR)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS,
Duration.ofMillis(500))
+ .customProperty("hudi.precombine.field", "date")
+
.customProperty("hudi.hoodie.datasource.write.recordkey.field", "id")
+ .build();
+ return Tuple2.of(
+ createTable(partitionedTablePath, partitionedTableDescriptor),
+ partitionedTableDescriptor);
+ }
+
+ private static InternalRow pkRow(int id, String value) {
+ return row(
+ true,
+ id,
+ id + 400L,
+ 500.1f,
+ 600.0d,
+ value,
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new BigDecimal("1000"), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2026-02-25", DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0", DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10});
+ }
+
+ @Override
+ protected FlussClusterExtension getFlussClusterExtension() {
+ return FLUSS_CLUSTER_EXTENSION;
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringTest.java
new file mode 100644
index 000000000..815daf069
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiTieringTest.java
@@ -0,0 +1,274 @@
+/*
+ * 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.lake.hudi.tiering;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
+import org.apache.fluss.lake.committer.CommitterInitContext;
+import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.hudi.HudiLakeCatalog;
+import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
+import org.apache.fluss.lake.writer.LakeWriter;
+import org.apache.fluss.lake.writer.WriterInitContext;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.record.GenericRecord;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+
+import static
org.apache.fluss.lake.committer.LakeCommitter.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static
org.apache.fluss.lake.writer.LakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for tiering Fluss records to Hudi via {@link HudiLakeTieringFactory}.
*/
+class HudiTieringTest {
+
+ @TempDir private File tempWarehouseDir;
+
+ private Configuration hudiConfig;
+ private HudiLakeCatalog hudiLakeCatalog;
+ private HudiLakeTieringFactory hudiLakeTieringFactory;
+
+ @BeforeEach
+ void beforeEach() {
+ hudiConfig = new Configuration();
+ hudiConfig.setString("catalog.path",
tempWarehouseDir.toURI().toString());
+ hudiConfig.setString("mode", "dfs");
+ hudiLakeCatalog = new HudiLakeCatalog(hudiConfig);
+ hudiLakeTieringFactory = new HudiLakeTieringFactory(hudiConfig);
+ }
+
+ @AfterEach
+ void afterEach() {
+ if (hudiLakeCatalog != null) {
+ hudiLakeCatalog.close();
+ }
+ }
+
+ @Test
+ void testWriteAndCommitLogTable() throws Exception {
+ TablePath tablePath = TablePath.of("hudi",
"test_write_and_commit_log_table");
+ TableDescriptor tableDescriptor = createLogTableDescriptor();
+ hudiLakeCatalog.createTable(
+ tablePath, tableDescriptor, new
TestingLakeCatalogContext(tableDescriptor));
+ TableInfo tableInfo = TableInfo.of(tablePath, 1L, 0, tableDescriptor,
null, 1L, 1L);
+
+ HudiWriteResult writeResult;
+ try (LakeWriter<HudiWriteResult> writer =
+ hudiLakeTieringFactory.createLakeWriter(
+ new TestingWriterInitContext(
+ tablePath, new TableBucket(1L, 0), tableInfo,
0, 0L))) {
+ writer.write(logRecord(0L, 1, "v1"));
+ writer.write(logRecord(1L, 2, "v2"));
+ writeResult = writer.complete();
+ }
+
+ long committedSnapshotId;
+ try (LakeCommitter<HudiWriteResult, HudiCommittable> committer =
+ createLakeCommitter(tablePath, tableInfo)) {
+ assertThat(committer.getMissingLakeSnapshot(null)).isNull();
+
+ HudiCommittable committable =
+
committer.toCommittable(Collections.singletonList(writeResult));
+ committedSnapshotId =
+ committer
+ .commit(
+ committable,
+ Collections.singletonMap(
+
FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
+ "fluss-offset-file"))
+ .getCommittedSnapshotId();
+ }
+
+ try (LakeCommitter<HudiWriteResult, HudiCommittable> committer =
+ createLakeCommitter(tablePath, tableInfo)) {
+ CommittedLakeSnapshot missingLakeSnapshot =
+ committer.getMissingLakeSnapshot(committedSnapshotId - 1);
+
+ assertThat(missingLakeSnapshot).isNotNull();
+
assertThat(missingLakeSnapshot.getLakeSnapshotId()).isEqualTo(committedSnapshotId);
+ assertThat(missingLakeSnapshot.getSnapshotProperties())
+ .containsEntry("commit-user",
FLUSS_LAKE_TIERING_COMMIT_USER)
+ .containsEntry(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
"fluss-offset-file");
+
assertThat(committer.getMissingLakeSnapshot(committedSnapshotId)).isNull();
+ }
+ }
+
+ @Test
+ void testRejectCompactionWriteStatusesOnCommit() throws Exception {
+ TablePath tablePath =
+ TablePath.of("hudi",
"test_reject_compaction_write_statuses_on_commit");
+ TableDescriptor tableDescriptor = createLogTableDescriptor();
+ hudiLakeCatalog.createTable(
+ tablePath, tableDescriptor, new
TestingLakeCatalogContext(tableDescriptor));
+ TableInfo tableInfo = TableInfo.of(tablePath, 1L, 0, tableDescriptor,
null, 1L, 1L);
+
+ try (LakeCommitter<HudiWriteResult, HudiCommittable> committer =
+ createLakeCommitter(tablePath, tableInfo)) {
+ HudiCommittable committable =
+ new HudiCommittable(
+ Collections.emptyMap(),
+ Collections.singletonMap(
+ "20260623000100000",
+ new HudiWriteStats(
+
Collections.singletonList(writeStat()), 0L)));
+
+ assertThatThrownBy(() -> committer.commit(committable,
Collections.emptyMap()))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Hudi compaction write stats are not
supported yet");
+ }
+ }
+
+ private LakeCommitter<HudiWriteResult, HudiCommittable>
createLakeCommitter(
+ TablePath tablePath, TableInfo tableInfo) throws IOException {
+ return hudiLakeTieringFactory.createLakeCommitter(
+ new TestingCommitterInitContext(tablePath, tableInfo));
+ }
+
+ private static TableDescriptor createLogTableDescriptor() {
+ return TableDescriptor.builder()
+ .schema(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .build())
+ .distributedBy(1, "id")
+ .customProperty("hudi." + FlinkOptions.RECORD_KEY_FIELD.key(),
"id")
+ .customProperty("hudi.precombine.field", "name")
+ .build();
+ }
+
+ private static LogRecord logRecord(long offset, int id, String name) {
+ GenericRow row = new GenericRow(2);
+ row.setField(0, id);
+ row.setField(1, BinaryString.fromString(name));
+ return new GenericRecord(offset, 1_000L + offset,
ChangeType.APPEND_ONLY, row);
+ }
+
+ private static HoodieWriteStat writeStat() {
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setFileId("file-1");
+ writeStat.setPartitionPath("");
+ writeStat.setPath("file-1.parquet");
+ return writeStat;
+ }
+
+ private static class TestingWriterInitContext implements WriterInitContext
{
+
+ private final TablePath tablePath;
+ private final TableBucket tableBucket;
+ private final TableInfo tableInfo;
+ private final int splitIndex;
+ private final long tieringRoundTimestamp;
+
+ private TestingWriterInitContext(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ TableInfo tableInfo,
+ int splitIndex,
+ long tieringRoundTimestamp) {
+ this.tablePath = tablePath;
+ this.tableBucket = tableBucket;
+ this.tableInfo = tableInfo;
+ this.splitIndex = splitIndex;
+ this.tieringRoundTimestamp = tieringRoundTimestamp;
+ }
+
+ @Override
+ public TablePath tablePath() {
+ return tablePath;
+ }
+
+ @Override
+ public TableBucket tableBucket() {
+ return tableBucket;
+ }
+
+ @Nullable
+ @Override
+ public String partition() {
+ return null;
+ }
+
+ @Override
+ public TableInfo tableInfo() {
+ return tableInfo;
+ }
+
+ @Override
+ public int splitIndex() {
+ return splitIndex;
+ }
+
+ @Override
+ public long tieringRoundTimestamp() {
+ return tieringRoundTimestamp;
+ }
+ }
+
+ private static class TestingCommitterInitContext implements
CommitterInitContext {
+
+ private final TablePath tablePath;
+ private final TableInfo tableInfo;
+
+ private TestingCommitterInitContext(TablePath tablePath, TableInfo
tableInfo) {
+ this.tablePath = tablePath;
+ this.tableInfo = tableInfo;
+ }
+
+ @Override
+ public TablePath tablePath() {
+ return tablePath;
+ }
+
+ @Override
+ public TableInfo tableInfo() {
+ return tableInfo;
+ }
+
+ @Override
+ public Configuration lakeTieringConfig() {
+ return new Configuration();
+ }
+
+ @Override
+ public Configuration flussClientConfig() {
+ return new Configuration();
+ }
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializerTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializerTest.java
index 33c93d777..428176b4c 100644
---
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializerTest.java
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializerTest.java
@@ -17,15 +17,17 @@
package org.apache.fluss.lake.hudi.tiering;
-import org.apache.fluss.utils.InstantiationUtils;
-
+import org.apache.hudi.common.model.HoodieWriteStat;
import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import static java.util.Collections.singletonList;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -41,8 +43,24 @@ class HudiWriteResultSerializerTest {
HudiWriteResult deserialized =
serializer.deserialize(serializer.getVersion(),
serializer.serialize(writeResult));
- assertThat(deserialized.getWriteStatuses()).isEmpty();
- assertThat(deserialized.getCompactionWriteStatuses()).isEmpty();
+ assertThat(deserialized.getWriteStats()).isEmpty();
+ assertThat(deserialized.getCompactionWriteStats()).isEmpty();
+ }
+
+ @Test
+ void testSerializeAndDeserializeWriteResult() throws Exception {
+ HudiWriteResultSerializer serializer = new HudiWriteResultSerializer();
+ HudiWriteResult writeResult =
+ new HudiWriteResult(
+ writeStats("20260622000100000", "file-1", 2L),
Collections.emptyMap());
+
+ HudiWriteResult deserialized =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(writeResult));
+
+ HudiWriteStats writeStats =
deserialized.getWriteStats().get("20260622000100000");
+ assertThat(writeStats.getWriteStats()).hasSize(1);
+
assertThat(writeStats.getWriteStats().get(0).getFileId()).isEqualTo("file-1");
+ assertThat(writeStats.getTotalErrorRecords()).isEqualTo(2L);
}
@Test
@@ -57,18 +75,29 @@ class HudiWriteResultSerializerTest {
@Test
void testRejectCorruptedLength() throws Exception {
HudiWriteResultSerializer serializer = new HudiWriteResultSerializer();
- byte[] emptyMapBytes =
InstantiationUtils.serializeObject(Collections.emptyMap());
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try (DataOutputStream dos = new DataOutputStream(baos)) {
- dos.writeInt(emptyMapBytes.length);
- dos.write(emptyMapBytes);
- dos.writeInt(emptyMapBytes.length);
+ dos.writeInt(0);
+ dos.writeInt(1);
+ dos.writeInt(1024);
}
assertThatThrownBy(
() -> serializer.deserialize(serializer.getVersion(),
baos.toByteArray()))
.isInstanceOf(IOException.class)
- .hasMessageContaining("Corrupted serialization: invalid
CompactionWriteResult");
+ .hasMessageContaining("Corrupted serialization: invalid
CompactionWriteStats");
+ }
+
+ private static Map<String, HudiWriteStats> writeStats(
+ String instant, String fileId, long totalErrorRecords) {
+ Map<String, HudiWriteStats> writeStats = new HashMap<>();
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setFileId(fileId);
+ writeStat.setPartitionPath("partition");
+ writeStat.setPath("partition/" + fileId + ".parquet");
+ writeStat.setNumWrites(1L);
+ writeStats.put(instant, new HudiWriteStats(singletonList(writeStat),
totalErrorRecords));
+ return writeStats;
}
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/resources/log4j2-test.properties
b/fluss-lake/fluss-lake-hudi/src/test/resources/log4j2-test.properties
index 2100a34e0..edfee0abf 100644
--- a/fluss-lake/fluss-lake-hudi/src/test/resources/log4j2-test.properties
+++ b/fluss-lake/fluss-lake-hudi/src/test/resources/log4j2-test.properties
@@ -23,4 +23,4 @@ appender.testlogger.name=TestLogger
appender.testlogger.type=CONSOLE
appender.testlogger.target=SYSTEM_ERR
appender.testlogger.layout.type=PatternLayout
-appender.testlogger.layout.pattern=%-4r [%t] %-5p %c %x - %m%n
\ No newline at end of file
+appender.testlogger.layout.pattern=%-4r [%t] %-5p %c %x - %m%n
