wombatu-kun commented on code in PR #19202:
URL: https://github.com/apache/hudi/pull/19202#discussion_r3592372586
##########
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
+ public void open(HoodieSourceSplit split) {
+ this.currentIterator = createRecordIterator(split);
+ try {
+ // Skip the records already emitted before the last checkpoint so a
recovered split resumes at
+ // the right position; matches the validation the old BatchRecords#seek
performed.
+ long consumed = split.getConsumed();
+ for (long i = 0; i < consumed; i++) {
+ if (currentIterator.hasNext()) {
+ currentIterator.next();
+ } else {
+ throw new IllegalStateException(
+ String.format("Invalid starting record offset %d for split %s",
consumed, split.splitId()));
+ }
+ }
+ this.nextRecordOffset = consumed;
+ } catch (RuntimeException | Error e) {
+ // Close on failure so the split's I/O resources are released even if
the resume-skip fails.
+ closeCurrentSplit();
+ throw e;
+ }
+ }
+
+ @Override
+ public BatchRecords<RowData> readBatch(HoodieSourceSplit split, int
batchSize) {
+ ValidationUtils.checkState(currentIterator != null,
+ "readBatch called before open for split " + split.splitId());
+ RowDataSerializer serializer = getCopySerializer();
+ List<RowData> buffer = new ArrayList<>();
+ try {
+ while (buffer.size() < batchSize && currentIterator.hasNext()) {
+ RowData next = currentIterator.next();
+ RowData copy = serializer.copy(next);
+ copy.setRowKind(next.getRowKind());
Review Comment:
Done ed4adeeaf895. Removed the redundant re-apply and fixed the javadoc that
wrongly cited the serialize/deserialize round-trip; RowDataSerializer.copy
preserves RowKind on all RowData subtypes, guarded by
TestAbstractSplitReaderFunction.
--
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]
