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 dfbf1a595 [lake/hudi] Introduce Hudi lake writer support for tiering
tables (#3507)
dfbf1a595 is described below
commit dfbf1a5952b94610b6955192cf995a18b7a8bf7a
Author: fhan <[email protected]>
AuthorDate: Mon Jun 22 21:57:16 2026 +0800
[lake/hudi] Introduce Hudi lake writer support for tiering tables (#3507)
---
fluss-lake/fluss-lake-hudi/pom.xml | 6 +
.../lake/hudi/tiering/HudiCatalogProvider.java | 47 +++
.../lake/hudi/tiering/HudiLakeTieringFactory.java | 67 ++++
.../fluss/lake/hudi/tiering/HudiLakeWriter.java | 139 ++++++++
.../fluss/lake/hudi/tiering/HudiWriteResult.java | 48 +++
.../hudi/tiering/HudiWriteResultSerializer.java | 106 ++++++
.../lake/hudi/tiering/HudiWriteTableInfo.java | 285 +++++++++++++++
.../fluss/lake/hudi/tiering/RecordWriter.java | 160 +++++++++
.../hudi/tiering/writer/FlussArrayAsHudiArray.java | 193 ++++++++++
.../hudi/tiering/writer/FlussMapAsHudiMap.java | 58 +++
.../hudi/tiering/writer/FlussRecordAsHudiRow.java | 90 +++++
.../hudi/tiering/writer/FlussRowAsHudiRow.java | 183 ++++++++++
.../hudi/tiering/writer/HudiRecordConverter.java | 44 +++
.../lake/hudi/tiering/writer/HudiRecordWriter.java | 61 ++++
.../hudi/tiering/writer/RecordWriteBuffer.java | 392 +++++++++++++++++++++
.../fluss/lake/hudi/utils/HudiConversions.java | 17 +
.../fluss/lake/hudi/utils/meta/CkpMessage.java | 97 +++++
.../fluss/lake/hudi/utils/meta/CkpMetadata.java | 209 +++++++++++
.../lake/hudi/utils/meta/CkpMetadataFactory.java | 40 +++
.../lake/hudi/utils/meta/CkpMetadataProvider.java | 65 ++++
.../tiering/HudiWriteResultSerializerTest.java | 74 ++++
.../tiering/writer/FlussRecordAsHudiRowTest.java | 119 +++++++
.../hudi/tiering/writer/RecordWriteBufferTest.java | 198 +++++++++++
.../lake/hudi/utils/meta/CkpMetadataTest.java | 60 ++++
24 files changed, 2758 insertions(+)
diff --git a/fluss-lake/fluss-lake-hudi/pom.xml
b/fluss-lake/fluss-lake-hudi/pom.xml
index 0f2bc76d3..64f1ca2c1 100644
--- a/fluss-lake/fluss-lake-hudi/pom.xml
+++ b/fluss-lake/fluss-lake-hudi/pom.xml
@@ -156,6 +156,12 @@
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
<!-- test dependency -->
<dependency>
<groupId>org.apache.fluss</groupId>
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCatalogProvider.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCatalogProvider.java
new file mode 100644
index 000000000..377e9475c
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiCatalogProvider.java
@@ -0,0 +1,47 @@
+/*
+ * 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.hudi.utils.catalog.HudiCatalogUtils;
+
+import org.apache.flink.table.catalog.Catalog;
+
+import java.io.Serializable;
+
+/** Serializable provider for creating Hudi catalogs in tiering subtasks. */
+public class HudiCatalogProvider implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Configuration hudiConfig;
+
+ public HudiCatalogProvider(Configuration hudiConfig) {
+ this.hudiConfig = hudiConfig;
+ }
+
+ public Configuration getHudiConfig() {
+ return hudiConfig;
+ }
+
+ public Catalog get() {
+ Catalog hudiCatalog = HudiCatalogUtils.createHudiCatalog(hudiConfig);
+ hudiCatalog.open();
+ return hudiCatalog;
+ }
+}
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
new file mode 100644
index 000000000..a126cbd17
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeTieringFactory.java
@@ -0,0 +1,67 @@
+/*
+ * 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.CommitterInitContext;
+import org.apache.fluss.lake.committer.LakeCommitter;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadataProvider;
+import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
+import org.apache.fluss.lake.writer.LakeTieringFactory;
+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> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final HudiCatalogProvider hudiCatalogProvider;
+ private final CkpMetadataProvider ckpMetadataProvider;
+
+ public HudiLakeTieringFactory(Configuration hudiConfig) {
+ this.hudiCatalogProvider = new HudiCatalogProvider(hudiConfig);
+ this.ckpMetadataProvider = new CkpMetadataProvider();
+ }
+
+ @Override
+ public LakeWriter<HudiWriteResult> createLakeWriter(WriterInitContext
writerInitContext)
+ throws IOException {
+ return new HudiLakeWriter(hudiCatalogProvider, ckpMetadataProvider,
writerInitContext);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<HudiWriteResult>
getWriteResultSerializer() {
+ return new HudiWriteResultSerializer();
+ }
+
+ @Override
+ public LakeCommitter<HudiWriteResult, Serializable> createLakeCommitter(
+ CommitterInitContext committerInitContext) {
+ throw new UnsupportedOperationException("Hudi lake committer is not
implemented yet.");
+ }
+
+ @Override
+ public SimpleVersionedSerializer<Serializable> getCommittableSerializer() {
+ throw new UnsupportedOperationException(
+ "Hudi lake committable serializer is not implemented yet.");
+ }
+}
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
new file mode 100644
index 000000000..f58bd2900
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeWriter.java
@@ -0,0 +1,139 @@
+/*
+ * 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.hudi.tiering.writer.HudiRecordWriter;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadataProvider;
+import org.apache.fluss.lake.writer.LakeWriter;
+import org.apache.fluss.lake.writer.WriterInitContext;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.record.LogRecord;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.CommitUtils;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.fluss.lake.writer.WriterInitContext.UNKNOWN_SPLIT_INDEX;
+import static
org.apache.fluss.lake.writer.WriterInitContext.UNKNOWN_TIERING_ROUND_TIMESTAMP;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/** Hudi implementation of {@link LakeWriter}. */
+public class HudiLakeWriter implements LakeWriter<HudiWriteResult> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(HudiLakeWriter.class);
+
+ private final RecordWriter recordWriter;
+ private final TableInfo tableInfo;
+ private final HudiWriteTableInfo hudiTableInfo;
+ private final CkpMetadata ckpMetadata;
+
+ public HudiLakeWriter(
+ HudiCatalogProvider hudiCatalogProvider,
+ CkpMetadataProvider ckpMetadataProvider,
+ WriterInitContext writerInitContext)
+ throws IOException {
+ validateWriterInitContext(writerInitContext);
+ this.tableInfo = writerInitContext.tableInfo();
+ this.hudiTableInfo =
+ HudiWriteTableInfo.create(hudiCatalogProvider,
tableInfo.getTablePath());
+ this.ckpMetadata = ckpMetadataProvider.get(tableInfo.getTablePath(),
hudiTableInfo);
+
+ if (writerInitContext.splitIndex() == 0) {
+ ckpMetadata.bootstrap();
+ initInstant(hudiTableInfo.getFlinkConfig(),
hudiTableInfo.getMetaClient());
+ LOG.info(
+ "Initialized Hudi instant for first split of table {},
bucket {}.",
+ tableInfo.getTablePath(),
+ writerInitContext.tableBucket());
+ }
+
+ this.recordWriter = new HudiRecordWriter(writerInitContext,
hudiTableInfo, ckpMetadata);
+ LOG.info("Created HudiLakeWriter with configuration {}.",
hudiTableInfo.getFlinkConfig());
+ }
+
+ @Override
+ public void write(LogRecord record) throws IOException {
+ try {
+ recordWriter.write(record);
+ } catch (Exception e) {
+ throw new IOException("Failed to write Fluss record to Hudi.", e);
+ }
+ }
+
+ @Override
+ public HudiWriteResult complete() throws IOException {
+ try {
+ Map<String, List<WriteStatus>> writeStatuses =
recordWriter.complete();
+ return new HudiWriteResult(writeStatuses, new HashMap<>());
+ } catch (Exception e) {
+ throw new IOException("Failed to complete Hudi write.", e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ recordWriter.close();
+ ckpMetadata.close();
+ } catch (Exception e) {
+ throw new IOException("Failed to close HudiLakeWriter.", e);
+ }
+ }
+
+ private void initInstant(Configuration configuration,
HoodieTableMetaClient metaClient) {
+ metaClient.reloadActiveTimeline();
+ WriteOperationType writeOperationType =
+
WriteOperationType.fromValue(configuration.get(FlinkOptions.OPERATION));
+ hudiTableInfo.getWriteClient().preTxn(writeOperationType, metaClient);
+
+ String commitAction =
+ CommitUtils.getCommitActionType(
+ writeOperationType,
+
HoodieTableType.valueOf(configuration.get(FlinkOptions.TABLE_TYPE)));
+ String instant =
hudiTableInfo.getWriteClient().startCommit(commitAction, metaClient);
+
metaClient.getActiveTimeline().transitionRequestedToInflight(commitAction,
instant);
+ hudiTableInfo.getWriteClient().setWriteTimer(commitAction);
+ ckpMetadata.startInstant(instant);
+ LOG.info(
+ "Created Hudi instant {} for table {} with type {}.",
+ instant,
+ tableInfo.getTablePath(),
+ configuration.get(FlinkOptions.TABLE_TYPE));
+ }
+
+ private static void validateWriterInitContext(WriterInitContext
writerInitContext) {
+ checkArgument(
+ writerInitContext.splitIndex() != UNKNOWN_SPLIT_INDEX,
+ "Hudi lake writer requires split index in WriterInitContext.");
+ checkArgument(
+ writerInitContext.tieringRoundTimestamp() !=
UNKNOWN_TIERING_ROUND_TIMESTAMP,
+ "Hudi lake writer requires tiering round timestamp in
WriterInitContext.");
+ }
+}
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
new file mode 100644
index 000000000..973911554
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResult.java
@@ -0,0 +1,48 @@
+/*
+ * 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 java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/** Write result produced by the Hudi lake writer and consumed by a future
Hudi committer. */
+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;
+
+ public HudiWriteResult(
+ Map<String, List<WriteStatus>> writeStatuses,
+ Map<String, List<WriteStatus>> compactionWriteStatuses) {
+ this.writeStatuses = writeStatuses;
+ this.compactionWriteStatuses = compactionWriteStatuses;
+ }
+
+ public Map<String, List<WriteStatus>> getWriteStatuses() {
+ return writeStatuses;
+ }
+
+ public Map<String, List<WriteStatus>> getCompactionWriteStatuses() {
+ return compactionWriteStatuses;
+ }
+}
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
new file mode 100644
index 000000000..57d60fa9b
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializer.java
@@ -0,0 +1,106 @@
+/*
+ * 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 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}. */
+public class HudiWriteResultSerializer implements
SimpleVersionedSerializer<HudiWriteResult> {
+
+ private static final int CURRENT_VERSION = 1;
+
+ @Override
+ public int getVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @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);
+ return baos.toByteArray();
+ }
+ }
+
+ @Override
+ public HudiWriteResult deserialize(int version, byte[] serialized) throws
IOException {
+ if (version != CURRENT_VERSION) {
+ throw new IOException(
+ "Unsupported HudiWriteResult version "
+ + version
+ + ", expected "
+ + CURRENT_VERSION);
+ }
+
+ Map<String, List<WriteStatus>> writeResult;
+ Map<String, List<WriteStatus>> compactionWriteResult;
+ 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());
+ 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);
+ }
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteTableInfo.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteTableInfo.java
new file mode 100644
index 000000000..d4711f261
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiWriteTableInfo.java
@@ -0,0 +1,285 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.client.common.HoodieFlinkEngineContext;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.exception.TableNotFoundException;
+import org.apache.hudi.org.apache.avro.Schema;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.table.catalog.CatalogOptions;
+import org.apache.hudi.util.AvroSchemaConverter;
+import org.apache.hudi.util.CompactionUtil;
+import org.apache.hudi.util.FlinkWriteClients;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.toHudiObjectPath;
+
+/** Resolved Hudi table metadata and write client used by the Hudi lake
writer. */
+public class HudiWriteTableInfo implements AutoCloseable {
+
+ private static final String DEFAULT_HUDI_WAREHOUSE = "/tmp/hudi_warehouse";
+ private static final String FLUSS_CONF_PREFIX = "fluss.";
+
+ private final TablePath tablePath;
+ private final Configuration hudiConfig;
+ private final HoodieFlinkWriteClient writeClient;
+ private final HoodieTableMetaClient metaClient;
+ private final HoodieEngineContext engineContext;
+ private final HoodieWriteConfig writeConfig;
+ private final org.apache.flink.configuration.Configuration flinkConfig;
+ private final Schema avroSchema;
+ private final RowType rowType;
+ private final List<String> logicalTypes;
+ private final String basePath;
+ private boolean closed;
+
+ private HudiWriteTableInfo(
+ TablePath tablePath,
+ Configuration hudiConfig,
+ HoodieFlinkWriteClient writeClient,
+ HoodieTableMetaClient metaClient,
+ HoodieEngineContext engineContext,
+ HoodieWriteConfig writeConfig,
+ org.apache.flink.configuration.Configuration flinkConfig,
+ Schema avroSchema,
+ RowType rowType,
+ List<String> logicalTypes,
+ String basePath) {
+ this.tablePath = tablePath;
+ this.hudiConfig = hudiConfig;
+ this.writeClient = writeClient;
+ this.metaClient = metaClient;
+ this.engineContext = engineContext;
+ this.writeConfig = writeConfig;
+ this.flinkConfig = flinkConfig;
+ this.avroSchema = avroSchema;
+ this.rowType = rowType;
+ this.logicalTypes = logicalTypes;
+ this.basePath = basePath;
+ }
+
+ public static HudiWriteTableInfo create(
+ HudiCatalogProvider hudiCatalogProvider, TablePath tablePath)
throws IOException {
+ Catalog hudiCatalog = hudiCatalogProvider.get();
+ try {
+ CatalogBaseTable hudiTable =
hudiCatalog.getTable(toHudiObjectPath(tablePath));
+ Map<String, String> hudiOptions = new
HashMap<>(hudiTable.getOptions());
+ hudiOptions.putAll(hudiCatalogProvider.getHudiConfig().toMap());
+
+ String basePath = resolveBasePath(tablePath, hudiOptions);
+ hudiOptions.put(FlinkOptions.TABLE_NAME.key(),
tablePath.getTableName());
+ hudiOptions.put(FlinkOptions.PATH.key(), basePath);
+
+ boolean isCowTable =
+ Objects.equals(
+ hudiOptions.get(FlinkOptions.TABLE_TYPE.key()),
+ HoodieTableType.COPY_ON_WRITE.name());
+ hudiOptions.putIfAbsent(
+ FlinkOptions.OPERATION.key(),
+ isCowTable
+ ? WriteOperationType.INSERT.value()
+ : WriteOperationType.UPSERT.value());
+ hudiOptions.putIfAbsent(
+ FlinkOptions.COMPACTION_SCHEDULE_ENABLED.key(),
+ hudiOptions.getOrDefault(
+ FLUSS_CONF_PREFIX +
ConfigOptions.TABLE_DATALAKE_AUTO_COMPACTION.key(),
+ "false"));
+
+ HoodieTableMetaClient metaClient =
+ createMetaClient(basePath,
hudiCatalogProvider.getHudiConfig());
+ org.apache.flink.configuration.Configuration flinkConfig =
+
org.apache.flink.configuration.Configuration.fromMap(hudiOptions);
+ try {
+ CompactionUtil.setAvroSchema(flinkConfig, metaClient);
+ } catch (Exception e) {
+ throw new IOException("Failed to resolve Hudi Avro schema for
" + tablePath, e);
+ }
+
+ HoodieWriteConfig writeConfig =
+ FlinkWriteClients.getHoodieClientConfig(flinkConfig,
false, false);
+ HoodieEngineContext engineContext =
+ new HoodieLocalEngineContext(metaClient.getStorageConf());
+ HoodieFlinkWriteClient writeClient =
+ new
HoodieFlinkWriteClient<>(HoodieFlinkEngineContext.DEFAULT, writeConfig);
+ Schema avroSchema = Schema.parse(writeConfig.getSchema());
+ RowType rowType =
+ (RowType)
AvroSchemaConverter.convertToDataType(avroSchema).getLogicalType();
+
+ List<String> logicalTypes = new ArrayList<>();
+ for (RowType.RowField rowField : rowType.getFields()) {
+ logicalTypes.add(rowField.getType().getTypeRoot().name());
+ }
+
+ return new HudiWriteTableInfo(
+ tablePath,
+ hudiCatalogProvider.getHudiConfig(),
+ writeClient,
+ metaClient,
+ engineContext,
+ writeConfig,
+ flinkConfig,
+ avroSchema,
+ rowType,
+ logicalTypes,
+ basePath);
+ } catch (Exception e) {
+ if (e instanceof IOException) {
+ throw (IOException) e;
+ }
+ throw new IOException("Failed to resolve Hudi write table info for
" + tablePath, e);
+ } finally {
+ closeCatalog(hudiCatalog);
+ }
+ }
+
+ private static HoodieTableMetaClient createMetaClient(String basePath,
Configuration hudiConfig)
+ throws IOException {
+ try {
+ return HoodieTableMetaClient.builder()
+ .setBasePath(basePath)
+ .setConf(new
HadoopStorageConfiguration(getHadoopConfiguration(hudiConfig)))
+ .build();
+ } catch (TableNotFoundException e) {
+ throw new IOException("Hudi table not found at " + basePath + ".",
e);
+ }
+ }
+
+ public static org.apache.hadoop.conf.Configuration getHadoopConfiguration(
+ Configuration hudiConfig) {
+ org.apache.flink.configuration.Configuration flinkConfig =
+
org.apache.flink.configuration.Configuration.fromMap(hudiConfig.toMap());
+ return HadoopConfigurations.getHadoopConf(flinkConfig);
+ }
+
+ private static String resolveBasePath(TablePath tablePath, Map<String,
String> tableOptions) {
+ String path = tableOptions.get(FlinkOptions.PATH.key());
+ if (path != null && !path.trim().isEmpty()) {
+ return path;
+ }
+ String catalogPath =
+ tableOptions.getOrDefault(
+ CatalogOptions.CATALOG_PATH.key(),
DEFAULT_HUDI_WAREHOUSE);
+ return ensureEndsWithSlash(catalogPath)
+ + tablePath.getDatabaseName()
+ + "/"
+ + tablePath.getTableName();
+ }
+
+ private static String ensureEndsWithSlash(String path) {
+ return path.endsWith("/") ? path : path + "/";
+ }
+
+ private static void closeCatalog(Catalog hudiCatalog) {
+ IOUtils.closeQuietly(hudiCatalog::close, "hudi catalog");
+ }
+
+ public HoodieTableMetaClient getMetaClient() {
+ return metaClient;
+ }
+
+ public HoodieEngineContext getEngineContext() {
+ return engineContext;
+ }
+
+ public HoodieTableFileSystemView getFileSystemView() {
+ return HoodieTableFileSystemView.fileListingBasedFileSystemView(
+ engineContext, metaClient, getCompletedCommitsTimeline());
+ }
+
+ public HoodieTimeline getCompletedCommitsTimeline() {
+ return
metaClient.getCommitsAndCompactionTimeline().filterCompletedAndCompactionInstants();
+ }
+
+ public HoodieTableType getTableType() {
+ return metaClient.getTableType();
+ }
+
+ public String getBasePath() {
+ return metaClient.getBasePath().toString();
+ }
+
+ public String getTableBasePath() {
+ return basePath;
+ }
+
+ public TablePath getTablePath() {
+ return tablePath;
+ }
+
+ public Configuration getHudiConfig() {
+ return hudiConfig;
+ }
+
+ public HoodieWriteConfig getWriteConfig() {
+ return writeConfig;
+ }
+
+ public org.apache.flink.configuration.Configuration getFlinkConfig() {
+ return flinkConfig;
+ }
+
+ public Schema getAvroSchema() {
+ return avroSchema;
+ }
+
+ public HoodieFlinkWriteClient getWriteClient() {
+ return writeClient;
+ }
+
+ public RowType getRowType() {
+ return rowType;
+ }
+
+ public List<String> getLogicalTypes() {
+ return logicalTypes;
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ IOUtils.closeQuietly(writeClient::close, "hudi write client");
+ }
+}
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
new file mode 100644
index 000000000..e4a0e2d06
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/RecordWriter.java
@@ -0,0 +1,160 @@
+/*
+ * 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.hudi.tiering.writer.FlussRecordAsHudiRow;
+import org.apache.fluss.lake.hudi.tiering.writer.RecordWriteBuffer;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata;
+import org.apache.fluss.lake.writer.WriterInitContext;
+import org.apache.fluss.record.LogRecord;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.index.bucket.BucketIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Base writer that writes Fluss log records to Hudi. */
+public abstract class RecordWriter implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RecordWriter.class);
+
+ protected final int bucketNum;
+ protected final HudiWriteTableInfo hudiTableInfo;
+ protected final RecordWriteBuffer recordWriteBuffer;
+ protected final FlussRecordAsHudiRow flussRecordAsHudiRecord;
+ protected final CkpMetadata ckpMetadata;
+
+ private final Map<String, Map<Integer, String>> bucketIndex = new
HashMap<>();
+ private final Set<String> incBucketIndex = new HashSet<>();
+
+ public RecordWriter(
+ WriterInitContext writerInitContext,
+ HudiWriteTableInfo hudiTableInfo,
+ CkpMetadata ckpMetadata) {
+ this.bucketNum = writerInitContext.tableBucket().getBucket();
+ this.flussRecordAsHudiRecord =
+ new FlussRecordAsHudiRow(bucketNum,
hudiTableInfo.getRowType());
+ this.hudiTableInfo = hudiTableInfo;
+ this.ckpMetadata = ckpMetadata;
+ this.recordWriteBuffer =
+ new RecordWriteBuffer(
+ hudiTableInfo, ckpMetadata,
writerInitContext.tieringRoundTimestamp());
+ }
+
+ public abstract void write(LogRecord record) throws Exception;
+
+ public Map<String, List<WriteStatus>> complete() throws Exception {
+ try {
+ recordWriteBuffer.flushRemaining();
+ Map<String, List<WriteStatus>> writeResult =
recordWriteBuffer.getWriteStatuses();
+ LOG.info("Wrote Hudi records and got write statuses {}.",
writeResult);
+ return writeResult;
+ } catch (Exception e) {
+ throw new IOException("Failed to write Hudi records.", e);
+ }
+ }
+
+ @Override
+ public void close() {
+ hudiTableInfo.close();
+ }
+
+ public void setRecordLocation(HoodieFlinkInternalRow internalRow,
Configuration configuration) {
+ String partition = internalRow.getPartitionPath();
+
+ if (!configuration
+ .getString(FlinkOptions.OPERATION)
+ .equals(WriteOperationType.INSERT.value())) {
+ bootstrapIndexIfNeed(partition);
+ }
+
+ Map<Integer, String> bucketToFileId =
+ bucketIndex.computeIfAbsent(partition, ignored -> new
HashMap<>());
+ String bucketId = partition + "/" + bucketNum;
+
+ if (incBucketIndex.contains(bucketId)) {
+ internalRow.setInstantTime("I");
+ internalRow.setFileId(bucketToFileId.get(bucketNum));
+ } else if (bucketToFileId.containsKey(bucketNum)) {
+ internalRow.setInstantTime("U");
+ internalRow.setFileId(bucketToFileId.get(bucketNum));
+ } else {
+ String newFileId =
+ isNonBlockingConcurrencyControl(configuration)
+ ?
BucketIdentifier.newBucketFileIdForNBCC(bucketNum)
+ :
BucketIdentifier.newBucketFileIdPrefix(bucketNum);
+ internalRow.setInstantTime("I");
+ internalRow.setFileId(newFileId);
+ bucketToFileId.put(bucketNum, newFileId);
+ incBucketIndex.add(bucketId);
+ }
+ }
+
+ public boolean isNonBlockingConcurrencyControl(Configuration config) {
+ return WriteConcurrencyMode.isNonBlockingConcurrencyControl(
+ config.getString(
+ HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(),
+
HoodieWriteConfig.WRITE_CONCURRENCY_MODE.defaultValue()));
+ }
+
+ private void bootstrapIndexIfNeed(String partition) {
+ if (bucketIndex.containsKey(partition) &&
!bucketIndex.get(partition).isEmpty()) {
+ return;
+ }
+
+ Map<Integer, String> bucketToFileIDMap = new HashMap<>();
+ hudiTableInfo
+ .getWriteClient()
+ .getHoodieTable()
+ .getHoodieView()
+ .getLatestFileSlices(partition)
+ .forEach(
+ fileSlice -> {
+ String fileId = fileSlice.getFileId();
+ int bucketNumber =
BucketIdentifier.bucketIdFromFileId(fileId);
+ if (bucketNumber != bucketNum) {
+ return;
+ }
+ if (bucketToFileIDMap.containsKey(bucketNumber)
+ && hudiTableInfo.getTableType()
+ == HoodieTableType.MERGE_ON_READ) {
+ throw new IllegalStateException(
+ String.format(
+ "Duplicate fileId %s for
bucket %s of partition %s.",
+ fileId, bucketNumber,
partition));
+ }
+ bucketToFileIDMap.put(bucketNumber, fileId);
+ });
+ bucketIndex.put(partition, bucketToFileIDMap);
+ LOG.debug("Bootstrapped Hudi bucket index {} for partition {}.",
bucketIndex, partition);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussArrayAsHudiArray.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussArrayAsHudiArray.java
new file mode 100644
index 000000000..8103bc99d
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussArrayAsHudiArray.java
@@ -0,0 +1,193 @@
+/*
+ * 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.writer;
+
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RawValueData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.RowType;
+
+/** Wraps a Fluss {@link InternalArray} as a Hudi/Flink {@link ArrayData}. */
+public class FlussArrayAsHudiArray implements ArrayData {
+
+ private final InternalArray flussArray;
+ private final LogicalType elementType;
+
+ public FlussArrayAsHudiArray(InternalArray flussArray, LogicalType
elementType) {
+ this.flussArray = flussArray;
+ this.elementType = elementType;
+ }
+
+ @Override
+ public int size() {
+ return flussArray.size();
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return flussArray.isNullAt(pos);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return flussArray.getBoolean(pos);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return flussArray.getByte(pos);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return flussArray.getShort(pos);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return flussArray.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return flussArray.getLong(pos);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return flussArray.getFloat(pos);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return flussArray.getDouble(pos);
+ }
+
+ @Override
+ public StringData getString(int pos) {
+ return StringData.fromBytes(flussArray.getString(pos).toBytes());
+ }
+
+ @Override
+ public DecimalData getDecimal(int pos, int precision, int scale) {
+ Decimal flussDecimal = flussArray.getDecimal(pos, precision, scale);
+ if (flussDecimal.isCompact()) {
+ return DecimalData.fromUnscaledLong(flussDecimal.toUnscaledLong(),
precision, scale);
+ }
+ return DecimalData.fromBigDecimal(flussDecimal.toBigDecimal(),
precision, scale);
+ }
+
+ @Override
+ public TimestampData getTimestamp(int pos, int precision) {
+ LogicalTypeRoot typeRoot = elementType.getTypeRoot();
+ if (typeRoot == LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) {
+ TimestampNtz timestampNtz = flussArray.getTimestampNtz(pos,
precision);
+ return
TimestampData.fromLocalDateTime(timestampNtz.toLocalDateTime());
+ } else if (typeRoot == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)
{
+ TimestampLtz timestampLtz = flussArray.getTimestampLtz(pos,
precision);
+ return TimestampData.fromEpochMillis(
+ timestampLtz.getEpochMillisecond(),
timestampLtz.getNanoOfMillisecond());
+ }
+ throw new UnsupportedOperationException("Unsupported timestamp type: "
+ typeRoot);
+ }
+
+ @Override
+ public <T> RawValueData<T> getRawValue(int pos) {
+ throw new UnsupportedOperationException("getRawValue is not supported
for Fluss records.");
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ return flussArray.getBytes(pos);
+ }
+
+ @Override
+ public ArrayData getArray(int pos) {
+ InternalArray innerArray = flussArray.getArray(pos);
+ return innerArray == null
+ ? null
+ : new FlussArrayAsHudiArray(innerArray, ((ArrayType)
elementType).getElementType());
+ }
+
+ @Override
+ public MapData getMap(int pos) {
+ InternalMap flussMap = flussArray.getMap(pos);
+ MapType mapType = (MapType) elementType;
+ return flussMap == null
+ ? null
+ : new FlussMapAsHudiMap(flussMap, mapType.getKeyType(),
mapType.getValueType());
+ }
+
+ @Override
+ public RowData getRow(int pos, int numFields) {
+ InternalRow nestedFlussRow = flussArray.getRow(pos, numFields);
+ return nestedFlussRow == null
+ ? null
+ : new FlussRowAsHudiRow(nestedFlussRow, (RowType) elementType);
+ }
+
+ @Override
+ public boolean[] toBooleanArray() {
+ return flussArray.toBooleanArray();
+ }
+
+ @Override
+ public byte[] toByteArray() {
+ return flussArray.toByteArray();
+ }
+
+ @Override
+ public short[] toShortArray() {
+ return flussArray.toShortArray();
+ }
+
+ @Override
+ public int[] toIntArray() {
+ return flussArray.toIntArray();
+ }
+
+ @Override
+ public long[] toLongArray() {
+ return flussArray.toLongArray();
+ }
+
+ @Override
+ public float[] toFloatArray() {
+ return flussArray.toFloatArray();
+ }
+
+ @Override
+ public double[] toDoubleArray() {
+ return flussArray.toDoubleArray();
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussMapAsHudiMap.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussMapAsHudiMap.java
new file mode 100644
index 000000000..8089609d1
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussMapAsHudiMap.java
@@ -0,0 +1,58 @@
+/*
+ * 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.writer;
+
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.types.logical.LogicalType;
+
+/** Wraps a Fluss {@link InternalMap} as a Hudi/Flink {@link MapData}. */
+public class FlussMapAsHudiMap implements MapData {
+
+ private final InternalMap flussMap;
+ private final LogicalType keyType;
+ private final LogicalType valueType;
+
+ public FlussMapAsHudiMap(InternalMap flussMap, LogicalType keyType,
LogicalType valueType) {
+ this.flussMap = flussMap;
+ this.keyType = keyType;
+ this.valueType = valueType;
+ }
+
+ @Override
+ public int size() {
+ return flussMap.size();
+ }
+
+ @Override
+ public ArrayData keyArray() {
+ InternalArray flussKeyArray = flussMap.keyArray();
+ return flussKeyArray == null ? null : new
FlussArrayAsHudiArray(flussKeyArray, keyType);
+ }
+
+ @Override
+ public ArrayData valueArray() {
+ InternalArray flussValueArray = flussMap.valueArray();
+ return flussValueArray == null
+ ? null
+ : new FlussArrayAsHudiArray(flussValueArray, valueType);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java
new file mode 100644
index 000000000..f62adc76a
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRow.java
@@ -0,0 +1,90 @@
+/*
+ * 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.writer;
+
+import org.apache.fluss.record.LogRecord;
+
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+
+import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS;
+import static org.apache.fluss.lake.hudi.utils.HudiConversions.toRowKind;
+import static org.apache.fluss.utils.Preconditions.checkState;
+
+/** Wraps a Fluss {@link LogRecord} as a Hudi/Flink row with Fluss system
columns. */
+public class FlussRecordAsHudiRow extends FlussRowAsHudiRow {
+
+ private final int bucket;
+
+ private LogRecord logRecord;
+ private int originRowFieldCount;
+
+ public FlussRecordAsHudiRow(int bucket, RowType rowType) {
+ super(rowType);
+ this.bucket = bucket;
+ }
+
+ public void setFlussRecord(LogRecord logRecord) {
+ this.logRecord = logRecord;
+ this.internalRow = logRecord.getRow();
+ this.originRowFieldCount = internalRow.getFieldCount();
+ checkState(
+ originRowFieldCount == rowType.getFieldCount() -
SYSTEM_COLUMNS.size(),
+ "Hudi table field count must equal Fluss LogRecord field count
plus system fields.");
+ }
+
+ @Override
+ public RowKind getRowKind() {
+ return toRowKind(logRecord.getChangeType());
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ if (pos < originRowFieldCount) {
+ return super.isNullAt(pos);
+ }
+ return false;
+ }
+
+ @Override
+ public int getInt(int pos) {
+ if (pos == originRowFieldCount) {
+ return bucket;
+ }
+ return super.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ if (pos == originRowFieldCount + 1) {
+ return logRecord.logOffset();
+ } else if (pos == originRowFieldCount + 2) {
+ return logRecord.timestamp();
+ }
+ return super.getLong(pos);
+ }
+
+ @Override
+ public TimestampData getTimestamp(int pos, int precision) {
+ if (pos == originRowFieldCount + 2) {
+ return TimestampData.fromEpochMillis(logRecord.timestamp());
+ }
+ return super.getTimestamp(pos, precision);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRowAsHudiRow.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRowAsHudiRow.java
new file mode 100644
index 000000000..36bc34b7d
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRowAsHudiRow.java
@@ -0,0 +1,183 @@
+/*
+ * 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.writer;
+
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalArray;
+import org.apache.fluss.row.InternalMap;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.row.compacted.CompactedRow;
+import org.apache.fluss.row.indexed.IndexedRow;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RawValueData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.RowKind;
+import org.apache.hudi.common.util.ObjectSizeCalculator;
+
+/** Wraps a Fluss {@link InternalRow} as a Hudi/Flink {@link RowData}. */
+public class FlussRowAsHudiRow implements RowData {
+
+ protected InternalRow internalRow;
+ protected final RowType rowType;
+
+ public FlussRowAsHudiRow(RowType rowType) {
+ this.rowType = rowType;
+ }
+
+ public FlussRowAsHudiRow(InternalRow internalRow, RowType rowType) {
+ this.internalRow = internalRow;
+ this.rowType = rowType;
+ }
+
+ @Override
+ public RowKind getRowKind() {
+ return RowKind.INSERT;
+ }
+
+ @Override
+ public boolean isNullAt(int pos) {
+ return internalRow.isNullAt(pos);
+ }
+
+ @Override
+ public boolean getBoolean(int pos) {
+ return internalRow.getBoolean(pos);
+ }
+
+ @Override
+ public byte getByte(int pos) {
+ return internalRow.getByte(pos);
+ }
+
+ @Override
+ public short getShort(int pos) {
+ return internalRow.getShort(pos);
+ }
+
+ @Override
+ public int getInt(int pos) {
+ return internalRow.getInt(pos);
+ }
+
+ @Override
+ public long getLong(int pos) {
+ return internalRow.getLong(pos);
+ }
+
+ @Override
+ public float getFloat(int pos) {
+ return internalRow.getFloat(pos);
+ }
+
+ @Override
+ public double getDouble(int pos) {
+ return internalRow.getDouble(pos);
+ }
+
+ @Override
+ public StringData getString(int pos) {
+ return StringData.fromBytes(internalRow.getString(pos).toBytes());
+ }
+
+ @Override
+ public DecimalData getDecimal(int pos, int precision, int scale) {
+ Decimal flussDecimal = internalRow.getDecimal(pos, precision, scale);
+ if (flussDecimal.isCompact()) {
+ return DecimalData.fromUnscaledLong(flussDecimal.toUnscaledLong(),
precision, scale);
+ }
+ return DecimalData.fromBigDecimal(flussDecimal.toBigDecimal(),
precision, scale);
+ }
+
+ @Override
+ public TimestampData getTimestamp(int pos, int precision) {
+ LogicalTypeRoot typeRoot = rowType.getTypeAt(pos).getTypeRoot();
+ if (typeRoot == LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) {
+ TimestampNtz timestampNtz = internalRow.getTimestampNtz(pos,
precision);
+ return
TimestampData.fromLocalDateTime(timestampNtz.toLocalDateTime());
+ } else if (typeRoot == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)
{
+ TimestampLtz timestampLtz = internalRow.getTimestampLtz(pos,
precision);
+ return TimestampData.fromEpochMillis(
+ timestampLtz.getEpochMillisecond(),
timestampLtz.getNanoOfMillisecond());
+ }
+ throw new UnsupportedOperationException("Unsupported timestamp type: "
+ typeRoot);
+ }
+
+ @Override
+ public <T> RawValueData<T> getRawValue(int pos) {
+ throw new UnsupportedOperationException("getRawValue is not supported
for Fluss records.");
+ }
+
+ @Override
+ public byte[] getBinary(int pos) {
+ return internalRow.getBytes(pos);
+ }
+
+ @Override
+ public ArrayData getArray(int pos) {
+ InternalArray array = internalRow.getArray(pos);
+ return array == null
+ ? null
+ : new FlussArrayAsHudiArray(
+ array, ((ArrayType)
rowType.getTypeAt(pos)).getElementType());
+ }
+
+ @Override
+ public MapData getMap(int pos) {
+ InternalMap map = internalRow.getMap(pos);
+ MapType mapType = (MapType) rowType.getTypeAt(pos);
+ return map == null
+ ? null
+ : new FlussMapAsHudiMap(map, mapType.getKeyType(),
mapType.getValueType());
+ }
+
+ @Override
+ public RowData getRow(int pos, int numFields) {
+ InternalRow row = internalRow.getRow(pos, numFields);
+ return row == null ? null : new FlussRowAsHudiRow(row, (RowType)
rowType.getTypeAt(pos));
+ }
+
+ @Override
+ public int getArity() {
+ return rowType.getFieldCount();
+ }
+
+ @Override
+ public void setRowKind(RowKind rowKind) {
+ // Fluss records expose their row kind through LogRecord change type.
+ }
+
+ public long sizeInBytes() {
+ if (internalRow instanceof IndexedRow) {
+ return ((IndexedRow) internalRow).getSizeInBytes();
+ } else if (internalRow instanceof CompactedRow) {
+ return ((CompactedRow) internalRow).getSizeInBytes();
+ }
+ return ObjectSizeCalculator.getObjectSize(internalRow);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/HudiRecordConverter.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/HudiRecordConverter.java
new file mode 100644
index 000000000..817522725
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/HudiRecordConverter.java
@@ -0,0 +1,44 @@
+/*
+ * 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.writer;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.hudi.client.model.HoodieFlinkRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.sink.bulk.RowDataKeyGen;
+import org.apache.hudi.table.action.commit.BucketInfo;
+
+import java.io.Serializable;
+
+/** Converts buffered Flink rows into Hudi records. */
+public interface HudiRecordConverter extends Serializable {
+
+ HoodieRecord<?> convert(RowData rowData, BucketInfo bucketInfo);
+
+ static HudiRecordConverter getInstance(RowDataKeyGen keyGen) {
+ return (rowData, bucketInfo) -> {
+ String key = keyGen.getRecordKey(rowData);
+ HoodieOperation operation =
+
HoodieOperation.fromValue(rowData.getRowKind().toByteValue());
+ HoodieKey hoodieKey = new HoodieKey(key,
bucketInfo.getPartitionPath());
+ return new HoodieFlinkRecord(hoodieKey, operation, rowData);
+ };
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/HudiRecordWriter.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/HudiRecordWriter.java
new file mode 100644
index 000000000..d059daa5c
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/HudiRecordWriter.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.tiering.writer;
+
+import org.apache.fluss.lake.hudi.tiering.HudiWriteTableInfo;
+import org.apache.fluss.lake.hudi.tiering.RecordWriter;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata;
+import org.apache.fluss.lake.writer.WriterInitContext;
+import org.apache.fluss.record.LogRecord;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.HoodieOperation;
+import org.apache.hudi.sink.bulk.RowDataKeyGen;
+
+/** Hudi {@link RecordWriter} implementation. */
+public class HudiRecordWriter extends RecordWriter {
+
+ private final Configuration config;
+ private final RowDataKeyGen keyGen;
+
+ public HudiRecordWriter(
+ WriterInitContext writerInitContext,
+ HudiWriteTableInfo hudiTableInfo,
+ CkpMetadata ckpMetadata) {
+ super(writerInitContext, hudiTableInfo, ckpMetadata);
+ this.config = hudiTableInfo.getFlinkConfig();
+ this.keyGen =
+ RowDataKeyGen.instance(hudiTableInfo.getFlinkConfig(),
hudiTableInfo.getRowType());
+ }
+
+ @Override
+ public void write(LogRecord record) throws Exception {
+ flussRecordAsHudiRecord.setFlussRecord(record);
+ HoodieFlinkInternalRow internalRow =
+ new HoodieFlinkInternalRow(
+ keyGen.getRecordKey(flussRecordAsHudiRecord),
+ keyGen.getPartitionPath(flussRecordAsHudiRecord),
+ HoodieOperation.fromValue(
+
flussRecordAsHudiRecord.getRowKind().toByteValue())
+ .getName(),
+ flussRecordAsHudiRecord);
+ setRecordLocation(internalRow, config);
+ recordWriteBuffer.bufferRecord(internalRow);
+ }
+}
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
new file mode 100644
index 000000000..3a99ede2a
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBuffer.java
@@ -0,0 +1,392 @@
+/*
+ * 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.writer;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.lake.hudi.tiering.HudiWriteTableInfo;
+import org.apache.fluss.lake.hudi.utils.meta.CkpMetadata;
+import org.apache.fluss.lake.writer.WriterInitContext;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.operators.sort.BinaryInMemorySortBuffer;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.MappingIterator;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.sink.buffer.MemorySegmentPoolFactory;
+import org.apache.hudi.sink.bulk.RowDataKeyGen;
+import org.apache.hudi.sink.exception.MemoryPagesExhaustedException;
+import org.apache.hudi.sink.utils.BufferUtils;
+import org.apache.hudi.sink.utils.TimeWait;
+import org.apache.hudi.table.action.commit.BucketInfo;
+import org.apache.hudi.table.action.commit.BucketType;
+import org.apache.hudi.util.MutableIteratorWrapperIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+
+/** Buffers records and writes them to Hudi in batches. */
+public class RecordWriteBuffer {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RecordWriteBuffer.class);
+ private static final long WAIT_INSTANT_TIMEOUT_MS = 5 * 60 * 1000L;
+
+ private DataBucket bucket;
+
+ private final TotalSizeTracer tracer;
+ private final HoodieFlinkWriteClient writeClient;
+ private final HudiWriteTableInfo hudiTableInfo;
+ private final Configuration config;
+ private final CkpMetadata ckpMetadata;
+ private final Map<String, List<WriteStatus>> writeStatuses;
+ private final HudiRecordConverter recordConverter;
+ private final long tieringRoundTimestamp;
+
+ public RecordWriteBuffer(
+ HudiWriteTableInfo hudiTableInfo, CkpMetadata ckpMetadata, long
tieringRoundTimestamp) {
+ this(
+ hudiTableInfo,
+ hudiTableInfo.getFlinkConfig(),
+ hudiTableInfo.getWriteClient(),
+ ckpMetadata,
+ HudiRecordConverter.getInstance(
+ RowDataKeyGen.instance(
+ hudiTableInfo.getFlinkConfig(),
hudiTableInfo.getRowType())),
+ tieringRoundTimestamp);
+ }
+
+ @VisibleForTesting
+ RecordWriteBuffer(
+ HudiWriteTableInfo hudiTableInfo,
+ Configuration config,
+ HoodieFlinkWriteClient writeClient,
+ CkpMetadata ckpMetadata,
+ HudiRecordConverter recordConverter,
+ long tieringRoundTimestamp) {
+ this.hudiTableInfo = hudiTableInfo;
+ this.config = config;
+ this.tracer = new TotalSizeTracer(config);
+ this.writeClient = writeClient;
+ this.ckpMetadata = ckpMetadata;
+ this.writeStatuses = new HashMap<>();
+ this.tieringRoundTimestamp = tieringRoundTimestamp;
+ this.recordConverter = recordConverter;
+ }
+
+ public void bufferRecord(HoodieFlinkInternalRow internalRow) throws
IOException {
+ BucketInfo bucketInfo = getBucketInfo(internalRow);
+ if (bucket != null && !isSameBucketInfo(bucket.getBucketInfo(),
bucketInfo)) {
+ flushAndDisposeBucket();
+ }
+
+ boolean success = doBufferRecord(internalRow, bucketInfo);
+ if (!success) {
+ flushAndDisposeBucket();
+ success = doBufferRecord(internalRow, bucketInfo);
+ if (!success) {
+ throw new IOException("Hudi write buffer is too small to hold
a single record.");
+ }
+ }
+
+ if (bucket != null) {
+ boolean totalSizeExceedsThreshold =
tracer.trace(bucket.getLastRecordSize());
+ if (bucket.isFull() || totalSizeExceedsThreshold) {
+ flushAndDisposeBucket();
+ }
+ }
+ }
+
+ public void flushRemaining() {
+ flushAndDisposeBucket();
+ tracer.reset();
+ writeClient.cleanHandles();
+ }
+
+ public Map<String, List<WriteStatus>> getWriteStatuses() {
+ return writeStatuses;
+ }
+
+ private boolean doBufferRecord(HoodieFlinkInternalRow internalRow,
BucketInfo bucketInfo)
+ throws IOException {
+ try {
+ if (bucket == null) {
+ bucket = createDataBucket(bucketInfo);
+ LOG.debug("Initialized a new Hudi data bucket.");
+ }
+ return bucket.writeRow(internalRow.getRowData());
+ } catch (MemoryPagesExhaustedException e) {
+ LOG.warn("Hudi write buffer memory pages are exhausted, flushing
current bucket.", e);
+ return false;
+ }
+ }
+
+ private void flushAndDisposeBucket() {
+ if (bucket == null) {
+ return;
+ }
+ if (bucket.isEmpty()) {
+ bucket.dispose();
+ bucket = null;
+ return;
+ }
+ if (flushBucket(bucket)) {
+ tracer.countDown(bucket.getBufferSize());
+ bucket.dispose();
+ bucket = null;
+ }
+ }
+
+ private boolean flushBucket(DataBucket bucket) {
+ String instantTime = getLastPendingInstant();
+ LOG.info(
+ "Flushing Hudi records for instant {}, size {}.",
+ instantTime,
+ bucket.getBufferSize());
+
+ 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));
+ writeStatuses.put(instantTime, writeStatus);
+ return true;
+ }
+
+ protected DataBucket createDataBucket(BucketInfo bucketInfo) {
+ return new DataBucket(
+ BufferUtils.createBuffer(
+ hudiTableInfo.getRowType(),
+
MemorySegmentPoolFactory.createMemorySegmentPool(config)),
+ bucketInfo,
+ config.getDouble(FlinkOptions.WRITE_BATCH_SIZE));
+ }
+
+ protected List<WriteStatus> writeRecords(
+ String instant, String writeOperation, DataBucket dataBucket) {
+ Iterator<BinaryRowData> rowItr =
+ new MutableIteratorWrapperIterator<>(
+ dataBucket.getDataIterator(),
+ () -> new
BinaryRowData(hudiTableInfo.getRowType().getFieldCount()));
+ Iterator<HoodieRecord> recordItr =
+ new MappingIterator<>(
+ rowItr,
+ rowData -> recordConverter.convert(rowData,
dataBucket.getBucketInfo()));
+
+ if (WriteOperationType.UPSERT.value().equals(writeOperation)) {
+ return writeClient.upsert(recordItr, dataBucket.bucketInfo,
instant);
+ } else if (WriteOperationType.INSERT.value().equals(writeOperation)) {
+ return writeClient.insert(recordItr, dataBucket.bucketInfo,
instant);
+ }
+ LOG.warn("Unsupported Hudi write operation {}, skip writing records.",
writeOperation);
+ return Collections.emptyList();
+ }
+
+ public String getLastPendingInstant() {
+ String instant = ckpMetadata.lastPendingInstant();
+ TimeWait timeWait =
+ TimeWait.builder()
+ .timeout(WAIT_INSTANT_TIMEOUT_MS)
+ .action("instant initialize")
+ .build();
+ String roundLowerBoundInstant = roundLowerBoundInstant();
+ while (instant == null
+ || (roundLowerBoundInstant != null
+ && roundLowerBoundInstant.compareTo(instant) >= 0)) {
+ timeWait.waitFor();
+ instant = ckpMetadata.lastPendingInstant();
+ LOG.info("Waiting for Hudi instant to be initialized: {}",
instant);
+ }
+ return instant;
+ }
+
+ private String roundLowerBoundInstant() {
+ if (tieringRoundTimestamp ==
WriterInitContext.UNKNOWN_TIERING_ROUND_TIMESTAMP) {
+ return null;
+ }
+ return HoodieInstantTimeGenerator.formatDate(new
Date(tieringRoundTimestamp));
+ }
+
+ private static BucketInfo getBucketInfo(HoodieFlinkInternalRow
internalRow) {
+ BucketType bucketType;
+ switch (internalRow.getInstantTime()) {
+ case "I":
+ bucketType = BucketType.INSERT;
+ break;
+ case "U":
+ bucketType = BucketType.UPDATE;
+ break;
+ default:
+ throw new HoodieException(
+ "Unexpected Hudi bucket type: " +
internalRow.getInstantTime());
+ }
+ return new BucketInfo(bucketType, internalRow.getFileId(),
internalRow.getPartitionPath());
+ }
+
+ private static boolean isSameBucketInfo(BucketInfo left, BucketInfo right)
{
+ return left.getBucketType() == right.getBucketType()
+ && Objects.equals(left.getFileIdPrefix(),
right.getFileIdPrefix())
+ && Objects.equals(left.getPartitionPath(),
right.getPartitionPath());
+ }
+
+ /** Buffered rows for one Hudi file bucket. */
+ protected static class DataBucket {
+
+ private final BinaryInMemorySortBuffer dataBuffer;
+ private final BucketInfo bucketInfo;
+ private final BufferSizeDetector detector;
+
+ protected DataBucket(
+ BinaryInMemorySortBuffer dataBuffer, BucketInfo bucketInfo,
double batchSize) {
+ this.dataBuffer = dataBuffer;
+ this.bucketInfo = bucketInfo;
+ this.detector = new BufferSizeDetector(batchSize);
+ }
+
+ public MutableObjectIterator<BinaryRowData> getDataIterator() {
+ return dataBuffer.getIterator();
+ }
+
+ public boolean writeRow(RowData rowData) throws IOException {
+ boolean success = dataBuffer.write(rowData);
+ if (success) {
+ detector.detect(rowData);
+ }
+ return success;
+ }
+
+ public BucketInfo getBucketInfo() {
+ return bucketInfo;
+ }
+
+ public long getBufferSize() {
+ return detector.totalSize;
+ }
+
+ public boolean isEmpty() {
+ return dataBuffer.isEmpty();
+ }
+
+ public boolean isFull() {
+ return detector.isFull();
+ }
+
+ public long getLastRecordSize() {
+ return detector.getLastRecordSize();
+ }
+
+ public void dispose() {
+ dataBuffer.dispose();
+ detector.reset();
+ }
+ }
+
+ private static class BufferSizeDetector {
+
+ private static final int DENOMINATOR = 100;
+
+ private final Random random = new Random(47);
+ private final double batchSizeBytes;
+
+ private long lastRecordSize = -1L;
+ private long totalSize;
+
+ BufferSizeDetector(double batchSizeMb) {
+ this.batchSizeBytes = batchSizeMb * 1024 * 1024;
+ }
+
+ void detect(Object record) {
+ if (record instanceof BinaryRowData) {
+ lastRecordSize = ((BinaryRowData) record).getSizeInBytes();
+ } else if (lastRecordSize == -1 || sampling()) {
+ lastRecordSize = ((FlussRowAsHudiRow) record).sizeInBytes();
+ }
+ totalSize += lastRecordSize;
+ }
+
+ boolean isFull() {
+ return totalSize > batchSizeBytes;
+ }
+
+ long getLastRecordSize() {
+ return lastRecordSize;
+ }
+
+ void reset() {
+ lastRecordSize = -1L;
+ totalSize = 0L;
+ }
+
+ private boolean sampling() {
+ return random.nextInt(DENOMINATOR) == 1;
+ }
+ }
+
+ private static class TotalSizeTracer {
+
+ private long bufferSize;
+ private final double maxBufferSize;
+
+ TotalSizeTracer(Configuration conf) {
+ long mergeReaderMem = 100;
+ long mergeMapMaxMem =
conf.getInteger(FlinkOptions.WRITE_MERGE_MAX_MEMORY);
+ this.maxBufferSize =
+ (conf.getDouble(FlinkOptions.WRITE_TASK_MAX_SIZE)
+ - mergeReaderMem
+ - mergeMapMaxMem)
+ * 1024
+ * 1024;
+ ValidationUtils.checkState(
+ maxBufferSize > 0,
+ String.format(
+ "'%s' should be greater than '%s' plus merge
reader memory.",
+ FlinkOptions.WRITE_TASK_MAX_SIZE.key(),
+ FlinkOptions.WRITE_MERGE_MAX_MEMORY.key()));
+ }
+
+ boolean trace(long recordSize) {
+ bufferSize += recordSize;
+ return bufferSize > maxBufferSize;
+ }
+
+ void countDown(long size) {
+ bufferSize -= size;
+ }
+
+ void reset() {
+ bufferSize = 0L;
+ }
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java
index 5c4a66214..84d5416dc 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java
@@ -319,6 +319,23 @@ public class HudiConversions {
return primaryKeys;
}
+ /** Converts Fluss change type to Flink row kind used by Hudi row data. */
+ public static RowKind toRowKind(ChangeType changeType) {
+ switch (changeType) {
+ case APPEND_ONLY:
+ case INSERT:
+ return RowKind.INSERT;
+ case UPDATE_BEFORE:
+ return RowKind.UPDATE_BEFORE;
+ case UPDATE_AFTER:
+ return RowKind.UPDATE_AFTER;
+ case DELETE:
+ return RowKind.DELETE;
+ default:
+ throw new IllegalArgumentException("Unsupported change type: "
+ changeType);
+ }
+ }
+
public static ChangeType toChangeType(RowKind rowKind) {
switch (rowKind) {
case INSERT:
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMessage.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMessage.java
new file mode 100644
index 000000000..7b40f0350
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMessage.java
@@ -0,0 +1,97 @@
+/*
+ * 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.utils.meta;
+
+import org.apache.hadoop.fs.FileStatus;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.apache.fluss.utils.Preconditions.checkState;
+
+/** A checkpoint message used to coordinate Hudi instant initialization across
tiering writers. */
+public class CkpMessage implements Serializable, Comparable<CkpMessage> {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final Comparator<CkpMessage> COMPARATOR =
+
Comparator.comparing(CkpMessage::getInstant).thenComparing(CkpMessage::getState);
+
+ private final String instant;
+ private final State state;
+
+ public CkpMessage(String instant, String state) {
+ this.instant = instant;
+ this.state = State.valueOf(state);
+ }
+
+ public CkpMessage(FileStatus fileStatus) {
+ String fileName = fileStatus.getPath().getName();
+ String[] nameAndExt = fileName.split("\\.");
+ checkState(nameAndExt.length == 2, "Invalid checkpoint metadata file:
%s", fileName);
+ this.instant = nameAndExt[0];
+ this.state = State.valueOf(nameAndExt[1]);
+ }
+
+ public String getInstant() {
+ return instant;
+ }
+
+ public State getState() {
+ return state;
+ }
+
+ public boolean isComplete() {
+ return State.COMPLETED == state;
+ }
+
+ public boolean isInflight() {
+ return State.INFLIGHT == state;
+ }
+
+ public static String getFileName(String instant, State state) {
+ return instant + "." + state.name();
+ }
+
+ public static List<String> getAllFileNames(String instant) {
+ List<String> fileNames = new ArrayList<>();
+ for (State state : State.values()) {
+ fileNames.add(getFileName(instant, state));
+ }
+ return fileNames;
+ }
+
+ @Override
+ public int compareTo(CkpMessage other) {
+ return COMPARATOR.compare(this, other);
+ }
+
+ /** Hudi instant state tracked by checkpoint metadata. */
+ public enum State {
+ INFLIGHT,
+ ABORTED,
+ COMPLETED
+ }
+
+ @Override
+ public String toString() {
+ return "CkpMessage{" + "instant='" + instant + '\'' + ", state=" +
state + '}';
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadata.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadata.java
new file mode 100644
index 000000000..59e9f6c45
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadata.java
@@ -0,0 +1,209 @@
+/*
+ * 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.utils.meta;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.storage.StoragePath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/** DFS-backed checkpoint metadata used as a lightweight message bus for Hudi
instants. */
+public class CkpMetadata implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CkpMetadata.class);
+
+ private static final int MAX_RETAIN_CKP_NUM = 1;
+ private static final String CKP_META = "ckp_meta";
+
+ private final FileSystem fs;
+ private final Path path;
+
+ private List<CkpMessage> messages;
+ private List<String> instantCache;
+
+ CkpMetadata(FileSystem fs, String basePath, String uniqueId) {
+ this.fs = fs;
+ this.path = new Path(ckpMetaPath(basePath, uniqueId));
+ }
+
+ public void bootstrap() throws IOException {
+ if (!fs.exists(path)) {
+ fs.mkdirs(path);
+ }
+ }
+
+ public void startInstant(String instant) {
+ Path instantPath = fullPath(CkpMessage.getFileName(instant,
CkpMessage.State.INFLIGHT));
+ try {
+ fs.createNewFile(instantPath);
+ } catch (IOException e) {
+ throw new HoodieException(
+ "Exception while adding checkpoint start metadata for
instant: " + instant, e);
+ }
+ cache(instant);
+ clean();
+ }
+
+ public void commitInstant(String instant) {
+ Path instantPath = fullPath(CkpMessage.getFileName(instant,
CkpMessage.State.COMPLETED));
+ try {
+ fs.createNewFile(instantPath);
+ } catch (IOException e) {
+ throw new HoodieException(
+ "Exception while adding checkpoint commit metadata for
instant: " + instant, e);
+ }
+ }
+
+ public void abortInstant(String instant) {
+ Path instantPath = fullPath(CkpMessage.getFileName(instant,
CkpMessage.State.ABORTED));
+ try {
+ fs.createNewFile(instantPath);
+ } catch (IOException e) {
+ throw new HoodieException(
+ "Exception while adding checkpoint abort metadata for
instant: " + instant, e);
+ }
+ }
+
+ @Nullable
+ public String lastPendingInstant() {
+ load();
+ if (!messages.isEmpty()) {
+ CkpMessage ckpMsg = messages.get(messages.size() - 1);
+ if (ckpMsg.isInflight()) {
+ return ckpMsg.getInstant();
+ }
+ }
+ return null;
+ }
+
+ public List<CkpMessage> getMessages() {
+ load();
+ return messages;
+ }
+
+ @Nullable
+ @VisibleForTesting
+ public String lastCompleteInstant() {
+ load();
+ for (int i = messages.size() - 1; i >= 0; i--) {
+ CkpMessage ckpMsg = messages.get(i);
+ if (ckpMsg.isComplete()) {
+ return ckpMsg.getInstant();
+ }
+ }
+ return null;
+ }
+
+ @VisibleForTesting
+ public List<String> getInstantCache() {
+ return instantCache;
+ }
+
+ @Override
+ public void close() {
+ instantCache = null;
+ }
+
+ private void cache(String newInstant) {
+ if (instantCache == null) {
+ instantCache = new ArrayList<>();
+ }
+ instantCache.add(newInstant);
+ }
+
+ private void clean() {
+ load();
+ if (messages.size() <= MAX_RETAIN_CKP_NUM) {
+ return;
+ }
+ String instant = messages.get(0).getInstant();
+ for (String fileName : CkpMessage.getAllFileNames(instant)) {
+ Path filePath = fullPath(fileName);
+ try {
+ fs.delete(filePath, false);
+ LOG.info("Delete checkpoint metadata {}", filePath);
+ } catch (IOException e) {
+ LOG.warn("Exception while cleaning the checkpoint metadata
file: {}", filePath, e);
+ }
+ }
+ }
+
+ private void load() {
+ try {
+ messages = scanCkpMetadata(path);
+ } catch (IOException e) {
+ throw new HoodieException(
+ "Exception while scanning the checkpoint metadata files
under path: " + path,
+ e);
+ }
+ }
+
+ private Path fullPath(String fileName) {
+ return new Path(path, fileName);
+ }
+
+ protected Stream<CkpMessage> fetchCkpMessages(Path ckpMetaPath) throws
IOException {
+ if (!fs.exists(ckpMetaPath)) {
+ return Stream.empty();
+ }
+ return Arrays.stream(fs.listStatus(ckpMetaPath)).map(CkpMessage::new);
+ }
+
+ protected List<CkpMessage> scanCkpMetadata(Path ckpMetaPath) throws
IOException {
+ return fetchCkpMessages(ckpMetaPath)
+ .collect(Collectors.groupingBy(CkpMessage::getInstant))
+ .values()
+ .stream()
+ .map(
+ ckpMessages ->
+ ckpMessages.stream()
+ .reduce(
+ (left, right) ->
+
left.getState().compareTo(right.getState())
+ >= 0
+ ? left
+ : right)
+ .get())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ protected static String ckpMetaPath(String basePath, String uniqueId) {
+ String metaPath =
+ basePath
+ + StoragePath.SEPARATOR
+ + HoodieTableMetaClient.AUXILIARYFOLDER_NAME
+ + StoragePath.SEPARATOR
+ + CKP_META;
+ return uniqueId == null || uniqueId.isEmpty() ? metaPath : metaPath +
"_" + uniqueId;
+ }
+}
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
new file mode 100644
index 000000000..40408b1cc
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.utils.meta;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+
+/** Factory for {@link CkpMetadata}. */
+public class CkpMetadataFactory {
+
+ private 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);
+ return new CkpMetadata(fs, basePath, uniqueId);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataProvider.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataProvider.java
new file mode 100644
index 000000000..8a8a5e48c
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataProvider.java
@@ -0,0 +1,65 @@
+/*
+ * 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.utils.meta;
+
+import org.apache.fluss.lake.hudi.tiering.HudiWriteTableInfo;
+import org.apache.fluss.metadata.TablePath;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Provides table-scoped Hudi checkpoint metadata. */
+public class CkpMetadataProvider implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient volatile Map<TablePath, CkpMetadata> ckpMetadatas;
+
+ public CkpMetadata get(TablePath tablePath, HudiWriteTableInfo
hudiTableInfo)
+ throws IOException {
+ Map<TablePath, CkpMetadata> metadataCache = getMetadataCache();
+ CkpMetadata ckpMetadata = metadataCache.get(tablePath);
+ if (ckpMetadata != null) {
+ return ckpMetadata;
+ }
+ synchronized (this) {
+ metadataCache = getMetadataCache();
+ ckpMetadata = metadataCache.get(tablePath);
+ if (ckpMetadata == null) {
+ ckpMetadata =
CkpMetadataFactory.getCkpMetadata(hudiTableInfo.getFlinkConfig());
+ metadataCache.put(tablePath, ckpMetadata);
+ }
+ return ckpMetadata;
+ }
+ }
+
+ private Map<TablePath, CkpMetadata> getMetadataCache() {
+ Map<TablePath, CkpMetadata> metadataCache = ckpMetadatas;
+ if (metadataCache != null) {
+ return metadataCache;
+ }
+ synchronized (this) {
+ if (ckpMetadatas == null) {
+ ckpMetadatas = new ConcurrentHashMap<>();
+ }
+ return ckpMetadatas;
+ }
+ }
+}
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
new file mode 100644
index 000000000..33c93d777
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/HudiWriteResultSerializerTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.utils.InstantiationUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HudiWriteResultSerializer}. */
+class HudiWriteResultSerializerTest {
+
+ @Test
+ void testSerializeAndDeserializeEmptyWriteResult() throws Exception {
+ HudiWriteResultSerializer serializer = new HudiWriteResultSerializer();
+ HudiWriteResult writeResult =
+ new HudiWriteResult(Collections.emptyMap(),
Collections.emptyMap());
+
+ HudiWriteResult deserialized =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(writeResult));
+
+ assertThat(deserialized.getWriteStatuses()).isEmpty();
+ assertThat(deserialized.getCompactionWriteStatuses()).isEmpty();
+ }
+
+ @Test
+ void testRejectUnsupportedVersion() {
+ HudiWriteResultSerializer serializer = new HudiWriteResultSerializer();
+
+ assertThatThrownBy(() ->
serializer.deserialize(serializer.getVersion() + 1, new byte[0]))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported HudiWriteResult version");
+ }
+
+ @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);
+ }
+
+ assertThatThrownBy(
+ () -> serializer.deserialize(serializer.getVersion(),
baos.toByteArray()))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Corrupted serialization: invalid
CompactionWriteResult");
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.java
new file mode 100644
index 000000000..1e291fc13
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/FlussRecordAsHudiRowTest.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.writer;
+
+import org.apache.fluss.record.GenericRecord;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BinaryType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LocalZonedTimestampType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+
+import static org.apache.fluss.record.ChangeType.APPEND_ONLY;
+import static org.apache.fluss.record.ChangeType.DELETE;
+import static org.apache.fluss.record.ChangeType.UPDATE_AFTER;
+import static org.apache.fluss.record.ChangeType.UPDATE_BEFORE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FlussRecordAsHudiRow}. */
+class FlussRecordAsHudiRowTest {
+
+ @Test
+ void testLogRecordFieldsAndSystemColumns() {
+ RowType rowType =
+ RowType.of(
+ new BooleanType(),
+ new IntType(),
+ new BigIntType(),
+ new VarCharType(),
+ new DecimalType(10, 2),
+ new LocalZonedTimestampType(6),
+ new TimestampType(6),
+ new BinaryType(),
+ new IntType(),
+ new BigIntType(),
+ new TimestampType(6));
+ GenericRow row = new GenericRow(8);
+ row.setField(0, true);
+ row.setField(1, 1);
+ row.setField(2, 2L);
+ row.setField(3, BinaryString.fromString("fluss"));
+ row.setField(4, Decimal.fromBigDecimal(new BigDecimal("12.34"), 10,
2));
+ row.setField(5, TimestampLtz.fromEpochMillis(1698235273182L, 5678));
+ row.setField(6, TimestampNtz.fromMillis(1698235273182L, 5678));
+ row.setField(7, new byte[] {1, 2, 3});
+
+ int bucket = 3;
+ long offset = 11L;
+ long timestamp = 1698235273999L;
+ LogRecord logRecord = new GenericRecord(offset, timestamp,
APPEND_ONLY, row);
+ FlussRecordAsHudiRow hudiRow = new FlussRecordAsHudiRow(bucket,
rowType);
+ hudiRow.setFlussRecord(logRecord);
+
+ assertThat(hudiRow.getBoolean(0)).isTrue();
+ assertThat(hudiRow.getInt(1)).isEqualTo(1);
+ assertThat(hudiRow.getLong(2)).isEqualTo(2L);
+ assertThat(hudiRow.getString(3).toString()).isEqualTo("fluss");
+ assertThat(hudiRow.getDecimal(4, 10, 2).toBigDecimal()).isEqualTo(new
BigDecimal("12.34"));
+ assertThat(hudiRow.getTimestamp(5,
6).getMillisecond()).isEqualTo(1698235273182L);
+ assertThat(hudiRow.getTimestamp(5,
6).getNanoOfMillisecond()).isEqualTo(5678);
+ assertThat(hudiRow.getTimestamp(6,
6).getMillisecond()).isEqualTo(1698235273182L);
+ assertThat(hudiRow.getBinary(7)).containsExactly(1, 2, 3);
+
+ assertThat(hudiRow.getInt(8)).isEqualTo(bucket);
+ assertThat(hudiRow.getLong(9)).isEqualTo(offset);
+ assertThat(hudiRow.getLong(10)).isEqualTo(timestamp);
+ assertThat(hudiRow.getTimestamp(10,
6).getMillisecond()).isEqualTo(timestamp);
+ assertThat(hudiRow.getArity()).isEqualTo(11);
+ assertThat(hudiRow.getRowKind()).isEqualTo(RowKind.INSERT);
+ }
+
+ @Test
+ void testChangeTypeToRowKind() {
+ RowType rowType =
+ RowType.of(
+ new BooleanType(), new IntType(), new BigIntType(),
new TimestampType(6));
+ GenericRow row = new GenericRow(1);
+ row.setField(0, true);
+ FlussRecordAsHudiRow hudiRow = new FlussRecordAsHudiRow(0, rowType);
+
+ hudiRow.setFlussRecord(new GenericRecord(0, 1, UPDATE_BEFORE, row));
+ assertThat(hudiRow.getRowKind()).isEqualTo(RowKind.UPDATE_BEFORE);
+
+ hudiRow.setFlussRecord(new GenericRecord(0, 1, UPDATE_AFTER, row));
+ assertThat(hudiRow.getRowKind()).isEqualTo(RowKind.UPDATE_AFTER);
+
+ hudiRow.setFlussRecord(new GenericRecord(0, 1, DELETE, row));
+ assertThat(hudiRow.getRowKind()).isEqualTo(RowKind.DELETE);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBufferTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBufferTest.java
new file mode 100644
index 000000000..140211d00
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/tiering/writer/RecordWriteBufferTest.java
@@ -0,0 +1,198 @@
+/*
+ * 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.writer;
+
+import org.apache.fluss.lake.writer.WriterInitContext;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.table.action.commit.BucketInfo;
+import org.apache.hudi.table.action.commit.BucketType;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/** Test for {@link RecordWriteBuffer}. */
+class RecordWriteBufferTest {
+
+ @Test
+ void testFlushesCurrentBucketWhenBucketInfoChanges() throws Exception {
+ TestingRecordWriteBuffer buffer = new TestingRecordWriteBuffer(false);
+ BucketInfo firstBucketInfo = bucketInfo("partition_a", "file_a",
BucketType.INSERT);
+ BucketInfo secondBucketInfo = bucketInfo("partition_b", "file_b",
BucketType.INSERT);
+
+ buffer.bufferRecord(row("partition_a", "file_a", "I"));
+ buffer.bufferRecord(row("partition_a", "file_a", "I"));
+ assertThat(buffer.flushedBucketInfos).isEmpty();
+
+ buffer.bufferRecord(row("partition_b", "file_b", "I"));
+ assertThat(buffer.flushedBucketInfos).containsExactly(firstBucketInfo);
+
+ buffer.flushRemaining();
+ assertThat(buffer.flushedBucketInfos).containsExactly(firstBucketInfo,
secondBucketInfo);
+ }
+
+ @Test
+ void testDisposesEmptyBucketBeforeRetryingRecord() throws Exception {
+ TestingRecordWriteBuffer buffer = new TestingRecordWriteBuffer(true);
+ BucketInfo bucketInfo = bucketInfo("partition", "file",
BucketType.INSERT);
+
+ buffer.bufferRecord(row("partition", "file", "I"));
+
+ assertThat(buffer.createdBuckets).hasSize(2);
+ assertThat(buffer.createdBuckets.get(0).disposed).isTrue();
+ assertThat(buffer.createdBuckets.get(1).isEmpty()).isFalse();
+ assertThat(buffer.flushedBucketInfos).isEmpty();
+
+ buffer.flushRemaining();
+ assertThat(buffer.flushedBucketInfos).containsExactly(bucketInfo);
+ }
+
+ private static HoodieFlinkInternalRow row(String partition, String fileId,
String instantTime) {
+ HoodieFlinkInternalRow internalRow =
+ new HoodieFlinkInternalRow(
+ "record_key", partition,
WriteOperationType.INSERT.value(), rowData());
+ internalRow.setFileId(fileId);
+ internalRow.setInstantTime(instantTime);
+ return internalRow;
+ }
+
+ private static RowData rowData() {
+ return mock(RowData.class);
+ }
+
+ private static BucketInfo bucketInfo(String partition, String fileId,
BucketType bucketType) {
+ return new BucketInfo(bucketType, fileId, partition);
+ }
+
+ @SuppressWarnings("rawtypes")
+ private static class TestingRecordWriteBuffer extends RecordWriteBuffer {
+
+ private final List<TestingDataBucket> createdBuckets = new
ArrayList<>();
+ private final List<BucketInfo> flushedBucketInfos = new ArrayList<>();
+ private boolean failNextBucketWrite;
+
+ private TestingRecordWriteBuffer(boolean failFirstBucketWrite) {
+ super(
+ null,
+ createConfig(),
+ mock(HoodieFlinkWriteClient.class),
+ null,
+ null,
+ WriterInitContext.UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ this.failNextBucketWrite = failFirstBucketWrite;
+ }
+
+ @Override
+ protected DataBucket createDataBucket(BucketInfo bucketInfo) {
+ TestingDataBucket dataBucket = new TestingDataBucket(bucketInfo,
failNextBucketWrite);
+ failNextBucketWrite = false;
+ createdBuckets.add(dataBucket);
+ return dataBucket;
+ }
+
+ @Override
+ protected List<WriteStatus> writeRecords(
+ String instant, String writeOperation, DataBucket dataBucket) {
+ flushedBucketInfos.add(dataBucket.getBucketInfo());
+ return Collections.emptyList();
+ }
+
+ @Override
+ public String getLastPendingInstant() {
+ return "001";
+ }
+ }
+
+ private static Configuration createConfig() {
+ Configuration config = new Configuration();
+ config.set(FlinkOptions.OPERATION, WriteOperationType.INSERT.value());
+ config.set(FlinkOptions.WRITE_TASK_MAX_SIZE, 128.0);
+ config.set(FlinkOptions.WRITE_BATCH_SIZE, 128.0);
+ config.set(FlinkOptions.WRITE_MERGE_MAX_MEMORY, 1);
+ return config;
+ }
+
+ private static class TestingDataBucket extends
RecordWriteBuffer.DataBucket {
+
+ private final boolean failFirstWrite;
+
+ private boolean failed;
+ private boolean empty = true;
+ private boolean disposed;
+
+ private TestingDataBucket(BucketInfo bucketInfo, boolean
failFirstWrite) {
+ super(null, bucketInfo, 128.0);
+ this.failFirstWrite = failFirstWrite;
+ }
+
+ @Override
+ public MutableObjectIterator<BinaryRowData> getDataIterator() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean writeRow(RowData rowData) throws IOException {
+ if (failFirstWrite && !failed) {
+ failed = true;
+ return false;
+ }
+ empty = false;
+ return true;
+ }
+
+ @Override
+ public long getBufferSize() {
+ return empty ? 0L : 1L;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return empty;
+ }
+
+ @Override
+ public boolean isFull() {
+ return false;
+ }
+
+ @Override
+ public long getLastRecordSize() {
+ return 1L;
+ }
+
+ @Override
+ public void dispose() {
+ disposed = true;
+ empty = true;
+ }
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataTest.java
new file mode 100644
index 000000000..8fae89dcc
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/meta/CkpMetadataTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.utils.meta;
+
+import org.apache.fluss.utils.InstantiationUtils;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.Serializable;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link CkpMetadata}. */
+class CkpMetadataTest {
+
+ @TempDir private Path tempDir;
+
+ @Test
+ void testAbortedInstantIsNotPending() throws Exception {
+ FileSystem fs = FileSystem.getLocal(new
org.apache.hadoop.conf.Configuration());
+ CkpMetadata ckpMetadata = new CkpMetadata(fs, tempDir.toString(), "");
+
+ assertThat(ckpMetadata).isNotInstanceOf(Serializable.class);
+ ckpMetadata.bootstrap();
+ ckpMetadata.startInstant("20260622000100000");
+
assertThat(ckpMetadata.lastPendingInstant()).isEqualTo("20260622000100000");
+
+ ckpMetadata.abortInstant("20260622000100000");
+ assertThat(ckpMetadata.lastPendingInstant()).isNull();
+ }
+
+ @Test
+ void testCkpMetadataProviderSerializable() throws Exception {
+ CkpMetadataProvider ckpMetadataProvider = new CkpMetadataProvider();
+
+ byte[] serialized =
InstantiationUtils.serializeObject(ckpMetadataProvider);
+ CkpMetadataProvider deserialized =
+ InstantiationUtils.deserializeObject(serialized,
getClass().getClassLoader());
+
+ assertThat(deserialized).isNotNull();
+ }
+}
