You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/06/22 13:57:21 UTC
(fluss) branch main updated: [lake/hudi] Introduce Hudi lake writer support for tiering tables (#3507)
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 <aa...@gmail.com>
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();
+ }
+}