[core] Preserve timestamp values on changelog retraction records

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

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 Can I get a review on the PR.

[JingsongLi]

The Cassandra event-timestamp use case has end-to-end value, but rewriting ordinary before-image fields needs a different contract boundary before merging. The Flink source still advertises a normal all-kinds changelog, whose -U/-D rows must retract the previously emitted values.

For example, after +I(id=1, event_ts=50), an update to event_ts=100 now emits -U(id=1, event_ts=100), +U(id=1, event_ts=100). A downstream WHERE event_ts < 75 never sees the retraction and keeps the old row. GROUP BY event_ts also retains the old group because the retraction addresses the new group. I reproduced the emitted records using the exact merge/wrapper source and checked them with filter/group consumer probes; this was not a full Flink job.

Please expose the incoming event timestamp as separate metadata/a CDC envelope for the external sink, preserving the stored before-image, or define and enforce a separate restricted consumption mode instead of advertising these rows as a normal relational changelog. Add an end-to-end consumer test covering both a filter and an aggregation before accepting the new option. Also clarify that the current helper reads the merged result, which need not equal the incoming event for aggregation merge engines.