hudi-agent commented on code in PR #19202:
URL: https://github.com/apache/hudi/pull/19202#discussion_r3586443624
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/AbstractSplitReaderFunction.java:
##########
@@ -52,6 +77,85 @@ public AbstractSplitReaderFunction(
this.emitDelete = emitDelete;
}
+ /**
+ * Creates the record iterator (and its underlying I/O resources) for {@code
split}. Closing the
+ * returned iterator must release all of those resources.
+ */
+ protected abstract ClosableIterator<RowData>
createRecordIterator(HoodieSourceSplit split);
+
+ /** The {@link RowType} of the records produced by {@link
#createRecordIterator}. */
+ protected abstract RowType producedRowType();
+
+ @Override
Review Comment:
🤖 nit: `copySerializer` is grouped under the "Per-split cursor state"
comment, but it's actually per-function-instance — it's lazily created once in
`getCopySerializer()` and never reset between splits. It might be worth moving
it up with `writeConfig`/`hadoopConf`, or splitting the block comment so a
reader doesn't assume it's torn down with each split.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/reader/TestHoodieSourceSplitReader.java:
##########
@@ -512,25 +593,54 @@ public TestSplitReaderFunction(List<String> testData) {
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<String>>
read(HoodieSourceSplit split) {
- readCount++;
+ public void open(HoodieSourceSplit split) {
+ openCount++;
lastReadSplit = split;
- ClosableIterator<String> iterator = createClosableIterator(testData);
- return BatchRecords.forRecords(
- split.splitId(),
- iterator,
- split.getFileOffset(),
- split.getConsumed()
- );
+ cursor = testData.iterator();
+ long consumed = split.getConsumed();
+ for (long i = 0; i < consumed; i++) {
+ if (cursor.hasNext()) {
+ cursor.next();
+ } else {
+ throw new IllegalStateException(
+ "Invalid starting record offset " + consumed + " for split " +
split.splitId());
+ }
+ }
+ nextRecordOffset = consumed;
+ }
+
+ @Override
+ public BatchRecords<String> readBatch(HoodieSourceSplit split, int
batchSize) {
+ List<String> buffer = new ArrayList<>();
+ while (buffer.size() < batchSize && cursor.hasNext()) {
+ buffer.add(cursor.next());
+ }
+ if (buffer.isEmpty()) {
+ return null;
+ }
+ long startingRecordOffset = nextRecordOffset;
+ nextRecordOffset += buffer.size();
+ return BatchRecords.forRecords(split.splitId(), buffer,
split.getFileOffset(), startingRecordOffset);
}
@Override
- public void close() throws Exception {
+ public void closeCurrentSplit() {
+ closeCurrentSplitCount++;
+ cursor = null;
+ }
+
+ @Override
+ public void close() {
closed = true;
}
+ // Number of splits opened; mirrors the old per-split read() count.
public int getReadCount() {
- return readCount;
+ return openCount;
+ }
Review Comment:
🤖 nit: could you rename `getReadCount()` to `getOpenCount()` to match the
`openCount` field? The comment "mirrors the old per-split read() count"
explains why the mismatch exists but doesn't fix it — a future reader has to
stop and decode it. The two call-sites in the assertions above would be a
one-word update.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
