(hudi) branch master updated: feat: add hudi flink source split POJOs (#17483)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 0c33f72d1b98 feat: add hudi flink source split POJOs (#17483)
0c33f72d1b98 is described below

commit 0c33f72d1b98522a03d6caccf5527bf13ee6dcd6
Author: Peter Huang <hu...@gmail.com>
AuthorDate: Sat Dec 6 00:46:02 2025 -0800

    feat: add hudi flink source split POJOs (#17483)
---
 .../hudi/source/split/HoodieSourceSplit.java       | 140 +++++++++++++++++++++
 .../hudi/source/split/HoodieSourceSplitState.java  |  40 ++++++
 .../hudi/source/split/HoodieSourceSplitStatus.java |  31 +++++
 .../hudi/source/split/SerializableComparator.java  |  27 ++++
 .../hudi/source/split/SplitRequestEvent.java       |  57 +++++++++
 5 files changed, 295 insertions(+)

diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplit.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplit.java
new file mode 100644
index 000000000000..ee05fe0e8ba7
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplit.java
@@ -0,0 +1,140 @@
+/*
+ * 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.hudi.source.split;
+
+import org.apache.hudi.common.util.Option;
+
+import org.apache.flink.api.connector.source.SourceSplit;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Hoodie SourceSplit implementation for source V2.
+ */
+public class HoodieSourceSplit implements SourceSplit, Serializable {
+  private static final long serialVersionUID = 1L;
+
+  private static final long NUM_NO_CONSUMPTION = 0L;
+
+  // the split number
+  private final int splitNum;
+  // the base file of a file slice
+  private final Option<String> basePath;
+  // change log files of a file slice
+  private final Option<List<String>> logPaths;
+  // the base table path
+  private final String tablePath;
+  // source merge type
+  private final String mergeType;
+  // file id of file splice
+  protected String fileId;
+
+  // for streaming reader to record the consumed offset,
+  // which is the start of next round reading.
+  private long consumed = NUM_NO_CONSUMPTION;
+
+  // for failure recovering
+  private int fileOffset;
+  private long recordOffset;
+
+  public HoodieSourceSplit(
+      int splitNum,
+      @Nullable String basePath,
+      Option<List<String>> logPaths,
+      String tablePath,
+      String mergeType,
+      String fileId) {
+    this.splitNum = splitNum;
+    this.basePath = Option.ofNullable(basePath);
+    this.logPaths = logPaths;
+    this.tablePath = tablePath;
+    this.mergeType = mergeType;
+    this.fileId = fileId;
+    this.fileOffset = 0;
+    this.recordOffset = 0L;
+  }
+
+  @Override
+  public String splitId() {
+    return toString();
+  }
+
+  public String getFileId() {
+    return fileId;
+  }
+
+  public void setFileId(String fileId) {
+    this.fileId = fileId;
+  }
+
+  public Option<String> getBasePath() {
+    return basePath;
+  }
+
+  public Option<List<String>> getLogPaths() {
+    return logPaths;
+  }
+
+  public String getTablePath() {
+    return tablePath;
+  }
+
+  public String getMergeType() {
+    return mergeType;
+  }
+
+  public void consume() {
+    this.consumed += 1L;
+  }
+
+  public long getConsumed() {
+    return consumed;
+  }
+
+  public boolean isConsumed() {
+    return this.consumed != NUM_NO_CONSUMPTION;
+  }
+
+  public int getFileOffset() {
+    return fileOffset;
+  }
+
+  public long getRecordOffset() {
+    return recordOffset;
+  }
+
+  public void updatePosition(int newFileOffset, long newRecordOffset) {
+    fileOffset = newFileOffset;
+    recordOffset = newRecordOffset;
+  }
+
+  @Override
+  public String toString() {
+    return "HoodieSourceSplit{"
+        + "splitNum=" + splitNum
+        + ", basePath=" + basePath
+        + ", logPaths=" + logPaths
+        + ", tablePath='" + tablePath + '\''
+        + ", mergeType='" + mergeType + '\''
+        + '}';
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitState.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitState.java
new file mode 100644
index 000000000000..57dba50c3d35
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitState.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.hudi.source.split;
+
+/**
+ * Hoodie source split state that will be snapshot to checkpoint within enumerator.
+ */
+public class HoodieSourceSplitState {
+  private final HoodieSourceSplit split;
+  private final HoodieSourceSplitStatus status;
+
+  public HoodieSourceSplitState(HoodieSourceSplit split, HoodieSourceSplitStatus status) {
+    this.split = split;
+    this.status = status;
+  }
+
+  public HoodieSourceSplit split() {
+    return split;
+  }
+
+  public HoodieSourceSplitStatus status() {
+    return status;
+  }
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitStatus.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitStatus.java
new file mode 100644
index 000000000000..86c7a2a6701c
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/HoodieSourceSplitStatus.java
@@ -0,0 +1,31 @@
+/*
+ * 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.hudi.source.split;
+
+/**
+ * Hoodie source split status.
+ */
+public enum HoodieSourceSplitStatus {
+  // Created but not assigned
+  UNASSIGNED,
+  // Assigned to reader
+  ASSIGNED,
+  // Completed processing
+  COMPLETED
+}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/SerializableComparator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/SerializableComparator.java
new file mode 100644
index 000000000000..42a664113003
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/SerializableComparator.java
@@ -0,0 +1,27 @@
+/*
+ * 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.hudi.source.split;
+
+import java.io.Serializable;
+import java.util.Comparator;
+
+/**
+ * Comparator interface for HoodieSourceSplit.
+ */
+public interface SerializableComparator<T> extends Comparator<T>, Serializable {}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/SplitRequestEvent.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/SplitRequestEvent.java
new file mode 100644
index 000000000000..c8729f09a150
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/SplitRequestEvent.java
@@ -0,0 +1,57 @@
+/*
+ * 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.hudi.source.split;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.SourceEvent;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * SourceEvent for hoodie split request passed between the SourceReaders and Enumerators.
+ */
+@Internal
+public class SplitRequestEvent implements SourceEvent {
+  private static final long serialVersionUID = 1L;
+
+  private final Collection<String> finishedSplitIds;
+  private final String requesterHostname;
+
+  public SplitRequestEvent() {
+    this(Collections.emptyList());
+  }
+
+  public SplitRequestEvent(Collection<String> finishedSplitIds) {
+    this(finishedSplitIds, null);
+  }
+
+  public SplitRequestEvent(Collection<String> finishedSplitIds, String requesterHostname) {
+    this.finishedSplitIds = finishedSplitIds;
+    this.requesterHostname = requesterHostname;
+  }
+
+  public Collection<String> finishedSplitIds() {
+    return finishedSplitIds;
+  }
+
+  public String requesterHostname() {
+    return requesterHostname;
+  }
+}
\ No newline at end of file