[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;
@@ -36,9 +38,8 @@ public HudiLakeStorage(Configuration configuration) {
     @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 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,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());
+    }
+}