[discivigour]

Purpose

Add changelog-producer=on-read so streaming readers can reconstruct inserts, updates, and deletes from data files without storing separate changelog files. For example, changing an existing key from 4 to 5 emits UPDATE_BEFORE(4) and UPDATE_AFTER(5) when the new snapshot is read.

• Plan an incremental split per partition and bucket with the new APPEND files and the historical files needed to reconstruct previous values, then reuse lookup merge logic to generate changes.
• Reuse historical lookup state across splits, refresh it against each split's file set, and bound the number of cached states with changelog-on-read.lookup-state.max-size. Release lookup and spill resources on eviction and reader close.
• Integrate the mode with Flink changelog negotiation, checkpoint/restore, and checkpoint-aligned split assignment. Keep changelog-file handling limited to producers that actually materialize changelog files.
• Support fixed-bucket primary-key tables with deduplicate or aggregation merge engines, validate incompatible configurations, and document the new mode and cache option.

Tests

• Compilation and repository formatting/style checks passed.
• 56 test cases passed across ChangelogOnReadTest, FileStoreSourceReaderTest, AlignedContinuousFileSplitEnumeratorTest, AlignedSourceReaderTest, AlignedSplitAssignerTest, and TableScanUtilsTest.
• Two AlignedSourceReaderTest cases initially failed because the test runtime selected Avro 1.11.3, which lacks DataFileWriter.setEncoder. Both passed when rerun with the repository-declared Avro 1.11.4 through temporary test classpath overrides; no dependency files were changed.
• Full end-to-end tests were not run.