[luoyuxia]

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.

• No ordered handoff across retirement: a historical update can be emitted before unread records for the same key in the retiring ordinary bucket, affecting downstream materialization or aggregation. This applies to full and changelog-only startup, including explicit starting offsets. Existing checkpoint recovery does not add a retirement protocol.
• Existing unread-log loss remains: partition removal already unsubscribes the ordinary bucket without waiting for its backlog to drain. This PR neither introduces nor fixes that behavior, and subscribing to historical logs does not recover the skipped records.
• Consequently, this PR does not promise complete, ordered incremental consumption across partition retirement. The full-mode hybrid split only coordinates the historical lake baseline planned at startup with its corresponding historical log.

Performance and retention limits

• Historical business partitions' lake splits are grouped by bucket for full startup. Snapshot parallelism is at most 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.
• Multiple business partitions share the historical log. Partition predicates may still require reading unrelated business partitions' records before Flink filters them, increasing I/O and filtering work. The existing SQL residual-filter mechanism is unchanged.
• Existing log retention limits apply. Without a readable lake snapshot, full startup falls back to the earliest retained historical changelog and cannot guarantee a complete baseline when the required history is unavailable.
• Scope is streaming primary-key tables, within the existing historical feature's support for auto-partitioned Paimon tables with one partition key. This does not extend historical changelog subscription to log tables or batch execution.

Brief change log

• Resolve the historical partition explicitly through MetadataUpdater after pruning ordinary partitions. Preserve Admin.listPartitionInfos() semantics and bypass KV-snapshot initialization for the historical partition.
• Use the same readable lake snapshot's data and historical bucket offsets for full startup, reusing LakeSnapshotAndFlussLogSplit and the existing generation and reader paths.
• Continue past empty lake splits so a filtered-out split cannot finish the snapshot before subsequent splits are read.
• Keep existing SQL residual filters, split assignment, and checkpoint state formats.

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.

[Copilot]

🟡 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.

[luoyuxia]

@beryllw @loserwang1024 Could you please help review this pr

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Copilot review overview

6 open issues

• For non-historical partitions, if lakeSplits exist but snapshotLogOffset is missing/negative,… · New
• These retries assert an exact emitted-record count, which is brittle: if pollNext emits more than… · New
• Same resource-management issue here: UpsertWriter is not closed. Wrap the writer in… · New
• These retries assert an exact emitted-record count, which is brittle: if pollNext emits more than… · New
• UpsertWriter is created but never closed. If it holds buffers/network resources, this can leak… · New
• This doc update states that Flink streaming reads consume historical changelogs for both log tables… · New

2 resolved since last review

• This new historical-partition discovery path is also used when restoring a checkpoint created…
• When this loop skips a filtered-empty lake split, updateCurrentIterator() replaces…

_{Review effort: Lite}

[loserwang1024]

I left some comment

[loserwang1024]

Could we document what happens when a partition expires while the job is running? A historical update can arrive before an older record for the same key, and any unread records in the expired partition may be skipped.

[luoyuxia]

Added a warning in fc40c71 covering out-of-order updates and skipped unread records when a partition expires, including the impact on downstream materialized results and aggregations. It also clarifies that these limitations apply to both full and changelog-only startup modes.

[loserwang1024]

Could we add a note about restoring from an older checkpoint? If historical is discovered as a new partition after restore, the reader starts at the earliest retained offset, even with latest configured. This can replay changes already reflected in downstream state.

[luoyuxia]

Added a note in fc40c71: if an older checkpoint has no historical partition offsets and the source discovers __historical__ as new after restore, it starts at the earliest retained offset, even with scan.startup.mode = latest. The note explicitly calls out possible replay of changes already reflected in downstream state.

[luoyuxia]

@loserwang1024 Thanks for your review. Comments have been addressed. PTAL