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 03dad8871 [lake/hudi] Introduce Hudi source split planner (#3456)
03dad8871 is described below
commit 03dad8871bebed1bb7f27950d395e1690ff152c6
Author: fhan <[email protected]>
AuthorDate: Mon Jun 15 14:21:17 2026 +0800
[lake/hudi] Introduce Hudi source split planner (#3456)
---
.../apache/fluss/lake/hudi/HudiLakeStorage.java | 16 +-
.../fluss/lake/hudi/source/HudiLakeSource.java | 76 +++++
.../apache/fluss/lake/hudi/source/HudiSplit.java | 122 +++++++
.../fluss/lake/hudi/source/HudiSplitPlanner.java | 183 +++++++++++
.../lake/hudi/source/HudiSplitSerializer.java | 61 ++++
.../fluss/lake/hudi/utils/HudiConversions.java | 11 +-
.../fluss/lake/hudi/utils/HudiTableInfo.java | 361 +++++++++++++++++++++
.../fluss/lake/hudi/source/HudiLakeSourceTest.java | 86 +++++
.../lake/hudi/source/HudiSplitPlannerTest.java | 304 +++++++++++++++++
.../lake/hudi/source/HudiSplitSerializerTest.java | 73 +++++
.../fluss/lake/hudi/source/HudiSplitTest.java | 77 +++++
.../fluss/lake/hudi/utils/HudiConversionsTest.java | 94 ++++++
.../fluss/lake/hudi/utils/HudiTableInfoTest.java | 118 +++++++
13 files changed, 1571 insertions(+), 11 deletions(-)
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
index 66db56057..55ce818a9 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java
@@ -18,6 +18,8 @@
package org.apache.fluss.lake.hudi;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.hudi.source.HudiLakeSource;
+import org.apache.fluss.lake.hudi.source.HudiSplit;
import org.apache.fluss.lake.lakestorage.LakeCatalog;
import org.apache.fluss.lake.lakestorage.LakeStorage;
import org.apache.fluss.lake.source.LakeSource;
@@ -36,9 +38,8 @@ public class HudiLakeStorage implements LakeStorage {
@Override
public LakeTieringFactory<?, ?> createLakeTieringFactory() {
throw new UnsupportedOperationException(
- "HudiLakeStorage is currently a scaffold and does not support
creating a "
- + "LakeTieringFactory yet. Verify that Hudi lake
storage was selected "
- + "intentionally and that the required Hudi
support/module is available.");
+ "Hudi lake tiering writer is not implemented yet, so
HudiLakeStorage does not "
+ + "support creating a LakeTieringFactory.");
}
@Override
@@ -47,12 +48,7 @@ public class HudiLakeStorage implements LakeStorage {
}
@Override
- public LakeSource<?> createLakeSource(TablePath tablePath) {
- throw new UnsupportedOperationException(
- "HudiLakeStorage is currently a scaffold and does not support
creating a "
- + "LakeSource for table '"
- + tablePath
- + "' yet. Verify that Hudi lake storage was selected
intentionally "
- + "and that the required Hudi support/module is
available.");
+ public LakeSource<HudiSplit> createLakeSource(TablePath tablePath) {
+ return new HudiLakeSource(hudiConfig, tablePath);
}
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
new file mode 100644
index 000000000..c491ddfa6
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
@@ -0,0 +1,76 @@
+/*
+ * 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.source;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
+import org.apache.fluss.lake.source.LakeSource;
+import org.apache.fluss.lake.source.Planner;
+import org.apache.fluss.lake.source.RecordReader;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Hudi implementation of {@link LakeSource}. */
+public class HudiLakeSource implements LakeSource<HudiSplit> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Configuration hudiConfig;
+ private final TablePath tablePath;
+
+ public HudiLakeSource(Configuration hudiConfig, TablePath tablePath) {
+ this.hudiConfig = hudiConfig;
+ this.tablePath = tablePath;
+ }
+
+ @Override
+ public void withProject(int[][] project) {
+ // Projection is applied by the Hudi record reader, which is not
implemented yet.
+ }
+
+ @Override
+ public void withLimit(int limit) {
+ throw new UnsupportedOperationException("Hudi lake source does not
support limit yet.");
+ }
+
+ @Override
+ public FilterPushDownResult withFilters(List<Predicate> predicates) {
+ return FilterPushDownResult.of(Collections.emptyList(), new
ArrayList<>(predicates));
+ }
+
+ @Override
+ public Planner<HudiSplit> createPlanner(PlannerContext context) throws
IOException {
+ return new HudiSplitPlanner(hudiConfig, tablePath,
context.snapshotId());
+ }
+
+ @Override
+ public RecordReader createRecordReader(ReaderContext<HudiSplit> context)
throws IOException {
+ throw new UnsupportedOperationException(
+ "Hudi lake source does not support record reading yet.");
+ }
+
+ @Override
+ public SimpleVersionedSerializer<HudiSplit> getSplitSerializer() {
+ return new HudiSplitSerializer();
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplit.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplit.java
new file mode 100644
index 000000000..3b51d823f
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplit.java
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.source;
+
+import org.apache.fluss.lake.source.LakeSplit;
+
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/** A readable split of a Hudi table. */
+public class HudiSplit implements LakeSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileSlice fileSlice;
+ private final int bucket;
+ private final List<String> partition;
+
+ public HudiSplit(FileSlice fileSlice, int bucket, List<String> partition) {
+ this.fileSlice = Objects.requireNonNull(fileSlice, "fileSlice cannot
be null");
+ this.bucket = bucket;
+ this.partition =
+ Collections.unmodifiableList(
+ new ArrayList<>(
+ Objects.requireNonNull(partition, "partition
cannot be null")));
+ }
+
+ @Override
+ public int bucket() {
+ return bucket;
+ }
+
+ @Override
+ public List<String> partition() {
+ return partition;
+ }
+
+ public FileSlice getFileSlice() {
+ return fileSlice;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof HudiSplit)) {
+ return false;
+ }
+ HudiSplit hudiSplit = (HudiSplit) o;
+ return bucket == hudiSplit.bucket
+ && equalsFileSlice(fileSlice, hudiSplit.fileSlice)
+ && Objects.equals(partition, hudiSplit.partition);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ fileSlice.getFileGroupId(),
+ fileSlice.getBaseInstantTime(),
+ baseFilePath(fileSlice),
+ logFilePaths(fileSlice),
+ bucket,
+ partition);
+ }
+
+ @Override
+ public String toString() {
+ return "HudiSplit{"
+ + "fileSlice="
+ + fileSlice
+ + ", bucket="
+ + bucket
+ + ", partition="
+ + partition
+ + '}';
+ }
+
+ private static boolean equalsFileSlice(FileSlice first, FileSlice second) {
+ return Objects.equals(first.getFileGroupId(), second.getFileGroupId())
+ && Objects.equals(first.getBaseInstantTime(),
second.getBaseInstantTime())
+ && Objects.equals(baseFilePath(first), baseFilePath(second))
+ && Objects.equals(logFilePaths(first), logFilePaths(second));
+ }
+
+ private static String baseFilePath(FileSlice fileSlice) {
+ if (!fileSlice.getBaseFile().isPresent()) {
+ return null;
+ }
+ HoodieBaseFile baseFile = fileSlice.getBaseFile().get();
+ return baseFile.getPath();
+ }
+
+ private static List<String> logFilePaths(FileSlice fileSlice) {
+ return fileSlice
+ .getLogFiles()
+ .map(logFile -> logFile.getPath().toString())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java
new file mode 100644
index 000000000..b57c984b0
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.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.source;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.hudi.utils.HudiTableInfo;
+import org.apache.fluss.lake.source.Planner;
+import org.apache.fluss.metadata.TablePath;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.index.bucket.BucketIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.stream.Collectors;
+
+/** Planner for creating Hudi splits. */
+public class HudiSplitPlanner implements Planner<HudiSplit> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(HudiSplitPlanner.class);
+
+ private final Configuration hudiConfig;
+ private final TablePath tablePath;
+ private final long snapshotId;
+
+ public HudiSplitPlanner(Configuration hudiConfig, TablePath tablePath,
long snapshotId) {
+ this.hudiConfig = hudiConfig;
+ this.tablePath = tablePath;
+ this.snapshotId = snapshotId;
+ }
+
+ @Override
+ public List<HudiSplit> plan() throws IOException {
+ String snapshotTime = toHudiInstantTime(snapshotId);
+ try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath,
hudiConfig)) {
+ if
(!hudiTableInfo.getCompletedTimeline().containsInstant(snapshotTime)) {
+ throw new IOException(
+ String.format(
+ "Hudi instant time %s does not exist in table
%s.",
+ snapshotTime, tablePath));
+ }
+
+ List<String> partitionPaths =
+ FSUtils.getAllPartitionPaths(
+ hudiTableInfo.getEngineContext(),
hudiTableInfo.getMetaClient(), false);
+ if (partitionPaths.isEmpty()) {
+ if (hudiTableInfo.isPartitioned()) {
+ LOG.info(
+ "Planned no Hudi splits for partitioned table {}
with table type {} at instant {} because no Hudi partition paths were
discovered.",
+ tablePath,
+ hudiTableInfo.getTableType(),
+ snapshotTime);
+ return Collections.emptyList();
+ } else {
+ partitionPaths = Collections.singletonList("");
+ }
+ }
+
+ List<HudiSplit> splits = new ArrayList<>();
+ for (String partitionPath : partitionPaths) {
+ splits.addAll(planPartition(hudiTableInfo, snapshotTime,
partitionPath));
+ }
+ if (splits.isEmpty()) {
+ LOG.info(
+ "Planned no Hudi splits for table {} at instant {}
across {} Hudi partition paths.",
+ tablePath,
+ snapshotTime,
+ partitionPaths.size());
+ } else {
+ LOG.debug(
+ "Planned {} Hudi splits for table {} at instant {}.",
+ splits.size(),
+ tablePath,
+ snapshotTime);
+ }
+ return splits;
+ }
+ }
+
+ @VisibleForTesting
+ static String toHudiInstantTime(long snapshotId) {
+ return String.format(Locale.ROOT, "%017d", snapshotId);
+ }
+
+ private List<HudiSplit> planPartition(
+ HudiTableInfo hudiTableInfo, String snapshotTime, String
partitionPath)
+ throws IOException {
+ HoodieTableFileSystemView fileSystemView =
hudiTableInfo.getFileSystemView();
+ List<HudiSplit> splits = new ArrayList<>();
+ if (hudiTableInfo.getTableType() == HoodieTableType.MERGE_ON_READ) {
+ List<FileSlice> fileSlices =
+ fileSystemView
+
.getLatestMergedFileSlicesBeforeOrOn(partitionPath, snapshotTime)
+ .collect(Collectors.toList());
+ for (FileSlice fileSlice : fileSlices) {
+ splits.add(toHudiSplit(hudiTableInfo, partitionPath,
fileSlice));
+ }
+ return splits;
+ } else if (hudiTableInfo.getTableType() ==
HoodieTableType.COPY_ON_WRITE) {
+ List<HoodieBaseFile> baseFiles =
+ fileSystemView
+ .getLatestBaseFilesBeforeOrOn(partitionPath,
snapshotTime)
+ .collect(Collectors.toList());
+ for (HoodieBaseFile baseFile : baseFiles) {
+ splits.add(
+ toHudiSplit(
+ hudiTableInfo,
+ partitionPath,
+ toFileSlice(partitionPath, baseFile)));
+ }
+ return splits;
+ }
+ throw new IOException(
+ String.format(
+ "Unsupported Hudi table type %s for table %s.",
+ hudiTableInfo.getTableType(), tablePath));
+ }
+
+ private FileSlice toFileSlice(String partitionPath, HoodieBaseFile
baseFile) {
+ return new FileSlice(
+ new HoodieFileGroupId(partitionPath, baseFile.getFileId()),
+ baseFile.getCommitTime(),
+ baseFile,
+ Collections.emptyList());
+ }
+
+ private HudiSplit toHudiSplit(
+ HudiTableInfo hudiTableInfo, String partitionPath, FileSlice
fileSlice)
+ throws IOException {
+ return new HudiSplit(
+ fileSlice,
+ extractBucket(hudiTableInfo, fileSlice),
+ hudiTableInfo.partitionValues(partitionPath));
+ }
+
+ private int extractBucket(HudiTableInfo hudiTableInfo, FileSlice
fileSlice) throws IOException {
+ if (!hudiTableInfo.isBucketAware()) {
+ return -1;
+ }
+ String fileId = fileSlice.getFileGroupId().getFileId();
+ if (fileId == null || fileId.isEmpty()) {
+ throw new IOException(
+ String.format(
+ "Failed to extract Hudi bucket id for bucket-aware
table %s because file id is empty.",
+ tablePath));
+ }
+ try {
+ return BucketIdentifier.bucketIdFromFileId(fileId);
+ } catch (RuntimeException e) {
+ throw new IOException(
+ String.format(
+ "Failed to extract Hudi bucket id from file id
'%s' for bucket-aware table %s.",
+ fileId, tablePath),
+ e);
+ }
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializer.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializer.java
new file mode 100644
index 000000000..d9578bc8c
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializer.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.source;
+
+import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
+import org.apache.fluss.utils.InstantiationUtils;
+
+import java.io.IOException;
+
+/** Serializer for {@link HudiSplit}. */
+public class HudiSplitSerializer implements
SimpleVersionedSerializer<HudiSplit> {
+
+ private static final int CURRENT_VERSION = 1;
+
+ @Override
+ public int getVersion() {
+ return CURRENT_VERSION;
+ }
+
+ @Override
+ public byte[] serialize(HudiSplit hudiSplit) throws IOException {
+ return InstantiationUtils.serializeObject(hudiSplit);
+ }
+
+ @Override
+ public HudiSplit deserialize(int version, byte[] serialized) throws
IOException {
+ if (version != CURRENT_VERSION) {
+ throw new IOException("Unsupported HudiSplit serialization
version: " + version);
+ }
+ try {
+ Object deserialized =
+ InstantiationUtils.deserializeObject(serialized,
getClass().getClassLoader());
+ if (!(deserialized instanceof HudiSplit)) {
+ throw new IOException(
+ "Expected HudiSplit but found "
+ + (deserialized == null
+ ? "null"
+ : deserialized.getClass().getName())
+ + " during deserialization.");
+ }
+ return (HudiSplit) deserialized;
+ } catch (ClassNotFoundException e) {
+ throw new IOException("Failed to deserialize HudiSplit.", e);
+ }
+ }
+}
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 ae98201d0..2f90124e5 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
@@ -66,6 +66,10 @@ public class HudiConversions {
private static final String HUDI_RECORD_KEY_FIELD_OPTION =
HUDI_CONF_PREFIX + FlinkOptions.RECORD_KEY_FIELD.key();
+ public static final String FLUSS_BUCKET_KEYS_OPTION = "fluss.bucket.keys";
+ public static final String FLUSS_BUCKET_AWARE_OPTION =
"fluss.bucket-aware";
+ public static final String FLUSS_PARTITION_KEYS_OPTION =
"fluss.partition.keys";
+
/** Hudi config options set by Fluss should not be set by users. */
@VisibleForTesting public static final Set<String> HUDI_UNSETTABLE_OPTIONS
= new HashSet<>();
@@ -181,7 +185,7 @@ public class HudiConversions {
recordKeyField); // use primary key as index key
}
- // buket keys column
+ // bucket keys column
hudiProperties.put(FlinkOptions.INDEX_TYPE.key(),
HoodieIndex.IndexType.BUCKET.name());
List<String> bucketKeys = tableDescriptor.getBucketKeys();
int numBuckets =
@@ -210,6 +214,11 @@ public class HudiConversions {
.getCustomProperties()
.forEach((k, v) -> setFlussPropertyToHudi(k, v,
hudiProperties));
+ hudiProperties.put(FLUSS_BUCKET_KEYS_OPTION, String.join(DELIMITER,
bucketKeys));
+ hudiProperties.put(
+ FLUSS_BUCKET_AWARE_OPTION, String.valueOf(isPkTable ||
!bucketKeys.isEmpty()));
+ hudiProperties.put(FLUSS_PARTITION_KEYS_OPTION, String.join(DELIMITER,
partitionKeys));
+
return hudiProperties;
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java
new file mode 100644
index 000000000..cb819fab2
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java
@@ -0,0 +1,361 @@
+/*
+ * 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;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils;
+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.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.model.HoodieTableType;
+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.configuration.FlinkOptions;
+import org.apache.hudi.configuration.HadoopConfigurations;
+import org.apache.hudi.exception.TableNotFoundException;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.apache.hudi.table.catalog.CatalogOptions;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION;
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION;
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_PARTITION_KEYS_OPTION;
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.toHudiObjectPath;
+
+/** Resolved Hudi table metadata used by lake source planning. */
+public class HudiTableInfo implements AutoCloseable {
+
+ private static final String DEFAULT_HUDI_WAREHOUSE = "/tmp/hudi_warehouse";
+ private static final String DELIMITER = ",";
+
+ private final TablePath tablePath;
+ private final Catalog hudiCatalog;
+ private final CatalogBaseTable hudiTable;
+ private final Map<String, String> tableOptions;
+ private final HoodieTableMetaClient metaClient;
+ private final HoodieEngineContext engineContext;
+ private final HoodieTimeline completedTimeline;
+ private final HoodieTableFileSystemView fileSystemView;
+ private final HoodieTableType tableType;
+ private final String basePath;
+ private final List<String> partitionFields;
+ private final boolean bucketAware;
+ private boolean closed;
+
+ private HudiTableInfo(
+ TablePath tablePath,
+ Catalog hudiCatalog,
+ CatalogBaseTable hudiTable,
+ Map<String, String> tableOptions,
+ HoodieTableMetaClient metaClient,
+ HoodieEngineContext engineContext,
+ HoodieTimeline completedTimeline,
+ HoodieTableFileSystemView fileSystemView,
+ HoodieTableType tableType,
+ String basePath,
+ List<String> partitionFields,
+ boolean bucketAware) {
+ this.tablePath = tablePath;
+ this.hudiCatalog = hudiCatalog;
+ this.hudiTable = hudiTable;
+ this.tableOptions = tableOptions;
+ this.metaClient = metaClient;
+ this.engineContext = engineContext;
+ this.completedTimeline = completedTimeline;
+ this.fileSystemView = fileSystemView;
+ this.tableType = tableType;
+ this.basePath = basePath;
+ this.partitionFields = partitionFields;
+ this.bucketAware = bucketAware;
+ }
+
+ public static HudiTableInfo create(TablePath tablePath, Configuration
hudiConfig)
+ throws IOException {
+ Catalog hudiCatalog = HudiCatalogUtils.createHudiCatalog(hudiConfig);
+ hudiCatalog.open();
+ try {
+ CatalogBaseTable hudiTable =
hudiCatalog.getTable(toHudiObjectPath(tablePath));
+ Map<String, String> tableOptions = new
HashMap<>(hudiTable.getOptions());
+ tableOptions.putAll(hudiConfig.toMap());
+
+ String basePath = resolveBasePath(tablePath, tableOptions);
+ tableOptions.put(FlinkOptions.PATH.key(), basePath);
+
+ HoodieTableFileSystemView fileSystemView = null;
+ HoodieTableMetaClient metaClient = createMetaClient(basePath,
hudiConfig);
+ HoodieTimeline completedTimeline =
+ metaClient.getCommitsTimeline().filterCompletedInstants();
+ HoodieEngineContext engineContext =
+ new HoodieLocalEngineContext(metaClient.getStorageConf());
+ try {
+ HoodieTimeline fileSystemViewTimeline =
+ metaClient
+ .getCommitsAndCompactionTimeline()
+ .filterCompletedAndCompactionInstants();
+ fileSystemView =
+
HoodieTableFileSystemView.fileListingBasedFileSystemView(
+ engineContext, metaClient,
fileSystemViewTimeline);
+ HoodieTableType tableType = metaClient.getTableType();
+ List<String> partitionFields =
+ splitCommaSeparated(
+ tableOptions.getOrDefault(
+ FLUSS_PARTITION_KEYS_OPTION,
+
tableOptions.get(FlinkOptions.PARTITION_PATH_FIELD.key())));
+ boolean bucketAware = resolveBucketAware(tableOptions,
tableType);
+
+ return new HudiTableInfo(
+ tablePath,
+ hudiCatalog,
+ hudiTable,
+ Collections.unmodifiableMap(new
HashMap<>(tableOptions)),
+ metaClient,
+ engineContext,
+ completedTimeline,
+ fileSystemView,
+ tableType,
+ basePath,
+ Collections.unmodifiableList(new
ArrayList<>(partitionFields)),
+ bucketAware);
+ } catch (Exception e) {
+ closeFileSystemView(fileSystemView);
+ throw e;
+ }
+ } catch (Exception e) {
+ closeCatalog(hudiCatalog);
+ if (e instanceof IOException) {
+ throw (IOException) e;
+ }
+ throw new IOException("Failed to resolve Hudi table info for " +
tablePath + ".", e);
+ }
+ }
+
+ 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();
+ }
+
+ @VisibleForTesting
+ public static boolean resolveBucketAware(
+ Map<String, String> tableOptions, HoodieTableType tableType) {
+ String bucketAware = tableOptions.get(FLUSS_BUCKET_AWARE_OPTION);
+ if (bucketAware != null) {
+ return Boolean.parseBoolean(bucketAware.trim());
+ }
+ String bucketKeys = tableOptions.get(FLUSS_BUCKET_KEYS_OPTION);
+ if (bucketKeys != null) {
+ return !bucketKeys.trim().isEmpty();
+ }
+ return tableType == HoodieTableType.MERGE_ON_READ;
+ }
+
+ @VisibleForTesting
+ public static List<String> extractPartitionValues(
+ String partitionPath, List<String> partitionFields) {
+ if (partitionFields.isEmpty() || partitionPath == null ||
partitionPath.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ String[] pathSegments = partitionPath.split("/");
+ if (pathSegments.length != partitionFields.size()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Hudi partition path '%s' does not match partition
fields %s.",
+ partitionPath, partitionFields));
+ }
+
+ boolean hiveStyle = false;
+ boolean rawStyle = false;
+ Map<String, String> hivePartitionValues = new HashMap<>();
+ List<String> rawPartitionValues = new ArrayList<>(pathSegments.length);
+ for (String pathSegment : pathSegments) {
+ int valueStart = pathSegment.indexOf('=');
+ if (valueStart > 0) {
+ hiveStyle = true;
+ String field = pathSegment.substring(0, valueStart);
+ if (!partitionFields.contains(field)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Hudi partition path '%s' contains unknown
partition field '%s'.",
+ partitionPath, field));
+ }
+ if (hivePartitionValues.put(field,
pathSegment.substring(valueStart + 1)) != null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Hudi partition path '%s' contains
duplicate partition field '%s'.",
+ partitionPath, field));
+ }
+ } else {
+ rawStyle = true;
+ rawPartitionValues.add(pathSegment);
+ }
+ }
+
+ if (hiveStyle && rawStyle) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Hudi partition path '%s' mixes hive-style and raw
partition segments.",
+ partitionPath));
+ }
+
+ if (!hiveStyle) {
+ return rawPartitionValues;
+ }
+
+ List<String> partitionValues = new ArrayList<>(partitionFields.size());
+ for (String partitionField : partitionFields) {
+ String partitionValue = hivePartitionValues.get(partitionField);
+ if (partitionValue == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Hudi partition path '%s' does not contain
partition field '%s'.",
+ partitionPath, partitionField));
+ }
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ private static List<String> splitCommaSeparated(String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return Collections.emptyList();
+ }
+ String[] parts = value.split(DELIMITER);
+ List<String> values = new ArrayList<>(parts.length);
+ for (String part : parts) {
+ String trimmed = part.trim();
+ if (!trimmed.isEmpty()) {
+ values.add(trimmed);
+ }
+ }
+ return values;
+ }
+
+ private static String ensureEndsWithSlash(String path) {
+ return path.endsWith("/") ? path : path + "/";
+ }
+
+ private static void closeCatalog(Catalog hudiCatalog) {
+ if (hudiCatalog instanceof AutoCloseable) {
+ IOUtils.closeQuietly((AutoCloseable) hudiCatalog, "hudi catalog");
+ }
+ }
+
+ private static void closeFileSystemView(HoodieTableFileSystemView
fileSystemView) {
+ if (fileSystemView != null) {
+ IOUtils.closeQuietly(fileSystemView::close, "hudi table file
system view");
+ }
+ }
+
+ public TablePath getTablePath() {
+ return tablePath;
+ }
+
+ public CatalogBaseTable getHudiTable() {
+ return hudiTable;
+ }
+
+ public Map<String, String> getTableOptions() {
+ return tableOptions;
+ }
+
+ public HoodieTableMetaClient getMetaClient() {
+ return metaClient;
+ }
+
+ public HoodieEngineContext getEngineContext() {
+ return engineContext;
+ }
+
+ public HoodieTimeline getCompletedTimeline() {
+ return completedTimeline;
+ }
+
+ public HoodieTableFileSystemView getFileSystemView() {
+ return fileSystemView;
+ }
+
+ public HoodieTableType getTableType() {
+ return tableType;
+ }
+
+ public String getBasePath() {
+ return basePath;
+ }
+
+ public boolean isBucketAware() {
+ return bucketAware;
+ }
+
+ public boolean isPartitioned() {
+ return !partitionFields.isEmpty();
+ }
+
+ public List<String> partitionValues(String partitionPath) {
+ return extractPartitionValues(partitionPath, partitionFields);
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ closeFileSystemView(fileSystemView);
+ closeCatalog(hudiCatalog);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
new file mode 100644
index 000000000..1b0e2fe89
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.source;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.source.LakeSource;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
+import org.apache.fluss.predicate.PredicateBuilder;
+import org.apache.fluss.types.IntType;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HudiLakeSource}. */
+class HudiLakeSourceTest {
+
+ @Test
+ void testCreatePlannerReturnsHudiSplitPlanner() throws Exception {
+ HudiLakeSource source =
+ new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
+
+ assertThat(source.createPlanner(() ->
1L)).isInstanceOf(HudiSplitPlanner.class);
+ }
+
+ @Test
+ void testGetSplitSerializerReturnsHudiSplitSerializer() {
+ HudiLakeSource source =
+ new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
+
+
assertThat(source.getSplitSerializer()).isInstanceOf(HudiSplitSerializer.class);
+ }
+
+ @Test
+ void testFiltersRemainWhenPushDownIsNotSupportedYet() {
+ HudiLakeSource source =
+ new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
+ Predicate predicate = new PredicateBuilder(RowType.of(new
IntType())).equal(0, 1);
+
+ LakeSource.FilterPushDownResult result =
+ source.withFilters(Collections.singletonList(predicate));
+
+ assertThat(result.acceptedPredicates()).isEmpty();
+ assertThat(result.remainingPredicates()).containsExactly(predicate);
+ }
+
+ @Test
+ void testWithLimitIsExplicitlyUnsupported() {
+ HudiLakeSource source =
+ new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
+
+ assertThatThrownBy(() -> source.withLimit(1))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("limit");
+ }
+
+ @Test
+ void testCreateRecordReaderIsExplicitlyUnsupported() {
+ HudiLakeSource source =
+ new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
+
+ assertThatThrownBy(() -> source.createRecordReader(() -> null))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("record reading");
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitPlannerTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitPlannerTest.java
new file mode 100644
index 000000000..d3293dc0c
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitPlannerTest.java
@@ -0,0 +1,304 @@
+/*
+ * 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.source;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.lake.hudi.HudiLakeCatalog;
+import org.apache.fluss.lake.hudi.utils.HudiTableInfo;
+import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.types.DataTypes;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.index.bucket.BucketIdentifier;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HudiSplitPlanner}. */
+class HudiSplitPlannerTest {
+
+ private static final String INSTANT_TIME = "20260608010101000";
+ private static final String LEADING_ZERO_INSTANT_TIME =
"02026060801010100";
+ private static final String DATABASE = "db1";
+ private static final String TABLE = "table1";
+
+ @TempDir private File tempWarehouseDir;
+
+ private Configuration hudiConfig;
+
+ @BeforeEach
+ void setUp() {
+ hudiConfig = new Configuration();
+ hudiConfig.setString("catalog.path",
tempWarehouseDir.toURI().toString());
+ hudiConfig.setString("mode", "dfs");
+ }
+
+ @Test
+ void testPlanCopyOnWriteSplitsForCompletedInstant() throws Exception {
+ TablePath tablePath = TablePath.of(DATABASE, TABLE);
+ createCowTable(tablePath);
+ String partitionPath = "";
+ int bucket = 1;
+ String fileId = BucketIdentifier.newBucketFileIdPrefix(bucket);
+ String fileName = createBaseFile(tablePath, partitionPath,
INSTANT_TIME, fileId);
+ createCompletedCommit(tablePath, partitionPath, INSTANT_TIME, fileId,
fileName);
+ assertHudiFixtureVisible(tablePath, partitionPath);
+
+ List<HudiSplit> splits =
+ new HudiSplitPlanner(hudiConfig, tablePath,
Long.parseLong(INSTANT_TIME)).plan();
+
+ assertThat(splits).hasSize(1);
+ HudiSplit split = splits.get(0);
+ assertThat(split.bucket()).isEqualTo(bucket);
+ assertThat(split.partition()).isEmpty();
+
assertThat(split.getFileSlice().getPartitionPath()).isEqualTo(partitionPath);
+ assertThat(split.getFileSlice().getFileId()).isEqualTo(fileId);
+ assertThat(split.getFileSlice().getBaseFile().isPresent()).isTrue();
+ }
+
+ @Test
+ void testPlanFailsWhenInstantDoesNotExist() throws Exception {
+ TablePath tablePath = TablePath.of(DATABASE, TABLE);
+ createCowTable(tablePath);
+
+ assertThatThrownBy(
+ () ->
+ new HudiSplitPlanner(
+ hudiConfig, tablePath,
Long.parseLong(INSTANT_TIME))
+ .plan())
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("does not exist");
+ }
+
+ @Test
+ void testPlanLeadingZeroInstantTime() throws Exception {
+ TablePath tablePath = TablePath.of(DATABASE, TABLE);
+ createCowTable(tablePath);
+ String partitionPath = "";
+ int bucket = 1;
+ String fileId = BucketIdentifier.newBucketFileIdPrefix(bucket);
+ String fileName =
+ createBaseFile(tablePath, partitionPath,
LEADING_ZERO_INSTANT_TIME, fileId);
+ createCompletedCommit(
+ tablePath, partitionPath, LEADING_ZERO_INSTANT_TIME, fileId,
fileName);
+
+ List<HudiSplit> splits =
+ new HudiSplitPlanner(
+ hudiConfig, tablePath,
Long.parseLong(LEADING_ZERO_INSTANT_TIME))
+ .plan();
+
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0).getFileSlice().getBaseInstantTime())
+ .isEqualTo(LEADING_ZERO_INSTANT_TIME);
+ }
+
+ @Test
+ void
testPlanPartitionedTableWithoutDiscoveredPartitionsReturnsEmptySplits() throws
Exception {
+ TablePath tablePath = TablePath.of(DATABASE, TABLE);
+ createPartitionedCowTable(tablePath);
+ createCompletedCommit(tablePath, INSTANT_TIME);
+
+ List<HudiSplit> splits =
+ new HudiSplitPlanner(hudiConfig, tablePath,
Long.parseLong(INSTANT_TIME)).plan();
+
+ assertThat(splits).isEmpty();
+ }
+
+ @Test
+ void testPlanBucketUnawareCopyOnWriteSplitsAsBucketMinusOne() throws
Exception {
+ TablePath tablePath = TablePath.of(DATABASE, TABLE);
+ createCowTable(tablePath, false);
+ String partitionPath = "";
+ String fileId = "plainfileid";
+ String fileName = createBaseFile(tablePath, partitionPath,
INSTANT_TIME, fileId);
+ createCompletedCommit(tablePath, partitionPath, INSTANT_TIME, fileId,
fileName);
+ assertHudiFixtureVisible(tablePath, partitionPath);
+
+ List<HudiSplit> splits =
+ new HudiSplitPlanner(hudiConfig, tablePath,
Long.parseLong(INSTANT_TIME)).plan();
+
+ assertThat(splits).hasSize(1);
+ assertThat(splits.get(0).bucket()).isEqualTo(-1);
+ }
+
+ @Test
+ void testPlanFailsForBucketAwareTableWithUnparsableFileId() throws
Exception {
+ TablePath tablePath = TablePath.of(DATABASE, TABLE);
+ createCowTable(tablePath);
+ String partitionPath = "";
+ String fileId = "plainfileid";
+ String fileName = createBaseFile(tablePath, partitionPath,
INSTANT_TIME, fileId);
+ createCompletedCommit(tablePath, partitionPath, INSTANT_TIME, fileId,
fileName);
+ assertHudiFixtureVisible(tablePath, partitionPath);
+
+ assertThatThrownBy(
+ () ->
+ new HudiSplitPlanner(
+ hudiConfig, tablePath,
Long.parseLong(INSTANT_TIME))
+ .plan())
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Failed to extract Hudi bucket id")
+ .hasMessageContaining(fileId);
+ }
+
+ private void createCowTable(TablePath tablePath) throws Exception {
+ createCowTable(tablePath, true);
+ }
+
+ private void createCowTable(TablePath tablePath, boolean bucketAware)
throws Exception {
+ Schema schema = Schema.newBuilder().column("id",
DataTypes.BIGINT()).build();
+ TableDescriptor.Builder tableDescriptorBuilder =
+ TableDescriptor.builder()
+ .schema(schema)
+
.property("hudi.hoodie.datasource.write.recordkey.field", "id");
+ if (bucketAware) {
+ tableDescriptorBuilder.distributedBy(4, "id");
+ } else {
+ tableDescriptorBuilder.distributedBy(4);
+ }
+ TableDescriptor tableDescriptor = tableDescriptorBuilder.build();
+ try (HudiLakeCatalog catalog = new HudiLakeCatalog(hudiConfig)) {
+ catalog.createTable(tablePath, tableDescriptor, new
TestingLakeCatalogContext());
+ }
+ }
+
+ private void createPartitionedCowTable(TablePath tablePath) throws
Exception {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("dt", DataTypes.STRING())
+ .build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .partitionedBy("dt")
+ .distributedBy(4, "id")
+
.property("hudi.hoodie.datasource.write.recordkey.field", "id")
+ .build();
+ try (HudiLakeCatalog catalog = new HudiLakeCatalog(hudiConfig)) {
+ catalog.createTable(tablePath, tableDescriptor, new
TestingLakeCatalogContext());
+ }
+ }
+
+ private String createBaseFile(
+ TablePath tablePath, String partitionPath, String instantTime,
String fileId)
+ throws IOException {
+ File partitionDir =
+ partitionPath.isEmpty()
+ ? basePath(tablePath)
+ : new File(basePath(tablePath), partitionPath);
+ Files.createDirectories(partitionDir.toPath());
+ String fileName =
+ FSUtils.makeBaseFileName(
+ instantTime,
+ FSUtils.makeWriteToken(0, 0, 1),
+ fileId,
+ HoodieFileFormat.PARQUET.getFileExtension());
+ Files.write(new File(partitionDir, fileName).toPath(), new byte[] {1});
+ return fileName;
+ }
+
+ private void createCompletedCommit(
+ TablePath tablePath,
+ String partitionPath,
+ String instantTime,
+ String fileId,
+ String fileName) {
+ HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata();
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setPartitionPath(partitionPath);
+ writeStat.setFileId(fileId);
+ writeStat.setPath(partitionPath.isEmpty() ? fileName : partitionPath +
"/" + fileName);
+ writeStat.setTotalWriteBytes(1);
+ writeStat.setFileSizeInBytes(1);
+ writeStat.setNumWrites(1);
+ commitMetadata.addWriteStat(partitionPath, writeStat);
+
+ createCompletedCommit(tablePath, instantTime, commitMetadata);
+ }
+
+ private void createCompletedCommit(TablePath tablePath, String
instantTime) {
+ createCompletedCommit(tablePath, instantTime, new
HoodieCommitMetadata());
+ }
+
+ private void createCompletedCommit(
+ TablePath tablePath, String instantTime, HoodieCommitMetadata
commitMetadata) {
+ HoodieTableMetaClient metaClient =
+ HoodieTableMetaClient.builder()
+ .setBasePath(basePath(tablePath).getAbsolutePath())
+ .setConf(
+ new HadoopStorageConfiguration(
+
HudiTableInfo.getHadoopConfiguration(hudiConfig)))
+ .build();
+ HoodieInstant inflightInstant =
+ metaClient.createNewInstant(
+ HoodieInstant.State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, instantTime);
+ metaClient.getActiveTimeline().createNewInstant(inflightInstant);
+
+ HoodieInstant instant =
+ metaClient
+ .getActiveTimeline()
+ .saveAsComplete(inflightInstant,
Option.of(commitMetadata));
+ assertThat(instant.isCompleted()).isTrue();
+ metaClient.reloadActiveTimeline();
+ }
+
+ private void assertHudiFixtureVisible(TablePath tablePath, String
partitionPath)
+ throws IOException {
+ try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath,
hudiConfig)) {
+ assertThat(
+ FSUtils.getAllPartitionPaths(
+ hudiTableInfo.getEngineContext(),
+ hudiTableInfo.getMetaClient(),
+ false))
+ .isEmpty();
+ List<HoodieBaseFile> baseFiles =
+ hudiTableInfo
+ .getFileSystemView()
+ .getLatestBaseFilesBeforeOrOn(partitionPath,
INSTANT_TIME)
+ .collect(Collectors.toList());
+ assertThat(baseFiles).hasSize(1);
+ }
+ }
+
+ private File basePath(TablePath tablePath) {
+ return new File(
+ new File(tempWarehouseDir, tablePath.getDatabaseName()),
tablePath.getTableName());
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializerTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializerTest.java
new file mode 100644
index 000000000..16a768cc8
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializerTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.source;
+
+import org.apache.fluss.utils.InstantiationUtils;
+
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.junit.jupiter.api.Test;
+
+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 HudiSplitSerializer}. */
+class HudiSplitSerializerTest {
+
+ @Test
+ void testSerializeAndDeserialize() throws Exception {
+ HudiSplitSerializer serializer = new HudiSplitSerializer();
+ HudiSplit split =
+ new HudiSplit(
+ new FileSlice(
+ new HoodieFileGroupId(
+ "dt=20260608",
"00000000-0000-0000-0000-000000000001"),
+ "20260608010101000"),
+ 1,
+ Collections.singletonList("20260608"));
+
+ HudiSplit deserialized =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(split));
+
+ assertThat(deserialized).isEqualTo(split);
+ assertThat(deserialized.getFileSlice().getFileGroupId())
+ .isEqualTo(split.getFileSlice().getFileGroupId());
+ }
+
+ @Test
+ void testDeserializeRejectsUnknownVersion() {
+ HudiSplitSerializer serializer = new HudiSplitSerializer();
+
+ assertThatThrownBy(() -> serializer.deserialize(2, new byte[0]))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unsupported HudiSplit serialization
version");
+ }
+
+ @Test
+ void testDeserializeRejectsUnexpectedObjectType() throws Exception {
+ HudiSplitSerializer serializer = new HudiSplitSerializer();
+ byte[] serialized = InstantiationUtils.serializeObject("not a hudi
split");
+
+ assertThatThrownBy(() ->
serializer.deserialize(serializer.getVersion(), serialized))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Expected HudiSplit");
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitTest.java
new file mode 100644
index 000000000..ae611754e
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiSplitTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.source;
+
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link HudiSplit}. */
+class HudiSplitTest {
+
+ private static final String PARTITION_PATH = "dt=20260608";
+ private static final String FILE_ID =
"00000000-0000-0000-0000-000000000001";
+ private static final String INSTANT_TIME = "20260608010101000";
+
+ @Test
+ void testEqualsUsesStableFileSliceFields() {
+ HudiSplit first = new HudiSplit(fileSlice("/tmp/base-1.parquet", 1),
1, partition());
+ HudiSplit same = new HudiSplit(fileSlice("/tmp/base-1.parquet", 1), 1,
partition());
+ HudiSplit differentBaseFile =
+ new HudiSplit(fileSlice("/tmp/base-2.parquet", 1), 1,
partition());
+ HudiSplit differentLogFile =
+ new HudiSplit(fileSlice("/tmp/base-1.parquet", 2), 1,
partition());
+
+ assertThat(first).isEqualTo(same);
+ assertThat(first).hasSameHashCodeAs(same);
+ assertThat(first).isNotEqualTo(differentBaseFile);
+ assertThat(first).isNotEqualTo(differentLogFile);
+ }
+
+ private static FileSlice fileSlice(String baseFilePath, int logVersion) {
+ HoodieFileGroupId fileGroupId = new HoodieFileGroupId(PARTITION_PATH,
FILE_ID);
+ FileSlice fileSlice =
+ new FileSlice(
+ fileGroupId,
+ INSTANT_TIME,
+ new HoodieBaseFile(baseFilePath),
+ Collections.<HoodieLogFile>emptyList());
+ fileSlice.addLogFile(
+ new HoodieLogFile(
+ "/tmp/"
+ + FSUtils.makeLogFileName(
+ FILE_ID,
+ HoodieLogFile.DELTA_EXTENSION,
+ INSTANT_TIME,
+ logVersion,
+ FSUtils.makeWriteToken(0, 0, 1))));
+ return fileSlice;
+ }
+
+ private static List<String> partition() {
+ return Collections.singletonList("20260608");
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiConversionsTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiConversionsTest.java
new file mode 100644
index 000000000..07421c279
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiConversionsTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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;
+
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION;
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION;
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_PARTITION_KEYS_OPTION;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link HudiConversions}. */
+class HudiConversionsTest {
+
+ @Test
+ void testBuildHudiTablePropertiesCarriesFlussBucketMetadataForPkTable() {
+ Schema schema =
+ Schema.newBuilder().column("id",
DataTypes.BIGINT()).primaryKey("id").build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(schema).distributedBy(3,
"id").build();
+
+ Map<String, String> properties =
+ HudiConversions.buildHudiTableProperties(
+ TablePath.of("db1", "table1"), tableDescriptor, true);
+
+ assertThat(properties).containsEntry(FLUSS_BUCKET_AWARE_OPTION,
"true");
+ assertThat(properties).containsEntry(FLUSS_BUCKET_KEYS_OPTION, "id");
+ assertThat(properties).containsEntry(FLUSS_PARTITION_KEYS_OPTION, "");
+ }
+
+ @Test
+ void
testBuildHudiTablePropertiesCarriesFlussBucketMetadataForBucketUnawareLogTable()
{
+ Schema schema = Schema.newBuilder().column("id",
DataTypes.BIGINT()).build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(3)
+
.property("hudi.hoodie.datasource.write.recordkey.field", "id")
+ .build();
+
+ Map<String, String> properties =
+ HudiConversions.buildHudiTableProperties(
+ TablePath.of("db1", "table1"), tableDescriptor, false);
+
+ assertThat(properties).containsEntry(FLUSS_BUCKET_AWARE_OPTION,
"false");
+ assertThat(properties).containsEntry(FLUSS_BUCKET_KEYS_OPTION, "");
+ assertThat(properties).containsEntry(FLUSS_PARTITION_KEYS_OPTION, "");
+ }
+
+ @Test
+ void testBuildHudiTablePropertiesCarriesFlussPartitionMetadata() {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("dt", DataTypes.STRING())
+ .column("hr", DataTypes.STRING())
+ .build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(3, "id")
+ .partitionedBy("dt", "hr")
+
.property("hudi.hoodie.datasource.write.recordkey.field", "id")
+ .build();
+
+ Map<String, String> properties =
+ HudiConversions.buildHudiTableProperties(
+ TablePath.of("db1", "table1"), tableDescriptor, false);
+
+ assertThat(properties).containsEntry(FLUSS_PARTITION_KEYS_OPTION,
"dt,hr");
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiTableInfoTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiTableInfoTest.java
new file mode 100644
index 000000000..bab3a6c57
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/HudiTableInfoTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.model.HoodieTableType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION;
+import static
org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HudiTableInfo}. */
+class HudiTableInfoTest {
+
+ @Test
+ void testExtractPartitionValues() {
+ assertThat(
+ HudiTableInfo.extractPartitionValues(
+ "dt=20260608/hr=10", Arrays.asList("dt",
"hr")))
+ .containsExactly("20260608", "10");
+ assertThat(
+ HudiTableInfo.extractPartitionValues(
+ "hr=10/dt=20260608", Arrays.asList("dt",
"hr")))
+ .containsExactly("20260608", "10");
+ assertThat(HudiTableInfo.extractPartitionValues("20260608/10",
Arrays.asList("dt", "hr")))
+ .containsExactly("20260608", "10");
+ assertThat(HudiTableInfo.extractPartitionValues("",
Collections.singletonList("dt")))
+ .isEmpty();
+ }
+
+ @Test
+ void testExtractPartitionValuesRejectsMismatchedPartitionPath() {
+ assertThatThrownBy(
+ () ->
+ HudiTableInfo.extractPartitionValues(
+ "dt=20260608", Arrays.asList("dt",
"hr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("does not match partition fields");
+ assertThatThrownBy(
+ () ->
+ HudiTableInfo.extractPartitionValues(
+ "dt=20260608/unknown=10",
Arrays.asList("dt", "hr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("unknown partition field");
+ assertThatThrownBy(
+ () ->
+ HudiTableInfo.extractPartitionValues(
+ "dt=20260608/dt=20260609",
Arrays.asList("dt", "hr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("duplicate partition field");
+ assertThatThrownBy(
+ () ->
+ HudiTableInfo.extractPartitionValues(
+ "dt=20260608/10", Arrays.asList("dt",
"hr")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("mixes hive-style and raw partition
segments");
+ }
+
+ @Test
+ void testResolveBucketAwareUsesExplicitBucketAwareOption() {
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put(FLUSS_BUCKET_AWARE_OPTION, "false");
+ tableOptions.put(FLUSS_BUCKET_KEYS_OPTION, "id");
+
+ assertThat(HudiTableInfo.resolveBucketAware(tableOptions,
HoodieTableType.MERGE_ON_READ))
+ .isFalse();
+ }
+
+ @Test
+ void testResolveBucketAwareUsesBucketKeysOption() {
+ Map<String, String> bucketAwareOptions = new HashMap<>();
+ bucketAwareOptions.put(FLUSS_BUCKET_KEYS_OPTION, "id");
+ assertThat(
+ HudiTableInfo.resolveBucketAware(
+ bucketAwareOptions,
HoodieTableType.COPY_ON_WRITE))
+ .isTrue();
+
+ Map<String, String> bucketUnawareOptions = new HashMap<>();
+ bucketUnawareOptions.put(FLUSS_BUCKET_KEYS_OPTION, "");
+ assertThat(
+ HudiTableInfo.resolveBucketAware(
+ bucketUnawareOptions,
HoodieTableType.MERGE_ON_READ))
+ .isFalse();
+ }
+
+ @Test
+ void testResolveBucketAwareDefaultsByTableTypeWhenMetadataIsMissing() {
+ assertThat(
+ HudiTableInfo.resolveBucketAware(
+ Collections.emptyMap(),
HoodieTableType.MERGE_ON_READ))
+ .isTrue();
+ assertThat(
+ HudiTableInfo.resolveBucketAware(
+ Collections.emptyMap(),
HoodieTableType.COPY_ON_WRITE))
+ .isFalse();
+ }
+}
