[flink] Flink batch read based on SST merge. - #3677
Conversation
a9044ac to
1496616
Compare
naivedogger
left a comment
naivedogger
left a comment
There was a problem hiding this comment.
Thanks @loserwang1024 for the PR! Overall looks good to me — left a few comments, PTAL.
| IOUtils.closeQuietly(logScanner); | ||
| } | ||
|
|
||
| private static ProjectionPlan createProjectionPlan( |
There was a problem hiding this comment.
nit: this is logically identical to LakeSnapshotAndLogSplitScanner#getNeedProjectFields(pk-column completion + adjustProjectedFields mapping, just packaged differently). Maybe worth extracting into a shared helper in this package so future fixes don't need to be applied twice?
| return true; | ||
| } | ||
|
|
||
| CloseableIterator<InternalRow> snapshotRows = snapshotScanner.pollBatch(timeout); |
There was a problem hiding this comment.
initializeSnapshotIterator calls snapshotScanner.pollBatch() only once and treats the returned iterator as the complete snapshot stream. This assumes a single pollBatch returns the whole snapshot, which holds for KvSnapshotBatchScanner today but violates the poll-until-null contract of BatchScanner — a chunked implementation would silently lose data. Could we make SnapshotRecordIterator lazily poll until pollBatch returns null? If the underlying scanner ever becomes chunked (like KvBatchScanner), this would silently drop snapshot data.
| import static org.apache.fluss.utils.Preconditions.checkArgument; | ||
|
|
||
| /** | ||
| * The hybrid split for first reading the snapshot files and then switch to read the cdc log from a |
There was a problem hiding this comment.
nit: the class javadoc still only describes the streaming two-phase behavior ("first reading the snapshot files and then switch to read the cdc log"). Since this split now also models bounded batch reads (merged snapshot + bounded log, possibly with no snapshot at all via NO_SNAPSHOT_ID), could we update the javadoc and add a doc for the NO_SNAPSHOT_ID semantics?
91a5266 to
6083d11
Compare
6083d11 to
34b3eb7
Compare
leonardBang
left a comment
leonardBang
left a comment
There was a problem hiding this comment.
Thanks @loserwang1024 for the contribution and @naivedogger for the review, the PR is in a good scope, I only left one comment
| String.format( | ||
| "Batch only supports when table option '%s' is set to true.", | ||
| ConfigOptions.TABLE_DATALAKE_ENABLED)); | ||
| context.callAsync(flussOnlyBatchSplitGenerator::generate, this::handleSplitsAdd); |
There was a problem hiding this comment.
Would it make sense to exclude buckets that already have restored unassigned splits from batch split generation?
start() first restores unassignedSplits into pendingSplitAssignment, but then unconditionally calls startInBatchMode(). The generator's tableBucketSkipper only checks assignedTableBuckets. Therefore, if split generation completes before the restored split is assigned to a reader, the bucket is not skipped and a second split is generated for the same bucket. Both splits are subsequently added and assigned, which may produce duplicate rows after recovery.
I reproduced this by restoring one unassigned bounded PK split, running the one-time generator before registering readers, and observed two splits for the same bucket.
Could we include the pending/unassigned buckets in the skipper? It would also be helpful to add a recovery test that checkpoints a bounded split before assignment and verifies that exactly one split is assigned per bucket.
There was a problem hiding this comment.
Would it make sense to exclude buckets that already have restored unassigned splits from batch split generation
why need to considered restored unassigned splits in batch read?
There was a problem hiding this comment.
Correction to my previous comment: After re-checking Flink's execution-mode in AI-generated test, the RuntimeExecutionMode.BATCH is missed, therefore, the restore scenario I described is not reachable through the normal batch SQL path covered by this PR.
3 participants
Purpose
Linked issue: close http://github.com/apache/fluss/issues/3656
Brief change log
Tests
API and Format
Documentation