[FLINK-38647] Support changelog-mode=upsert + null-tolerance

[cansakiroglu]

FLINK-38647 reports a NullPointerException from the Postgres Pipeline source when a captured table has REPLICA IDENTITY DEFAULT and an UPDATE arrives. Root cause: PostgresDataSource hardcodes DebeziumChangelogMode.ALL, which makes the deserializer extract a before-image that is null under DEFAULT replication, hence NPE'ing.

This change fixes the issue from two angles:

1. Connector side: expose 'changelog-mode' YAML option on the Postgres Pipeline connector. Accepts 'all' (default, current behaviour) or 'upsert'. In upsert mode the source emits UPDATE events with before == null and only after populated, so the pipeline runs cleanly under REPLICA IDENTITY DEFAULT without requiring FULL (which increases WAL volume).

2. Runtime side: make serializer null-tolerant. Three changes:

• copy() null-guards each recordDataSerializer.copy() call so the chained CopyingChainingOutput.pushToOperator path stops NPE'ing on null before/after.
• serialize() writes 2 leading boolean presence flags before conditionally writing each record. Required because the previous serialize() already skipped writing null fields, but deserialize() always tried to read two records back.
• deserialize() reads the 2 flag bytes and skips the corresponding read when absent.

[haruki-830]

I noticed the existing SQL connector (flink-connector-postgres-cdc) defines this option using enumType(DebeziumChangelogMode.class). It might be worth aligning with that approach — it would give us compile-time safety and also avoid the null handling concern in the factory. Something like:

public static final ConfigOption<DebeziumChangelogMode> CHANGELOG_MODE =
        ConfigOptions.key("changelog-mode")
                .enumType(DebeziumChangelogMode.class)
                .defaultValue(DebeziumChangelogMode.ALL);

[haruki-830]

If enumType is adopted for CHANGELOG_MODE, this switch block could be simplified to a single config.get(CHANGELOG_MODE) call.
Also, it might be good to add a primary key validation for upsert mode, similar to what the SQL connector does in PostgreSQLTableFactory.

[haruki-830]

This looks like a wire format breaking change — old deserializers would interpret these 2 boolean bytes as the start of the before RecordData, which could cause stream corruption on checkpoint restore.Would it be possible to bump DataChangeEventSerializerSnapshot.CURRENT_VERSION to 2 in this PR and add a version-aware fallback?
I think this would be important to include before merging, since checkpoint restoration is a production-critical path.

[haruki-830]

Nice PR! I've left a few inline comments with suggestions on the implementation. It would also be great to have some test coverage — mainly roundtrip tests for the serializer with null before/after, and a basic unit test. Looking forward to the updates!

[dev-donghwan]

@cansakiroglu

Thank you for this PR.

This PR resolves the source part, but the sink part is still broken.

Case:

1. flink-cdc pipeline -> Paimon sink (changelog-mode: upsert)
2. the Paimon sink calls PaimonWriterHelper.convertEventToFullGenericRows()
3. the UPDATE branch requires before -> NPE, and the job fails permanently

SinkWrapperException: Failed to handle event ... DataChangeEvent{tableId=..., before=null, ..., op=UPDATE}
Caused by: java.lang.NullPointerException:
  Cannot invoke "org.apache.flink.cdc.common.data.RecordData.getArity()" because "recordData" is null
  at PaimonWriterHelper.convertRecordDataToGenericRow(...)
  at PaimonWriterHelper.convertEventToFullGenericRows(...)   // UPDATE branch reads before() unconditionally

The reason is that a PostgreSQL upsert event does not have before, only after — the same failure pattern FLINK-39684 fixed for REPLACE in the same method.

We verified a one-line guard end-to-end (PG -> Paimon, DML + schema evolution):

case UPDATE:
    if (hasPrimaryKey && dataChangeEvent.before() != null) {   // skip UPDATE_BEFORE when there is nothing to build it from
        ...

For primary-key tables this is safe: +U-only and -U++U converge to the same merged state, and PK-changing updates never arrive as UPDATE (Debezium decomposes them into DELETE + CREATE). Repro test: DataChangeEvent.updateEvent(tableId, null, afterData) in PaimonWriterHelperTest — every existing test passes a non-null before.

Note: the Kafka sink has the same pattern — CanalJsonSerializationSchema and DebeziumJsonSerializationSchema read before() for UPDATE (and REPLACE) too.

Would you prefer to include the sink-side null-tolerance in this PR, or should we file a separate JIRA as a follow-up to FLINK-39684?

cc. @haruki-830 @leonardBang @yuxiqian