Skip to content

[flink] Spill cdc log to rocksDB for pk batch read in case of OOM. #3930

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
loserwang1024 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[flink] Spill cdc log to rocksDB for pk batch read in case of OOM. #3930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f0c9810
[flink] Spill cdc log to rocksDB for pk batch read in case of OOM.
Aug 10, 2026
8040639
modified based on CR
Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 67 additions & 63 deletions ...c/main/java/org/apache/fluss/client/table/scanner/batch/KvSnapshotAndLogBatchScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,12 @@
import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.client.table.scanner.SortMergeReader;
import org.apache.fluss.client.table.scanner.log.LogScanner;
import org.apache.fluss.client.table.scanner.log.ScanRecords;
import org.apache.fluss.memory.MemorySegment;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.record.LogRecord;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.KeyValueRow;
import org.apache.fluss.row.encode.KeyEncoder;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.CloseableIterator;
Expand All @@ -43,10 +40,9 @@
import java.io.UncheckedIOException;
import java.time.Duration;
import java.util.Comparator;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.TreeMap;

import static org.apache.fluss.client.table.scanner.batch.SortedLogRows.DEFAULT_SPILL_THRESHOLD;
import static org.apache.fluss.utils.Preconditions.checkArgument;

/**
Expand All @@ -55,17 +51,14 @@
@Internal
public class KvSnapshotAndLogBatchScanner implements BatchScanner {

private final TableBucket tableBucket;
private final long logStoppingOffset;
private final int[] keyIndexesInScanRow;
@Nullable private final int[] adjustProjectedFields;
private final Comparator<InternalRow> primaryKeyComparator;
private final Map<InternalRow, KeyValueRow> logRows;
@Nullable private final SortedLogRows logRows;

@Nullable private final BatchScanner snapshotScanner;
@Nullable private final LogScanner logScanner;

private boolean logScanFinished;
private boolean logRowsLoaded;
private boolean finished;
private boolean closed;

Expand All @@ -79,18 +72,38 @@ public KvSnapshotAndLogBatchScanner(
long snapshotId,
long logStartingOffset,
long logStoppingOffset,
@Nullable int[] projectedFields) {
@Nullable int[] projectedFields,
String scannerTmpDir) {
this(
table,
tableBucket,
snapshotId,
logStartingOffset,
logStoppingOffset,
projectedFields,
scannerTmpDir,
DEFAULT_SPILL_THRESHOLD);
}

@VisibleForTesting
KvSnapshotAndLogBatchScanner(
Table table,
TableBucket tableBucket,
long snapshotId,
long logStartingOffset,
long logStoppingOffset,
@Nullable int[] projectedFields,
String scannerTmpDir,
int spillThreshold) {
checkArgument(
table.getTableInfo().hasPrimaryKey(),
"KvSnapshotAndLogBatchScanner only supports primary-key tables.");
this.tableBucket = tableBucket;
this.logStoppingOffset = logStoppingOffset;

ProjectionPlan projectionPlan = createProjectionPlan(table.getTableInfo(), projectedFields);
this.keyIndexesInScanRow = projectionPlan.keyIndexesInScanRow;
this.adjustProjectedFields = projectionPlan.adjustProjectedFields;
this.primaryKeyComparator = createPrimaryKeyComparator(table.getTableInfo());
this.logRows = new TreeMap<>(primaryKeyComparator);
KeyEncoder primaryKeyEncoder = createPrimaryKeyEncoder(table.getTableInfo());
this.primaryKeyComparator = createPrimaryKeyComparator(primaryKeyEncoder);

this.snapshotScanner =
snapshotId >= 0
Expand All @@ -100,18 +113,30 @@ public KvSnapshotAndLogBatchScanner(
: null;

boolean emptyLogRange = logStartingOffset >= logStoppingOffset || logStoppingOffset <= 0;
this.logScanFinished = emptyLogRange;
this.logRowsLoaded = emptyLogRange;
if (emptyLogRange) {
this.logScanner = null;
this.logRows = null;
} else {
this.logScanner =
LogScanner logScanner =
table.newScan().project(projectionPlan.scanProjectedFields).createLogScanner();
Long partitionId = tableBucket.getPartitionId();
if (partitionId == null) {
this.logScanner.subscribe(tableBucket.getBucket(), logStartingOffset);
logScanner.subscribe(tableBucket.getBucket(), logStartingOffset);
} else {
this.logScanner.subscribe(partitionId, tableBucket.getBucket(), logStartingOffset);
logScanner.subscribe(partitionId, tableBucket.getBucket(), logStartingOffset);
}
this.logRows =
new SortedLogRows(
table.getTableInfo()
.getRowType()
.project(projectionPlan.scanProjectedFields),
keyIndexesInScanRow,
primaryKeyEncoder,
logScanner,
tableBucket,
logStoppingOffset,
scannerTmpDir,
spillThreshold);
}
}

Expand All @@ -122,9 +147,9 @@ public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOExcep
return null;
}

// 1. collect the bounded log range into memory before merging.
if (!logScanFinished) {
pollLogRecords(timeout);
// 1. collect the bounded log range before merging.
if (logRows != null && !logRowsLoaded) {
logRowsLoaded = logRows.load(timeout);
return CloseableIterator.emptyIterator();
}

Expand All @@ -138,7 +163,9 @@ public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOExcep
keyIndexesInScanRow,
snapshotRecords,
primaryKeyComparator,
CloseableIterator.wrap(logRows.values().iterator()));
logRows == null
? CloseableIterator.emptyIterator()
: logRows.newIterator());
}

try {
Expand All @@ -154,37 +181,6 @@ public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOExcep
return sortMergeIterator;
}

private void pollLogRecords(Duration timeout) {
ScanRecords scanRecords = logScanner.poll(timeout);
for (ScanRecord scanRecord : scanRecords.records(tableBucket)) {
long logOffset = scanRecord.logOffset();
if (logOffset >= logStoppingOffset) {
logScanFinished = true;
break;
}

reduceLogRecord(scanRecord);
if (logOffset >= logStoppingOffset - 1) {
logScanFinished = true;
break;
}
}

Long consumedUpToOffset = scanRecords.consumedUpToOffset(tableBucket);
if (consumedUpToOffset != null && consumedUpToOffset >= logStoppingOffset) {
logScanFinished = true;
}
}

private void reduceLogRecord(ScanRecord scanRecord) {
ChangeType changeType = scanRecord.getChangeType();
boolean isDelete =
changeType == ChangeType.DELETE || changeType == ChangeType.UPDATE_BEFORE;
KeyValueRow keyValueRow =
new KeyValueRow(keyIndexesInScanRow, scanRecord.getRow(), isDelete);
logRows.put(keyValueRow.keyRow(), keyValueRow);
}

private CloseableIterator<LogRecord> createSnapshotRecordIterator(Duration timeout) {
if (snapshotScanner == null) {
snapshotRecordIterator = CloseableIterator.emptyIterator();
Expand All @@ -194,6 +190,11 @@ private CloseableIterator<LogRecord> createSnapshotRecordIterator(Duration timeo
return snapshotRecordIterator;
}

@VisibleForTesting
boolean isLogRowsSpilled() {
return logRows != null && logRows.isSpilled();
}

@Override
public void close() throws IOException {
if (closed) {
Expand All @@ -203,7 +204,7 @@ public void close() throws IOException {
IOUtils.closeQuietly(sortMergeIterator);
IOUtils.closeQuietly(snapshotRecordIterator);
IOUtils.closeQuietly(snapshotScanner);
IOUtils.closeQuietly(logScanner);
IOUtils.closeQuietly(logRows);

@naivedogger naivedogger Aug 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Closing logRows here does not appear to close the RocksDBLogRowsIterator that has already been passed to SortMergeReader.

The same path appears to exist in LakeSnapshotAndLogSplitScanner. Would it make sense for SortMergeReader to implement Closeable, own its input iterators, and be explicitly closed by the scanners before logRows?

@loserwang1024 loserwang1024 Aug 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Closing logRows here does not appear to close the RocksDBLogRowsIterator that has already been passed to SortMergeReader.

RocksDBLogRowsIterator is passed to org.apache.fluss.client.table.scanner.SortMergeReader.MergeIterator and will be closed here.

}

private static ProjectionPlan createProjectionPlan(
Expand All @@ -214,16 +215,19 @@ private static ProjectionPlan createProjectionPlan(
projectedFields);
}

private static Comparator<InternalRow> createPrimaryKeyComparator(TableInfo tableInfo) {
private static KeyEncoder createPrimaryKeyEncoder(TableInfo tableInfo) {
int[] physicalPrimaryKeyIndexes = getPhysicalPrimaryKeyIndexes(tableInfo);
RowType primaryKeyRowType =
Schema.getKeyRowType(tableInfo.getSchema(), physicalPrimaryKeyIndexes);
KeyEncoder primaryKeyEncoder =
KeyEncoder.ofPrimaryKeyEncoder(
primaryKeyRowType,
tableInfo.getPhysicalPrimaryKeys(),
tableInfo.getTableConfig(),
tableInfo.isDefaultBucketKey());
return KeyEncoder.ofPrimaryKeyEncoder(
primaryKeyRowType,
tableInfo.getPhysicalPrimaryKeys(),
tableInfo.getTableConfig(),
tableInfo.isDefaultBucketKey());
}

private static Comparator<InternalRow> createPrimaryKeyComparator(
KeyEncoder primaryKeyEncoder) {
return (row1, row2) -> {
byte[] key1 = primaryKeyEncoder.encodeKey(row1);
byte[] key2 = primaryKeyEncoder.encodeKey(row2);
Expand Down
Loading
Loading