Copilot commented on code in PR #3627:
URL: https://github.com/apache/fluss/pull/3627#discussion_r3558707270
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java:
##########
@@ -78,23 +80,129 @@ public SortMergeReader(
this.snapshotRecordIterator =
ConcatRecordIterator.wrap(snapshotRecordIterators,
userKeyComparator, pkIndexes);
this.changeLogIterator = changeLogIterator;
- this.changeLogIteratorWrapper = new ChangeLogIteratorWrapper();
- this.snapshotMergedRowIteratorWrapper = new
SnapshotMergedRowIteratorWrapper();
- // to project to fields provided by user
this.projectedRow = projectedFields == null ? null :
ProjectedRow.from(projectedFields);
}
+ /**
+ * Returns the merged-row iterator, or {@code null} when nothing is left.
The same instance is
+ * returned on every call, so repeated invocations keep draining the
snapshot and change log.
+ */
@Nullable
public CloseableIterator<InternalRow> readBatch() {
- if (!snapshotRecordIterator.hasNext()) {
- return changeLogIterator.hasNext()
- ? changeLogIteratorWrapper.replace(changeLogIterator)
- : null;
- } else {
- CloseableIterator<SortMergeRows> mergedRecordIterator =
- transform(snapshotRecordIterator,
this::sortMergeWithChangeLog);
-
- return
snapshotMergedRowIteratorWrapper.replace(mergedRecordIterator);
+ if (mergeIterator == null) {
+ mergeIterator = new MergeIterator();
+ }
+ return mergeIterator.hasNext() ? mergeIterator : null;
+ }
+
+ /** Two-way merge iterator over the snapshot and change log streams. */
+ private class MergeIterator implements CloseableIterator<InternalRow> {
+
+ // peeked head of each stream; null means not peeked yet or drained
+ private @Nullable LogRecord pendingSnapshot;
+ private @Nullable KeyValueRow pendingLog;
+
+ // next row to return (before projection) and whether it has been
computed
+ private @Nullable InternalRow nextRow;
+ private boolean nextComputed;
+
+ private @Nullable LogRecord peekSnapshot() {
+ if (pendingSnapshot == null && snapshotRecordIterator.hasNext()) {
+ pendingSnapshot = snapshotRecordIterator.next();
+ }
+ return pendingSnapshot;
+ }
+
+ private @Nullable KeyValueRow peekLog() {
+ if (pendingLog == null && changeLogIterator.hasNext()) {
+ pendingLog = changeLogIterator.next();
+ }
+ return pendingLog;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!nextComputed) {
+ nextRow = advance();
+ nextComputed = true;
+ }
+ return nextRow != null;
+ }
+
+ @Override
+ public InternalRow next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ InternalRow row = nextRow;
+ nextRow = null;
+ nextComputed = false;
+ return projectedRow == null ? row : projectedRow.replaceRow(row);
+ }
+
+ /** Advances to the next merged row, or {@code null} when both streams
are drained. */
+ private @Nullable InternalRow advance() {
+ while (true) {
+ LogRecord snapshot = peekSnapshot();
+ KeyValueRow log = peekLog();
+
+ if (snapshot == null && log == null) {
+ return null;
+ }
+
+ // only the change log remains
+ if (snapshot == null) {
+ InternalRow row = takeLog(log);
+ if (row != null) {
+ return row;
+ }
+ continue;
+ }
+
+ InternalRow snapshotRow = snapshot.getRow();
+
+ // only the snapshot remains
+ if (log == null) {
+ pendingSnapshot = null;
+ return snapshotRow;
+ }
+
+ int compareResult =
+ userKeyComparator.compare(
+
snapshotProjectedPkRow.replaceRow(snapshotRow), log.keyRow());
+ if (compareResult < 0) {
+ pendingSnapshot = null;
+ return snapshotRow;
+ } else if (compareResult > 0) {
+ InternalRow row = takeLog(log);
+ if (row != null) {
+ return row;
+ }
+ continue;
+ } else {
+ // same key: change log overrides snapshot, delete drops
both
+ pendingSnapshot = null;
+ InternalRow row = takeLog(log);
+ if (row != null) {
+ return row;
+ }
+ continue;
+ }
+ }
+ }
+
+ /**
+ * Consumes the change log head; returns its value row, or {@code
null} if it is a delete.
+ */
+ private @Nullable InternalRow takeLog(KeyValueRow log) {
+ pendingLog = null;
+ return log.isDelete() ? null : log.valueRow();
+ }
+
+ @Override
+ public void close() {
+ snapshotRecordIterator.close();
+ changeLogIterator.close();
}
Review Comment:
MergeIterator.close() closes snapshotRecordIterator/changeLogIterator, but
SortMergeReader.readBatch() returns the same MergeIterator instance across
calls (and advertises that in the Javadoc). If a caller follows the typical
CloseableIterator pattern of calling readBatch() in a loop and closing each
returned iterator (e.g., try-with-resources per batch), the first close will
permanently close the underlying streams and makes any subsequent
readBatch()/hasNext() behavior after close undefined for some iterator
implementations. Consider decoupling per-batch iterator closing from closing
the underlying inputs (e.g., return a lightweight wrapper whose close() is a
no-op/idempotent, and add an explicit SortMergeReader.close()/finalizer in the
owning scanner), or otherwise document/enforce that the returned iterator must
only be closed once at the end of the full drain.
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/SortMergeReader.java:
##########
@@ -78,23 +80,129 @@ public SortMergeReader(
this.snapshotRecordIterator =
ConcatRecordIterator.wrap(snapshotRecordIterators,
userKeyComparator, pkIndexes);
this.changeLogIterator = changeLogIterator;
- this.changeLogIteratorWrapper = new ChangeLogIteratorWrapper();
- this.snapshotMergedRowIteratorWrapper = new
SnapshotMergedRowIteratorWrapper();
- // to project to fields provided by user
this.projectedRow = projectedFields == null ? null :
ProjectedRow.from(projectedFields);
}
+ /**
+ * Returns the merged-row iterator, or {@code null} when nothing is left.
The same instance is
+ * returned on every call, so repeated invocations keep draining the
snapshot and change log.
+ */
Review Comment:
The readBatch() Javadoc says the same iterator instance is returned on every
call, but it doesn’t clarify the interaction with CloseableIterator.close().
Since closing the iterator closes the underlying snapshot/change-log iterators,
it would help to explicitly document that callers must treat the returned
iterator as a shared, stateful stream (i.e., avoid closing it until they are
done consuming all rows) and that readBatch() should not be called after the
iterator is closed.
--
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]
