[junmuz]

Purpose

• Adds a new config option changelog-producer.preserve-sequence-on-retract that accepts a comma-separated list of column names whose values should be taken from the incoming event (rather than the stored row) when producing changelog retraction records (-U, -D)
• This is useful when delete or update events carry an event timestamp that downstream consumers need — e.g., external systems like Cassandra that rely on WRITETIME for conflict resolution
• Only supported with the lookup changelog producer
• This also allows alignment with FlinkCDC that emits retract events with db event timestamps.
• Added clearly in document not to use the preserved field as aggregated values or filtering keys.

Tests

• Unit tests for delete retraction preserving the event's sequence field value
• Unit tests for update retraction (-U) carrying the new event's sequence field value
• Unit tests confirming original behavior is preserved when the option is not set

Manual verification

• Tested out after configuration, the table correctly produces the changelog events.

CREATE TABLE test.test_table ( 
    id INT, 
    name STRING, 
    updated_at TIMESTAMP, 
    op_type STRING
) TBLPROPERTIES (
    'primary-key' = 'id', 
    'changelog-producer' = 'lookup', 
    'write-only' = 'false',
    'changelog-producer.preserve-field-on-retract' = 'updated_at',
    'rowkind.field' = 'op_type'
)

+-------+---+-------+--------------------------+
|rowkind|id |name   |updated_at                |
+-------+---+-------+--------------------------+
|-D     |1  |one_new|2026-09-09 15:11:45.082928|
|+U     |1  |one_new|2026-09-09 15:11:03.804031|
|-U     |1  |one    |2026-09-09 15:11:03.804031|
|+I     |1  |one    |2026-09-09 15:10:31.101137|
+-------+---+-------+--------------------------+

[junmuz]

@JingsongLi Given the amount of changes in the other PR, I feel like this PR is a better approach. I have tried clearly mentioning that the fields shouldn't be used as aggregated values nor as filtering keys. I would love to get your thoughts on this.

[JingsongLi]

The event-timestamp preservation use case has end-to-end value, but the production review found two correctness gaps below. The ordinary deduplicate update/delete path was also checked through real table commits and persisted changelog reads.

[junmuz]

The event-timestamp preservation use case has end-to-end value, but the production review found two correctness gaps below. The ordinary deduplicate update/delete path was also checked through real table commits and persisted changelog reads.

Both of the comments have been addressed.

[JingsongLi]

Requirement fit: SUPPORTED. Implementation: FINDINGS.

The earlier partial-update and rename concerns are addressed: validation now restricts the option to deduplicate + lookup, and persisted rename/update/delete tests cover rewritten option references. The documented external sink restrictions are exercised by the Flink integration test.

A new Spark postpone-write interaction remains below. I reproduced it with a real filesystem table and the actual SparkPostponeStagedCommitter constructor using this head's Core/API classes. The unchanged Spark coordinator source is identical to the compiled class used by the probe. Removing the preserve option, clearing it on the runtime copy, or overlaying only the exact baseline SchemaValidation all make the same coordinator initialize successfully.

Verification: normal JDK 8 Maven verify passed 120 focused tests (116 Core, 4 real Flink external-upsert-sink cases), and current CI is green. The additional integration probe fails on this head as described inline.

[JingsongLi]

[P2] Clear preserved fields from internal rescale copies that disable changelog

A valid table with bucket=-2, postpone.batch-write-fixed-bucket=true, changelog-producer=lookup and preserve-field-on-retract=event_ts fails Spark batch-write coordinator initialization. SparkPostponeStagedCommitter eagerly builds rescaleWriteBuilder by copying the table with changelog-producer=none, retaining the new preserve option; AbstractFileStoreTable.copy runs this validation and throws before commit. The real table creation succeeds, and the actual coordinator constructor succeeds without the preserve option or with the exact baseline validation. Keep the CREATE/ALTER restriction, but clear the preserve option on internal copies that deliberately disable changelog production, and cover a Spark postpone fixed-bucket write with preservation enabled.

[junmuz]

The internal rescale copy in SparkPostponeStagedCommitter now has changelog-producer.preserve-field-on-retract removed. That copy forces changelog-producer=none and writes no changelog, so the option has no effect there. The CREATE/ALTER restriction is unchanged, and the table itself keeps the option. This is the only place that forces changelog-producer=none on a table copy, so no other internal copies need the same change.

I added a Spark test to PostponeBucketTableTest. It uses a postpone fixed-bucket table with changelog-producer=lookup and preserve-field-on-retract enabled. The test writes enough rows to trigger a load-factor rescale from 1 to 16 buckets. It then upserts an existing key and checks the table contents and options. Without the fix, the test fails with the reported validation error from the SparkPostponeStagedCommitter constructor. With the fix, the whole suite passes on Spark 3.