feat: the basic new hudi source reader

Author: HuangZhenQiu

hudi-flink-datasource/hudi-flink/pom.xml

@@ -246,6 +246,12 @@
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-metrics-dropwizard</artifactId>

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java

@@ -409,7 +409,7 @@ private List<MergeOnReadInputSplit> getInputSplits(
       String latestCommit = InstantComparison.minInstant(fileSlice.getLatestInstantTime(), endInstant);
       return new MergeOnReadInputSplit(cnt.getAndAdd(1),
           basePath, logPaths, latestCommit,
-          metaClient.getBasePath().toString(), maxCompactionMemoryInBytes, mergeType, instantRange, fileSlice.getFileId());
+          metaClient.getBasePath().toString(), fileSlice.getPartitionPath(), maxCompactionMemoryInBytes, mergeType, instantRange, fileSlice.getFileId());
     }).sorted(Comparator.comparing(MergeOnReadInputSplit::getLatestCommit)).collect(Collectors.toList());
   }
 

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java

@@ -0,0 +1,109 @@
+/*
+ * 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.reader;
+
+import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
+
+import java.util.Collections;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+
+/**
+ * Implementation of RecordsWithSplitIds with a list record inside.
+ *
+ * Type parameters: <T> – record type
+ */
+public class BatchRecords<T> implements RecordsWithSplitIds<HoodieRecordWithPosition<T>> {
+  private String splitId;
+  private final ClosableIterator<T> recordIterator;
+  private final Set<String> finishedSplits;
+  private final HoodieRecordWithPosition<T> recordAndPosition;
+
+  // point to current read position within the records list
+  private int position;
+
+  BatchRecords(
+      String splitId,
+      ClosableIterator<T> recordIterator,
+      int fileOffset,
+      long startingRecordOffset,
+      Set<String> finishedSplits) {
+    ValidationUtils.checkArgument(
+        finishedSplits != null, "finishedSplits can be empty but not null");
+    ValidationUtils.checkArgument(
+        recordIterator != null, "recordIterator can be empty but not null");
+
+    this.splitId = splitId;
+    this.recordIterator = recordIterator;
+    this.finishedSplits = finishedSplits;
+    this.recordAndPosition = new HoodieRecordWithPosition<>();
+    this.recordAndPosition.set(null, fileOffset, startingRecordOffset);
+    this.position = 0;
+  }
+
+  @Nullable
+  @Override
+  public String nextSplit() {
+    String nextSplit = this.splitId;
+    // set the splitId to null to indicate no more splits
+    // this class only contains record for one split
+    this.splitId = null;
+    return nextSplit;
+  }
+
+  @Nullable
+  @Override
+  public HoodieRecordWithPosition<T> nextRecordFromSplit() {
+    if (recordIterator.hasNext()) {
+      recordAndPosition.record(recordIterator.next());
+      position++;
+      return recordAndPosition;
+    } else {
+      return null;
+    }
+  }
+
+  @Override
+  public Set<String> finishedSplits() {
+    return finishedSplits;
+  }
+
+  public void seek(long startingRecordOffset) {
+    for (long i = 0; i < startingRecordOffset; ++i) {
+      if (recordIterator.hasNext()) {
+        recordIterator.next();
+      } else {
+        throw new IllegalStateException(
+            String.format(
+                "Invalid starting record offset %d for split %s",
+                startingRecordOffset,
+                splitId));
+      }
+    }
+  }
+
+  public static <T> BatchRecords<T> forRecords(
+      String splitId, ClosableIterator<T> recordIterator, int fileOffset, long startingRecordOffset) {
+
+    return new BatchRecords<>(
+        splitId, recordIterator, fileOffset, startingRecordOffset, Collections.emptySet());
+  }
+}

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordEmitter.java

@@ -0,0 +1,36 @@
+/*
+ * 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.reader;
+
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.connector.base.source.reader.RecordEmitter;
+import org.apache.hudi.source.split.HoodieSourceSplit;
+
+/**
+ * Default Hoodie record emitter.
+ * @param <T>
+ */
+public class HoodieRecordEmitter<T> implements RecordEmitter<HoodieRecordWithPosition<T>, T, HoodieSourceSplit> {
+
+  @Override
+  public void emitRecord(HoodieRecordWithPosition<T> record, SourceOutput<T> output, HoodieSourceSplit split) throws Exception {
+    output.collect(record.record());
+    split.updatePosition(record.fileOffset(), record.recordOffset());
+  }
+}

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieRecordWithPosition.java

@@ -0,0 +1,72 @@
+/*
+ * 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.reader;
+
+import java.util.Locale;
+
+/**
+ * The Hoodie record with position information.
+ */
+public class HoodieRecordWithPosition<T> {
+  private T record;
+  private int fileOffset;
+  private long recordOffset;
+
+  public HoodieRecordWithPosition(T record, int fileOffset, long recordOffset) {
+    this.record = record;
+    this.fileOffset = fileOffset;
+    this.recordOffset = recordOffset;
+  }
+
+  public HoodieRecordWithPosition() {
+
+  }
+
+  // ------------------------------------------------------------------------
+
+  public T record() {
+    return record;
+  }
+
+  public int fileOffset() {
+    return fileOffset;
+  }
+
+  public long recordOffset() {
+    return recordOffset;
+  }
+
+  /** Updates the record and position in this object. */
+  public void set(T newRecord, int newFileOffset, long newRecordOffset) {
+    this.record = newRecord;
+    this.fileOffset = newFileOffset;
+    this.recordOffset = newRecordOffset;
+  }
+
+  /** Sets the next record of a sequence. This increments the {@code recordOffset} by one. */
+  public void record(T nextRecord) {
+    this.record = nextRecord;
+    this.recordOffset++;
+  }
+
+  @Override
+  public String toString() {
+    return String.format(Locale.ROOT, "%s @ %d + %d", record, fileOffset, recordOffset);
+  }
+}

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceReader.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.hudi.source.reader;
+
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordEmitter;
+import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
+import org.apache.hudi.source.split.HoodieSourceSplit;
+import org.apache.hudi.source.split.SerializableComparator;
+import org.apache.hudi.source.split.SplitRequestEvent;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * The reader implementation of Hoodie Source.
+ * @param <T> record type
+ */
+public class HoodieSourceReader<T> extends
+        SingleThreadMultiplexSourceReaderBase<HoodieRecordWithPosition<T>, T, HoodieSourceSplit, HoodieSourceSplit> {
+
+  public HoodieSourceReader(
+          RecordEmitter<HoodieRecordWithPosition<T>, T, HoodieSourceSplit> recordEmitter,
+          Configuration config,
+          SourceReaderContext context,
+          SplitReaderFunction<T> readerFunction,
+          SerializableComparator<HoodieSourceSplit> splitComparator) {
+    super(() -> new HoodieSourceSplitReader<>(context, readerFunction, splitComparator), recordEmitter, config, context);
+  }
+
+  @Override
+  public void start() {
+    // We request a split only if we did not get splits during the checkpoint restore.
+    // Otherwise, reader restarts will keep requesting more and more splits.
+    if (getNumberOfCurrentlyAssignedSplits() == 0) {
+      requestSplit(new ArrayList<>());
+    }
+  }
+
+  @Override
+  protected void onSplitFinished(Map<String, HoodieSourceSplit> finishedSplitIds) {
+    requestSplit(new ArrayList<>(finishedSplitIds.keySet()));
+  }
+
+  @Override
+  protected HoodieSourceSplit initializedState(HoodieSourceSplit hoodieSourceSplit) {
+    return hoodieSourceSplit;
+  }
+
+  @Override
+  protected HoodieSourceSplit toSplitType(String splitId, HoodieSourceSplit hoodieSourceSplit) {
+    return hoodieSourceSplit;
+  }
+
+  private void requestSplit(Collection<String> finishedSplitIds) {
+    context.sendSourceEventToCoordinator(new SplitRequestEvent(finishedSplitIds));
+  }
+}