Conversation
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
One or more issues must be addressed before approval.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 1
Open (2)
What changed in this PR
Adds explicit lake snapshot pinning for Paimon scan-based historical lookups, improving consistency and avoiding unnecessary latest-snapshot scans.
Changes:
- Propagates required snapshot IDs from the historical lookup manager.
- Adds snapshot-aware Paimon scan contexts and tests.
- Updates lookup context and table configuration APIs.
| File | Description |
|---|---|
| fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManager.java | Updated as part of this pull request. |
| fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java | Updated as part of this pull request. |
| fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonScanBasedTableLookuper.java | Updated as part of this pull request. |
| fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookuper.java | Updated as part of this pull request. |
| fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java | Updated as part of this pull request. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| schemaInfo.getSchema().getRowType(), | ||
| lookupMetricRecorder); | ||
| lookupMetricRecorder, | ||
| requiredLakeSnapshotIds.get(tableInfo.getTableId())); |
There was a problem hiding this comment.
all the modification origins from the fluss, so this won't happen
There was a problem hiding this comment.
There is a stale-snapshot case here even when the context and acquireLookuper() observe the same snapshot ID:
- The historical bucket records
requiredLakeSnapshotIds[tableId] = S1, and its leader remains unchanged. - A regular partition
Pwrites a new key and tiers it to snapshotS2. If the historical bucket has no progress in that commit, its lookup manager can still retainS1. Pexpires, and the first historical lookup for that key misses the local historical state.- Both context creation and
acquireLookuper()seeS1. SCAN explicitly readsS1and misses the key committed inS2.
This can happen even while S1 is still retained, without concurrent writes. Making snapshot capture and acquire atomic would still select S1.
SST avoids this particular case because its file cache is keyed by the original partition and bucket. On the first lookup of P, getOrInitializeFiles() scans the latest snapshot to initialize that entry.
Could we address this case so that SCAN does not remain pinned to a snapshot that predates the queried partition’s tiered data?
There was a problem hiding this comment.
nice catch, this is uncovered (described into the following first case) in the current PR. I think we should distinguish two cases:
Data written before the regular partition expires.
The data is written through the regular bucket and committed in S2. The historical bucket may have no progress in that commit, so its lookup manager can remain on S1. When the regular partition expires, its existing rows are not copied into the historical bucket’s local KV. The first historical lookup therefore falls back to S1 and can miss the data. This is the gap you pointed out, and making snapshot capture and acquire atomic would not fix it.
For this case, I’ll make partition expiration trigger a refresh of the historical lookup manager’s cached snapshot to the latest version. This relies on the guarantee established by #3820 that the partition’s data has been fully tiered to the lake before expiration. I’ll include this fix in the current PR.
Updates or deletes written after the partition has expired.
Updates and deletes to expired partitions go through the historical bucket. Local values and deletion markers take priority over lake data. After tiering, the snapshot is updated before local state can be cleaned up. so for this case, there is no correctness risk for the cached expired snapshotId (that may be delayed updated with a potential async updating interval)
| Long lakeSnapshotId = context.lakeSnapshotId(); | ||
| if (lakeSnapshotId != null) { | ||
| // Paimon propagates the table's snapshot and manifest caches to this copy. |
| private final @Nullable Long lakeSnapshotId; | ||
|
|
||
| /** Creates a lookup context when the lake snapshot is unknown. */ | ||
| @VisibleForTesting |
There was a problem hiding this comment.
Could we remove this test-only overload? The tests can call the six-argument constructor directly and pass null when the lake snapshot is unknown.
thanks for your quick review. and for the potential correctess risk, I have replied in the thread of #4480 (comment). please take a look. |
Purpose
this is the followup PR for #4124 .
Each scan-based lookup plans a new Paimon scan. When no snapshot ID is provided, Paimon resolves the latest snapshot first, adding metadata I/O. Passing the snapshot ID already known to Fluss avoids that discovery step. Manifest reads may still add latency on cache misses.
Brief change log
Tests
API and Format
Documentation