feat(cdc): ALTER SOURCE ... REFRESH SCHEMA HISTORY — rebuild Debezium schema history at the current offset without losing the position

[yuhao-su]

Problem

For a shared MySQL CDC source, Debezium's schema history lives in object storage (OpendalSchemaHistory, <data_dir>/rw-cdc-schema-history/source-<id>/schema_history_<n>.dat). On every engine start Debezium replays that history and applies only the records whose position is "at or before" the current offset (BinlogHistoryRecordComparator.isPositionAtOrBefore). When the effective table definition produced by that replay no longer matches the rows in the binlog, the connector fails on the first row of that table:

Error processing row in <table>, internal schema size N, but row size N+1, restart connector with schema recovery mode.

and keeps failing on every restart, because each restart re-derives the same stale definition. The source emits nothing until someone intervenes.

We hit this in production after an upstream MySQL instance swap. The new instance was a replica that replayed several days of backlog with the original event timestamps, so the offset's ts_sec (event time) ended up older than the wall-clock ts_sec of the schema snapshot RisingWave had taken on the new instance. The comparator compares timestamps whenever server_id differs (snapshot records carry server_id=0), judged the newer-looking snapshot to be "in the future", skipped it, and fell back to a definition from weeks earlier. The stream then broke on the first restart and stayed broken. (Upstream design gap; will be reported to Debezium separately.)

There is no supported way to rebuild the schema history without losing the position

• Debezium's own advice, "restart connector with schema recovery mode" (delete the history and restart), goes through NoDataRecoverySnapshotter in RisingWave. That snapshotter does not override shouldStreamEventsStartingFromSnapshot() (default true), so BinlogSnapshotChangeEventSource.determineSnapshotOffset discards the stored offset and re-seeds from SHOW MASTER STATUS: the schema is rebuilt but everything between the old offset and head is silently skipped. (Debezium's RecoverySnapshotter returns false here for exactly this reason.)
• ALTER SOURCE ... RESET starts from head: same data gap.
• DROP SOURCE ... CASCADE + recreate: full re-snapshot of every table.

What actually worked was a manual procedure: ALTER SOURCE ... RESET + SET source_rate_limit = 0 to force a head snapshot (which writes every table's current definition into the history but at a position beyond the offset, so the comparator ignores it), read the history files back, copy the wanted records into a new schema_history_<n+1>.dat with the position rewritten to sit just before the stuck offset, PUT it to object storage, and re-inject the original offset with risectl meta inject-source-offsets. It recovered the source with no data loss and no rebuild, but it is far too fragile to hand to operators.

Proposal

Add a supported command for shared CDC sources (name to be bikeshed):

ALTER SOURCE <source_name> REFRESH SCHEMA HISTORY;

Semantics:

1. Pause the source's split reader (same mechanism as source_rate_limit = 0).
2. Take a schema-only snapshot of all captured tables from the upstream (the existing no_data snapshot path).
3. Stamp every produced history record with the source's current persisted offset (file/pos/server_id from the split state) instead of SHOW MASTER STATUS, so that the comparator applies them on the next recovery (same server_id → file/pos comparison; they sort at the offset and after every existing record).
4. Append them to the existing history as a new file (never rewrite or delete existing files), rebuild the reader with the unchanged offset, resume.

Guard rails:

• Refuse when the upstream reports replication lag (SHOW REPLICA STATUS / Seconds_Behind_Source > 0): the snapshot would describe a lagging state.
• Document that the refreshed definitions are the upstream's current schema. If an in-binlog DDL exists between the current offset and head for some table, that table will mismatch until the stream reaches the DDL; the command should be used as soon as the incident is detected, while binlog retention still covers the offset.

Alternative implementation: fix NoDataRecoverySnapshotter.shouldStreamEventsStartingFromSnapshot() to return false, then implement the command as "archive the history files + restart the engine", which makes Debezium walk the shouldSnapshotOnSchemaError path and position the new records at the previous offset by itself. That override is a correctness fix on its own (today a missing history + existing offset silently loses data) and probably deserves a separate issue.

Related follow-ups

• Observability: during recovery, count records that were skipped although their file/pos is at or before the offset (the signature of the timestamp fallback misfiring), log a warning and expose a metric. Wrap the Debezium "restart connector with schema recovery mode" error with a pointer to the RisingWave command instead.
• MySqlCdcSplit::update_offset has no monotonicity guard (the Postgres split has one); a backward file/pos jump should at least be logged loudly.
• Document ALTER SOURCE ... RESET (it is not in the docs today) and publish an upstream-migration runbook for CDC sources, with a separate variant for replica upstreams.

Code pointers

• java/connector-node/risingwave-source-cdc/.../debezium/internal/OpendalSchemaHistory.java — history storage, schema_history_<n>.dat, sequence numbering.
• java/connector-node/risingwave-source-cdc/.../debezium/internal/NoDataRecoverySnapshotter.java — missing shouldStreamEventsStartingFromSnapshot() override.
• java/connector-node/risingwave-connector-service/.../DbzConnectorConfig.java — snapshot.mode selection for shared sources.
• src/stream/src/executor/source/source_executor.rs — Mutation::ResetSource, Mutation::InjectSourceOffsets, throttle-driven reader rebuild (all reusable building blocks).
• src/frontend/src/handler/reset_source.rs, src/ctl/src/lib.rs (inject-source-offsets).
• Debezium 3.2.4 io.debezium.connector.binlog.history.BinlogHistoryRecordComparator#isPositionAtOrBefore, io.debezium.connector.binlog.BinlogSnapshotChangeEventSource#determineSnapshotOffset.