Skip to content

[Flink] Add configurable initial scan mode for streaming compaction - #9548

Open
lilei1128 wants to merge 6 commits into
apache:masterfrom
lilei1128:fix-compact-snapshot-expire
Open

[Flink] Add configurable initial scan mode for streaming compaction#9548
lilei1128 wants to merge 6 commits into
apache:masterfrom
lilei1128:fix-compact-snapshot-expire

Conversation

@lilei1128

@lilei1128 lilei1128 commented Sep 2, 2026

Copy link
Copy Markdown
Contributor

Purpose

When a dedicated streaming compaction job starts for a table with existing
snapshots but no COMPACT snapshot, it scans from the earliest available
snapshot to catch up with the historical backlog. For tables with many
snapshots, this catch-up can take a long time, while snapshot expiration may
remove snapshots that are still being scanned. The compaction job can then
fail and repeatedly restart.

Add compaction.initial-scan-mode to allow the initial compaction scan to use
the latest snapshot as an ALL baseline when historical changelog replay is not
required. The default remains earliest to preserve existing behavior and
changelog semantics. When a COMPACT snapshot exists, compaction continues from
the snapshot after the latest COMPACT snapshot.

Tests

Add tests covering:

  • default earliest behavior;
  • latest baseline behavior without a COMPACT snapshot;
  • latest baseline followed by incremental delta scans;
  • multiple partitions and buckets in the latest baseline;
  • existing COMPACT snapshot behavior.

@lilei1128 lilei1128 changed the title [core] Fix streaming compaction recovery from expired snapshots [Flink] Fix dedicated streaming compaction after snapshot expiration Sep 2, 2026
@lilei1128
lilei1128 force-pushed the fix-compact-snapshot-expire branch from cf95735 to e009f10 Compare September 3, 2026 08:41
@lilei1128 lilei1128 changed the title [Flink] Fix dedicated streaming compaction after snapshot expiration [Flink] Add configurable initial scan mode for streaming compaction Sep 3, 2026
+ "dedicated internal streaming scan.");

public static final ConfigOption<CompactionInitialScanMode> COMPACTION_INITIAL_SCAN_MODE =
key("compaction.initial-scan-mode")

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.

it is for ContinuousCompactorStartingScanner, I think the name should be related to continuous.

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.

continuous-compaction.initial-scan-mode is more appropriate name.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants