ref(ddl): Improve DDL handling and ordering

[iambriccardo]

Summary

This PR makes schema-change replication reliably ordered across streaming, retries, table-sync handover, and process restarts.

It establishes a stable schema-version contract:

• every applicable supported source change that affects the downstream schema produces an ordered schema snapshot;
• Event::Relation carries the effective schema for subsequent rows without participating in DML sequence numbering; and
• table-sync handover durably preserves the decoding state required by the apply worker.

Together, these changes give destinations a monotonic schema timeline while preserving existing DML ordering and deduplication semantics.

Motivation

Previously, publication column-list changes could modify the effective downstream schema without producing a new snapshot ID. A destination could therefore observe one physical table snapshot with different replication masks, without a stable way to order those states.

Relation events also carried DML sequence information even though PostgreSQL can emit them repeatedly across replication connections. This made DML ordinals depend on connection-local schema metadata rather than solely on source transactions.

Finally, during table-sync handover, PostgreSQL may have already sent the relevant relation to the apply connection while the table-sync worker still owned the table. PostgreSQL does not necessarily resend that relation after ownership changes, so the apply worker needs a durable fallback to decode the first post-handover row safely.

What changed

Stable schema snapshots

• Emit transactional schema snapshots for supported ALTER TABLE operations.
• Capture applicable ALTER PUBLICATION changes, including publication column-list changes.
• Derive a new snapshot ID from the transactional logical message LSN whenever an applicable supported schema message is emitted.
• Scope publication messages so pipelines process only changes affecting their configured publication.
• Update the rust-postgres dependency to expose the logical message LSN.

A publication-filter change snapshots the complete physical table schema even when that schema is unchanged. The logical message gives that snapshot a newer ID, while the following Relation supplies the new publication and replica-identity masks. Destinations therefore compare two ordered (snapshot ID, replication mask) states instead of two incomparable masks sharing one snapshot ID.

For ALTER PUBLICATION, snapshots are emitted for affected tables that are current members of the publication after the operation. Removing a table from a publication does not emit a new snapshot for the removed table.

Relation and DML ordering

• Remove EventSequenceKey from Event::Relation.
• Continue ordering DML with EventSequenceKey(commit_lsn, tx_ordinal).
• Treat Relation as connection-local schema metadata that establishes the active schema for following rows.
• Preserve relation ordering within destination event batches without allowing repeated relations to shift DML ordinals.

Destinations use the relation’s snapshot ID and replication mask to identify schema state. DML sequence keys remain responsible only for ordering and deduplicating row mutations.

Decoder ownership and table-sync handover

The apply loop now tracks explicit per-connection decoding state:

• WaitingForRelation { snapshot_id } means ETL knows which schema snapshot is required but does not yet have the relation masks needed to decode rows.
• WithSchema contains the complete replicated schema needed for decoding.

At table-sync completion, SyncDone durably stores:

• the schema snapshot ID;
• the publication replication mask; and
• the replica-identity mask.

The apply worker can restore this decoder if its first post-handover DML arrives without PostgreSQL resending the relation. ETL rejects incomplete handovers instead of guessing historical masks.

SyncDone to Ready

For a handover at LSN H, a table transitions from SyncDone(H) to Ready only when:

1. the current apply connection has reached or passed H;
2. the persisted apply checkpoint has reached or passed H; and
3. the current connection has materialized WithSchema decoding state.

These conditions protect two separate guarantees:

• the materialized decoder preserves same-connection continuity after handover; and
• the persisted checkpoint ensures that a restart resumes at or beyond the handover boundary.

Until all three conditions hold, SyncDone retains the durable decoder fallback.

Destination schema safety

All built-in destinations apply the same conservative validation before destination DDL:

• an exact match of snapshot ID and replication mask is idempotent;
• an older snapshot is rejected before reverse DDL can be attempted;
• the same snapshot with a different replication mask is rejected as ambiguously ordered; and
• a newer snapshot proceeds through destination-specific DDL handling.

DML deduplication is deliberately not used to justify stale schema replay. A row watermark cannot prove that reverse or destructive DDL is safe.

Recovery from interrupted destination DDL remains destination-specific:

Destination	Current behavior
BigQuery	Does not automatically repair interrupted Applying metadata.
ClickHouse	Can recover an interrupted operation when the arriving relation exactly matches the recorded target.
DuckLake	Can recover the exact recorded target, including during startup when the target can be reconstructed from durable schema state.
Snowflake	Retries supported initial-setup recovery, but not interrupted schema changes.
Iceberg	Accepts an identical relation but rejects schema changes.

Unsupported or ambiguous recovery states fail explicitly and may require table resynchronization. More sophisticated recovery can be added per destination without changing the source-side snapshot and relation contract.

Known limitations

Idle tables may remain in SyncDone

If no apply-owned relation or DML arrives after table-sync handover, the table may remain in SyncDone indefinitely.

Initial copy and catch-up are already durably complete. Retaining SyncDone preserves the decoder needed if the first future row arrives without a preceding relation. That relation or row will materialize the apply-worker decoder and allow the table to become Ready.

DDL without a following relation during catch-up

A table-sync worker may decode DDL during catch-up without receiving a following relation when no DML occurs for that table.

The DDL snapshot contains the physical schema but not the exact historical publication and replica-identity masks. Reading the current PostgreSQL catalog would also be unsafe because it may already reflect later changes.

ETL therefore fails the table sync before persisting an incomplete SyncDone. Retrying or resynchronizing the table after schema activity has settled is currently required.

Destination recovery is intentionally conservative

Stale relations and unsupported interrupted Applying states may require table resynchronization. This PR prioritizes explicit failure over reverse DDL or inferred recovery that could corrupt destination data.

Backward Compatibility

The system remains backward compatible with the previous SyncDone schema. If an existing SyncDone state lacks decoding, the system will simply be unable to install the SyncDone decoding state for tuple decoding. This results in the same failure scenario as before, so it is not a regression; the system behaves exactly as it did previously. That existing behavior is addressed in this PR with the proper decoding state handover.

[iambriccardo]

This was a much needed correctness fix, so that we persist state in memory only after the store has acknowledged it.

[iambriccardo]

I have just inlined this, since having 016x in the Display impl of EventSequenceKey made it too tied with BigQuery specifics.

[coveralls]

coverage: 74.37% (+0.1%) from 74.247% — fix-lsn-relation-event into main