[lake/hudi] Introduce Hudi source split planner

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;
@@ -47,12 +49,7 @@ public LakeCatalog createLakeCatalog() {
     }
 
     @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);
     }
 }

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();
+    }
+}

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplit.java

@@ -0,0 +1,91 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** 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
+                && Objects.equals(fileSlice, hudiSplit.fileSlice)
+                && Objects.equals(partition, hudiSplit.partition);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(fileSlice, bucket, partition);
+    }
+
+    @Override
+    public String toString() {
+        return "HudiSplit{"
+                + "fileSlice="
+                + fileSlice
+                + ", bucket="
+                + bucket
+                + ", partition="
+                + partition
+                + '}';
+    }
+}

fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java

@@ -0,0 +1,157 @@
+/*
+ * 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.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.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 = String.valueOf(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()) {
+                partitionPaths = Collections.singletonList("");
+            }
+
+            List<HudiSplit> splits = new ArrayList<>();
+            for (String partitionPath : partitionPaths) {
+                splits.addAll(planPartition(hudiTableInfo, snapshotTime, partitionPath));
+            }
+            LOG.debug(
+                    "Planned {} Hudi splits for table {} at instant {}.",
+                    splits.size(),
+                    tablePath,
+                    snapshotTime);
+            return splits;
+        } catch (IOException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new IOException("Failed to plan Hudi splits for table " + tablePath + ".", e);
+        }
+    }
+
+    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;
+        }
+
+        List<HoodieBaseFile> baseFiles =
+                fileSystemView
+                        .getLatestBaseFilesBeforeOrOn(partitionPath, snapshotTime)
+                        .collect(Collectors.toList());
+        for (HoodieBaseFile baseFile : baseFiles) {
+            splits.add(
+                    toHudiSplit(
+                            hudiTableInfo, partitionPath, toFileSlice(partitionPath, baseFile)));
+        }
+        return splits;
+    }
+
+    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);
+        }
+    }
+}