Conversation
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Exhausted filtered-empty lake iterators are not closed, and checkpoint-restore behavior lacks regression coverage.
Get a fresh assessment by requesting another Copilot review.
Pull request overview
This PR adds Flink streaming consumption for historical partition changelogs on supported primary-key tables.
Changes:
- Discovers and subscribes to
__historical__partitions. - Groups historical lake snapshots with bucket changelog offsets.
- Adds scanner, generator, integration, and documentation coverage.
File summaries
| File | Description |
|---|---|
FlinkSourceEnumerator.java |
Historical partition discovery and split initialization |
LakeSplitGenerator.java |
Historical lake/log hybrid split planning |
SeekableLakeSnapshotSplitScanner.java |
Skips empty lake splits |
HistoricalPartitionITCase.java |
End-to-end historical changelog coverage |
LakeSplitGeneratorTest.java |
Hybrid split unit tests |
SeekableLakeSnapshotSplitScannerTest.java |
Empty-split scanner test |
website/docs/engine-flink/reads.md |
Historical changelog documentation |
Review details
- Files reviewed: 7/7 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+75
to
77
| while (currentLakeRecordIterator != null && !currentLakeRecordIterator.hasNext()) { | ||
| updateCurrentIterator(); | ||
| } |
Comment on lines
+856
to
+858
| if (streaming | ||
| && hasPrimaryKey | ||
| && tableInfo.getTableConfig().isHistoricalPartitionEnabled()) { |
Consume historical changes and pair historical lake snapshots with their bucket logs. Document retirement and snapshot-parallelism limitations. Co-Authored-By: Codex <noreply@openai.com> AI-Model: gpt-6 AI-Contributed/Feature: 56/56 AI-Contributed/UT: 0/0
luoyuxia
force-pushed
the
feat/flink-historical-changelog
branch
from
September 17, 2026 09:34
9c13717 to
454f591
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: Related to #4371. This PR covers consumption of existing historical partitions; runtime partition retirement remains outside its scope.
Flink currently discovers ordinary partitions only, so changes written through the internal historical partition are invisible to streaming consumers. Subscribe to the existing historical changelog for primary-key tables with historical partition access enabled. Keep original business partition values and apply SQL partition predicates to the records.
Scope and correctness limits
This change assumes that the ordinary partitions being read do not transition to historical partitions while the job runs, including during full startup. This is a usage precondition, not an enforced guard. Long-running jobs on auto-partitioned tables can encounter this boundary.
Performance and retention limits
min(historical bucket count, source parallelism); partitions within a bucket are read sequentially. This can reduce snapshot parallelism compared with independently assigned lake splits, and slow snapshots delay log output on the same reader. Changelog-only startup has no additional snapshot grouping cost; it remains subject to the existing per-bucket log parallelism limit.Brief change log
MetadataUpdaterafter pruning ordinary partitions. PreserveAdmin.listPartitionInfos()semantics and bypass KV-snapshot initialization for the historical partition.LakeSnapshotAndFlussLogSplitand the existing generation and reader paths.API and Format
No public API, RPC, storage format, split serializer version, or checkpoint state format changes.
Admin.listPartitionInfos()continues to list ordinary partitions only.An old checkpoint from a source without historical subscription has no historical offset. If the historical partition is discovered as new after restoration, the existing discovery rule starts it at the earliest retained offset, which can replay history already represented in downstream state. This PR does not provide checkpoint-upgrade offset migration or deduplication; an unchanged state format is not an upgrade-consistency guarantee.
Documentation
Document historical changelog consumption in
website/docs/engine-flink/reads.md, including the retirement ordering limitation, existing unread-log loss, snapshot parallelism tradeoff, shared-log filtering cost, and retention boundary.