Skip to content

feat: add MySQL CDC source (BETA) - #97

Merged
dariomazzitellireplik-coder merged 20 commits into
mainfrom
feat/mysql-source
May 10, 2026
Merged

dariomazzitellireplik-coder merged 20 commits into
mainfrom
feat/mysql-source

Conversation

@dariomazzitellireplik-coder

@dariomazzitellireplik-coder dariomazzitellireplik-coder commented May 10, 2026 •

Copy link
Copy Markdown
Collaborator

Summary

Ships MySQL as a CDC source for dbmazz, end-to-end tested against both PostgreSQL and StarRocks sinks via ez-cdc verify. This PR bundles three layers of work that all need to land together for MySQL beta to be usable:

  1. Foundation — Tano's original implementation: refactor of the Source trait for multi-source dispatch, the MySQL source connector (binlog streaming + DBLog incremental snapshot + GTID watermarks), the mysql-source cargo feature flag, and Dockerfile.mysql.
  2. Pre-merge correctness fixes — eight bugs identified in code review that block production use (commit-boundary checkpointing, BEGIN/COMMIT/ROLLBACK detection, heartbeat-driven progress, Notify-race fix in await_drain, LOW/HIGH watermarks from consumer GTID, column-ordinal preservation on unknown types, GTID startup probe, removal of meaningless GTID ordering on SourcePosition).
  3. Temporal type fidelity (minimum-viable) — discovered during the first end-to-end verify run: MySQL TIMESTAMP/TIMESTAMP2 columns were passing through as raw epoch (Int) or epoch-as-ASCII (Bytes), failing the PG sink's MERGE cast and silently dropping every row of any production-shaped table. Fixed via column-type-aware ISO formatting; forward-compatible with the proper Value::Timestamp shape planned for mysql-cdc-beta-to-ga.

What's in the box

Foundation (commits 456cae2..de3be23, by @TanoMazzitelli)

  • Source trait with validate / setup / create_loop and per-source ReplicationLoop implementations. PostgreSQL is unchanged in behaviour; MySQL is a parallel implementation.
  • MySQL source connector under src/source/mysql/: binlog stream (mysql_async + mysql_common with the binlog feature), parser, GTID set arithmetic, schema introspection, value converter.
  • DBLog incremental snapshot reconciliation under src/engine/snapshot/: per-chunk (LOW, HIGH] GTID watermarks, eviction recording for binlog events in window, drain notification when consumer's GTID set covers HIGH.
  • New MysqlBinlog { file, position, gtid_executed } variant on SourcePosition.
  • New mysql-source cargo feature. New Dockerfile.mysql for local-dev builds (production image will enable the feature in a follow-up release pipeline change).

Pre-merge correctness fixes (commits ea945cd..d7eae77)

  • C1 — Notify race in await_drain: ActiveChunks::await_drain pre-pins the Notified future before checking the drained flag so notify_waiters calls fired between the check and the await still wake the waiter. Verified by a 10 000-iteration race-stress test.
  • C2 — Mid-transaction checkpoint persistence (data loss on crash): MysqlReplicationLoop now buffers commit-boundary positions in a pending_txn_positions: VecDeque and persists only the highest commit ≤ confirmed_lsn, never an in-flight gtid_executed. Capped at FLUSH_SIZE * 4 with structured WARN on overflow.
  • C3 — LOW-watermark race: chunks register against the binlog consumer's live GTID set (Arc<RwLock<GtidSet>> shared with the parser) rather than SELECT @@global.gtid_executed. Eliminates the duplicate-emission window where events with gtid > LOW arrived before the chunk was registered.
  • C4 — DDL transactions emit Commit boundary: parser dispatches on QueryEvent leading keyword (BEGIN / COMMIT / ROLLBACK) so DDL-only transactions advance the persisted checkpoint.
  • C5 — Heartbeats advance parser state: HeartbeatEvent no longer stalls the snapshot worker in idle databases; try_drain fires on every event including heartbeats, and the heartbeat carries the live MysqlBinlog triple.
  • C6 — Heartbeat carries real position: dropped the bogus SourcePosition::GtidSet("0") placeholder that would have corrupted any checkpoint persisting a heartbeat.
  • C8 — Column-ordinal preservation on unknown TableMapEvent types: extract_col_types no longer filter_map-shifts later columns when an early column has an unrecognised type code. Unknown columns get a sentinel 0xFE byte in their original slot.
  • C7 — SourcePosition::is_ahead_of for GtidSet returns None: GTID sets are partially ordered (set inclusion); the previous lex-string compare was a landmine. Callers needing partial-order semantics use GtidSet::is_superset_of.
  • Crash-recovery startup probe: validates the persisted gtid_executed is a subset of @@global.gtid_executed via GTID_SUBSET(...); fails loud on mismatch (binlog purge, server failover, pre-fix mid-transaction checkpoint).

Temporal type fidelity (commits f88b563, 60540ec)

  • MysqlValue::Int(epoch) for TIMESTAMP (code 0x07) → ISO string formatted UTC via chrono.
  • MysqlValue::Bytes(epoch_ascii) for TIMESTAMP2 (code 0x11) (mysql_common 0.32.4's binlog-mode wire shape) → same path.
  • Plain INT/BIGINT columns remain Value::Int64 — verified by test_int_for_plain_int_column_stays_int.
  • DATETIME (already MysqlValue::Date) is unchanged.
  • This is the minimum-viable shape for MySQL beta. The proper Value::Timestamp(i64 micros) + tz_aware: bool is forward-compatible and scoped to mysql-cdc-beta-to-ga.

Tests

  • 239 unit tests pass with the mysql-source feature; 154 pass without (PostgreSQL-only build is regression-free).
  • 13 new converter unit tests covering temporal-vs-int dispatch paths.
  • 10 000-iteration await_drain_no_lost_wakeup stress test.
  • Two #[ignore] integration test files (tests/mysql_*.rs) that exercise the LOW-watermark race and concurrent snapshot+CDC against a real MySQL container via TEST_MYSQL_URL.

End-to-end validation

Verified against both supported sinks via ez-cdc verify (CLI side shipped in https://github.com/ez-cdc/ez-cdc-cli/pull/8):

MySQL → PostgreSQL    17 passed   0 failed   1 skipped (C11)    50.6s
MySQL → StarRocks     16 passed   0 failed   2 skipped (C11+A4) 29.2s

Tier 1 checks that pass on both sinks: P1-P4 (precheck), A1-A4 (target schema), B1/B1b/B2/B3 (snapshot integrity), CDC (single-row INSERT/UPDATE/DELETE), D5 (multi-row INSERT in 1 TX), D4 (TOAST column survives unrelated UPDATE — 9240B preserved), C10 (NULL roundtrip), H1 (idempotency / no drift). Skipped: C11 (Type roundtrip) — DATE fidelity is scoped to mysql-cdc-beta-to-ga. A4 (metadata table) is StarRocks-N/A by convention.

Known limitations (deferred to mysql-cdc-beta-to-ga)

  • Type fidelity: BIGINT UNSIGNED overflow at the i64 boundary; DECIMAL precision/scale not propagated from information_schema; DATETIME microseconds discarded; ENUM/SET semantics; JSON-from-bytes path. Tracked in openspec/changes/mysql-cdc-beta-to-ga.
  • Value::Timestamp(micros) and DataType::Timestamp { tz_aware }: the temporal type fix in this PR is the bytes/int → ISO string shim; the proper typed roundtrip is the next change.
  • No ez-cdc demo --source mysql: BYO MySQL via the docker recipe in tests/common/mod.rs and ez-cdc-cli's README.
  • Default dbmazz image at ghcr.io/ez-cdc/dbmazz:latest does not include mysql-source: Dockerfile.mysql for local builds; release pipeline update tracked separately.

Test plan

  • cargo build --release --features mysql-source clean
  • cargo test --features mysql-source --lib — 239 pass, 0 fail
  • cargo test --no-default-features --features sink-starrocks,sink-postgres,sink-snowflake --lib — 154 pass, 0 fail (PG-only regression-free)
  • cargo fmt --all and cargo clippy --features mysql-source clean (deny warnings)
  • docker build -t dbmazz-mysql:dev -f Dockerfile.mysql . succeeds
  • ez-cdc verify --image dbmazz-mysql:dev --source local-mysql --sink demo-pg-target --skip C11 — all pass except C11 skipped
  • ez-cdc verify --image dbmazz-mysql:dev --source local-mysql --sink demo-starrocks --skip C11 — all pass except C11 skipped (A4 SR-skipped by convention)

Post-merge

  • Replace the orphan v2.3.0 tag (currently at de3be23) with one pointing at the merge commit. The standard tag-replacement dance applies.
  • Update the release pipeline to build the mysql-source feature into the published GHCR image (separate change in infra-control-plane/release workflow).
  • Schedule mysql-cdc-beta-to-ga work to close the type-fidelity gap and graduate MySQL from BETA.

mazzitano and others added 19 commits May 3, 2026 17:49
…ex log_pos

Restore restart correctness for the MySQL source. Previously every event
emitted SourcePosition::GtidSet(format!("{:X}", log_pos)) — a hex byte
offset mislabeled as a GTID — and the GtidEvent parser dropped the real
GTID. After a binlog rotation the persisted "checkpoint" was meaningless
and restart was undefined behaviour: re-snapshot, missing events, or
duplicates depending on what the engine did with the bogus value.

Now:
- New SourcePosition::MysqlBinlog { file, position, gtid_executed }
  variant carries the canonical MySQL checkpoint triple per event.
- ParserState tracks current_file (advanced on RotateEvent),
  current_gtid_set (extended on GtidEvent via GtidSet::add_gtid),
  and current_txn_gtid (used to label Commit events).
- StateStore gains save_mysql_checkpoint / load_mysql_checkpoint and
  the dbmazz_checkpoints table for the MySQL backend now stores
  (slot_name, mysql_binlog_file, mysql_binlog_position, mysql_gtid_set)
  natively. save_checkpoint / load_checkpoint are Postgres-only and
  reject MySQL calls; the corresponding MySQL methods reject Postgres.
- MysqlReplicationLoop assigns synthetic LSNs per emitted record and
  uses the pipeline feedback channel to persist the matching
  SourcePosition triple, restoring the CLAUDE.md invariant
  "checkpoint saved BEFORE confirming to source".
- MysqlBinlogStream::new accepts an initial_gtid_set so resumed streams
  start at the persisted Gtid_set instead of replaying from the
  earliest available binlog.
- engine/mod.rs loads the MySQL checkpoint via load_mysql_checkpoint
  and feeds it to MysqlSource::create_loop as start_position.
- MysqlSource no longer carries a current_gtid field — the live
  checkpoint state lives in the replication loop and the state store,
  not on the source struct.

Drive-by:
- The "missing TableMapEvent" panic in parser.rs becomes an Err with
  context, per the no-unwrap-in-prod rule in CLAUDE.md.
- BinlogEvent gains derive(Default) (previously a manual impl) and
  one local Update helper has #[allow(clippy::type_complexity)].

The MySQL feedback channel is intentionally not forwarded to the server
(MySQL purges binlogs by time, not consumer ack); the feedback is used
only client-side to drive the state_store checkpoint write.

Tests: 221 passed --features mysql-source (153 base + new for the
SourcePosition variant serde/display/comparison and GtidSet::add_gtid
contiguous/gap/idempotent/empty paths).
Previously every MySQL connection — source, snapshot pool, and state
store — hard-coded `with_danger_accept_invalid_certs(true)`, removing
the only protection against on-path attackers between dbmazz and the
customer MySQL. The crate authors named the flag `danger_` precisely
because this is unsafe by default; dbmazz went further by removing the
option entirely.

Now:
- New `MysqlSourceConfig::tls_skip_verify: bool` (default `false`),
  toggled via `MYSQL_TLS_SKIP_VERIFY=true`.
- Single `mysql_ssl_opts(skip_verify)` helper in `source/mysql/mod.rs`
  consumed by all three connection sites:
    * MysqlSource (binlog + query connections)
    * snapshot pool (engine/snapshot/mysql.rs::build_mysql_opts)
    * state_store::build_mysql_opts (reads the env var directly so the
      catalog connection honours the same opt-out without an extra knob)
- A one-shot `warn!` fires at startup when the bypass is enabled, so
  the unsafe posture is visible in logs.
- MysqlConnectionConfig::Debug surfaces the flag (it does not contain
  credentials, so it stays in the redacted Debug impl).

Tests: 222 passed --features mysql-source (+1 covering both modes of
the snapshot opts builder); 153 passed without the feature.
…ss query

The per-table progress query in mysql_chunk_counts_for_table mixed a
PostgreSQL-only filtered-aggregate clause into a query that runs against
MySQL. MySQL ≤ 8.4 (every LTS release in the field) rejects that syntax
with a parse error at runtime — the path is only exercised when an
operator queries per-table snapshot progress, so it never showed up in
basic happy-path tests.

Replaced with the universally-supported `SUM(CASE WHEN cond THEN x ELSE 0
END)` form. Added a small lint test that scans this file for the
forbidden literal (built from runtime concatenation so the test body
itself does not contain the pattern) so a regression is caught at
`cargo test` time, not in production.

Tests: 223 passed --features mysql-source (+1 lint guard).
…snapshot

Replaces the placeholder OSA step 4 with a working implementation of the
read-only variant of DBLog Incremental Snapshot (Debezium 2022 design).
Without it, every concurrent snapshot+CDC run produced stale or duplicate
rows in the sink — the snapshot worker had no way to know that the
binlog stream had already overridden the row it was about to emit.

Now:
- New `engine/snapshot/active_chunks` module: an `Arc<RwLock<HashMap>>`
  registry shared between the snapshot worker and the binlog consumer.
  Each entry holds (table, pk_column, pk_range, low_set, high_set,
  evicted_pks, drained, notify).
- `GtidSet::is_superset_of` (in `source/mysql/gtid.rs`) — interval-aware
  comparison used to detect when the consumer has drained past a chunk's
  HIGH watermark. Constant-time in the GNO range so it stays cheap on
  the consumer hot path even for sources with millions of executed
  transactions.
- `process_mysql_chunk` is wrapped in register → SELECT → set HIGH →
  await_drain → filter evicted PKs → emit. The chunk's window starts
  at the `@@global.gtid_executed` captured before the SELECT and closes
  at the one captured after; the read-only variant requires no writes
  to the source.
- `MysqlReplicationLoop::run` walks the registry on each event,
  records evictions for events whose PK falls in an active chunk's
  range AND whose GTID lies in the (LOW, HIGH] window, and calls
  `try_drain` after every batch to wake any worker whose window has
  closed.
- `LoopContext` carries `ActiveChunks`; Postgres ignores it via `..`.
- `CdcEngine` constructs one `ActiveChunks` per pipeline and threads
  it into both the snapshot worker spawn and the consumer.
- Snapshot rows now emit `SourcePosition::MysqlBinlog` (gtid_executed
  set to HIGH) instead of the legacy `GtidSet` variant — consistent
  with T1.

Tests: 237 passed --features mysql-source (+15 covering is_superset_of
across empty / proper-superset / disjoint / overlapping / split-interval
/ huge-range cases, and ActiveChunks register / eviction / drain /
notify / chunks_for_table / pk_in_range).

Integration testing against a real MySQL is deferred to a docker CI lane;
the unit-level coverage of the registry and the GTID predicate keeps
correctness within reach without it.
- README: replace the "🚧 Roadmap" row for MySQL with a "🧪 Beta" row that
  reflects what actually works (binlog streaming, snapshot, restart from
  checkpoint, concurrent snapshot+CDC via DBLog reconciliation), and add
  a blockquote below the supported-sources table calling out the BETA
  status, the gtid_mode=ON / binlog_format=ROW requirements, the
  MYSQL_TLS_SKIP_VERIFY opt-out, and a pointer to Discussions for
  reporting edge cases that don't show up in unit tests.
- CHANGELOG: add Added / Fixed / Security entries under [Unreleased]
  covering the four bloqueantes addressed in this PR (T1 checkpoint,
  T2 TLS opt-in, T3 FILTER (WHERE) → CASE WHEN, T4 reconciliation).
Bumps Cargo.toml from 2.2.0 to 2.3.0 (MINOR — backward-compatible
addition: MySQL source + reconciliation + TLS opt-in + checkpoint
correctness fixes). Promotes the [Unreleased] CHANGELOG section to
[2.3.0] - 2026-05-04 and starts a fresh [Unreleased] header for
future entries. Cargo.lock is regenerated by `cargo build`.

The git tag (v2.3.0) is intentionally not created here — tagging
happens after the PR merges to main, by whoever runs the release.
…orkers

Wrap ParserState::current_gtid_set in Arc<RwLock<GtidSet>> and expose the
same Arc through LoopContext::consumer_gtid so the MySQL snapshot worker
can read the consumer's live GTID set without going through the loop or
querying @@global.gtid_executed against the source.

The replication loop is the single writer; std::sync::RwLock is enough
because process_typed_event is synchronous and holds the lock for one
add_gtid (write) or one format (read) per event. Snapshot worker
plumbing follows in the watermark fix.

Foundation for the read-only DBLog watermark correctness fix.
GTID sets are partial orders (set inclusion of UUID:range pairs), not
totally ordered. The previous Some(a > b) string compare was meaningless
and waiting to be triggered by the first MySQL caller. Return None and
direct callers to GtidSet::is_superset_of for partial-order semantics.

No callers depend on the GTID arm today; verified via grep across the
tree. Adds a unit test asserting None for both equal and unequal pairs.
The previous loop polled the drained flag and then awaited
notify.notified(). If try_drain() fired notify_waiters() between the
poll and the await, the wake-up was lost (notify_waiters only wakes
futures already on the queue). Under load this would hang the snapshot
worker indefinitely.

Fix: create the Notified future BEFORE the flag check so the waiter is
registered when notify_waiters fires, even if it races with the check.
tokio::pin! is required because Notified is !Unpin.

Adds a 10_000-iteration stress test that hung reliably on the old code
and passes uniformly on the fix (~0.4s for 10K runs on 4 worker
threads). Verified with the existing await_drain_blocks_until_high_
watermark_covered test that the happy path is unaffected.
The previous code captured LOW via SELECT @@global.gtid_executed BEFORE
registering the chunk. Between those two operations the binlog consumer
could process events whose GTIDs > LOW that mutated rows in the chunk's
PK range — but with no chunk registered yet, no eviction was recorded.
The subsequent SELECT then read the post-mutation value and the
snapshot emitted an Insert on top of the CDC-emitted Update for the
same PK: duplicate emission, sink-dependent corruption.

Fix: read LOW from the binlog consumer's live GTID set
(consumer_gtid: Arc<RwLock<GtidSet>> shared with ParserState) AT the
moment of registration, atomically with chunk insertion into
ActiveChunks. HIGH is captured the same way after the chunk SELECT
completes. The server's @@global.gtid_executed is never queried — the
consumer's set is the only watermark that exactly tracks what the
consumer has processed.

This is the dbmazz adaptation of Debezium's read-only DBLog watermark
contract (https://debezium.io/blog/2022/04/07/read-only-incremental-snapshots/),
where Kafka Connect surfaces the consumer's offset; we surface our
ParserState GTID set directly.

Plumbs consumer_gtid through CdcEngine, both snapshot trigger paths
(startup DO_SNAPSHOT and on-demand), run_mysql_snapshot, and
process_mysql_chunk. No new SQL queries; two SELECT @@global ones
deleted.
…beat progress

Three correctness bugs that share the parser/loop boundary:

1. Mid-transaction checkpoints (data loss on crash). The replication loop
   used to persist `gtid_executed` per acknowledged record, including
   in-flight GTIDs added on GtidEvent. A crash after row N but before
   XidEvent stored a checkpoint past an uncommitted transaction; on
   restart MySQL skipped that GTID entirely. Fix: track every
   BinlogEvent::Commit boundary in pending_txn_positions: VecDeque, and
   on feedback persist only the highest commit boundary <= confirmed_lsn.
   Capped at FLUSH_SIZE * 4 with a structured WARN on overflow.

2. DDL transactions never produced a Commit (pipeline never flushed,
   schema changes stalled). The parser only emitted BinlogEvent::Commit
   on XidEvent, but DDL transactions terminate with QueryEvent("COMMIT")
   instead. Now QueryEvent dispatches on its leading keyword:
   BEGIN/COMMIT/ROLLBACK -> commit-boundary events, anything else ->
   Ddl. Avoids double-Commit when GtidEvent+QueryEvent("BEGIN") are
   followed by XidEvent (the standard InnoDB path) by checking
   `current_txn_gtid` before emitting redundant boundary events.

3. Heartbeats stalled snapshot drains (idle databases hung await_drain).
   HeartbeatEvent now carries the live MysqlBinlog triple via
   make_position so try_drain fires on every event including
   heartbeats; CdcRecord::Heartbeat no longer carries the bogus
   GtidSet("0") placeholder.

Also:
- extract_col_types preserves column ordinals when a column type is
  unknown (sentinel UNKNOWN_COLUMN_TYPE = 0xFE), instead of silently
  filter_map-shifting later columns one slot down.
- Crash-recovery startup probe runs SELECT GTID_SUBSET(persisted,
  @@global.gtid_executed); a 0 result indicates the persisted set has
  GTIDs the server doesn't (binlog purge or pre-fix daemon's mid-txn
  checkpoint) and exits non-zero with operator-actionable guidance.

ROLLBACK is treated as an informational commit boundary that advances
the persisted checkpoint past the rolled-back txn but does NOT drop
in-flight pipeline events (those are already streamed; non-InnoDB rows
that reach the binlog before a ROLLBACK are visible to other clients
per MySQL semantics, so the sink seeing them matches source behaviour).

Unit tests for parser-level transitions and TableMapEvent column
handling are deferred: mysql_common's typed binlog events don't have
public constructors usable from external crates. The integration test
in section 7 (dev-stack docker-compose) exercises the full
parser-to-pipeline path end-to-end.
…ADME

CHANGELOG entries land under the existing [2.3.0] section because
v2.3.0 will be retagged at the squash-merge commit (replacing the
orphan tag at de3be23 that points at the buggy pre-fix HEAD). README
beta blurb updated to surface commit-boundary checkpointing as a
Debezium-faithful design choice and call out remaining beta scope
(type fidelity, bootstrap, non-integer PKs) for the GA milestone.

cargo fmt + clippy reformat sweep included.
…ent CDC

Two integration tests under tests/, both #[ignore] by default and gated
behind TEST_MYSQL_URL so CI doesn't require Docker. Setup recipe lives
in tests/common/mod.rs as a one-line docker run command.

* tests/mysql_snapshot_low_watermark.rs — exercises the C3 fix (chunk
  registration before LOW capture). Populates a table with 1000 rows,
  manually triggers a single snapshot chunk, concurrently issues an
  UPDATE for the target PK, asserts the stale snapshot value is never
  emitted and the post-update value reaches the sink.

* tests/mysql_cdc_concurrent_snapshot.rs — end-to-end concurrent
  snapshot + CDC. 5000 rows, 500 deterministic-stride UPDATEs run in
  parallel with snapshot, asserts every PK ends with the post-update
  value as its latest sink-side observation and no PK gets >1 snapshot
  Insert.

To enable test-time injection of the recording sink, CdcEngine gains a
public `with_sink_factory(config, factory)` constructor — production
callers continue to use `new(config)`. The internal SinkFactory type
alias becomes `pub` so test code can construct one. RecordingSink in
tests/common/mod.rs implements the Sink trait against an
Arc<Mutex<Vec<CdcRecord>>>.

Also fixes a pre-existing clippy lint (approx_constant) in a converter
test that uses 3.14 / 2.71828 as arbitrary float test values.
…epoch ints

MySQL TIMESTAMP arrives as MysqlValue::Int(epoch_seconds) in the binlog
ROW protocol — indistinguishable from a plain INT column at the wire
level. Without consulting the column type code from the preceding
TableMapEvent, our converter emitted Value::Int64(epoch_seconds), which
the PG sink JSON-encoded as the literal string "1778368097" and the
generated MERGE then tried to cast `(_data->>'created_at')::timestamp
with time zone` — failing the entire batch and silently dropping every
row of any table with a TIMESTAMP column.

Surfaced by the first MySQL→PG E2E run via ez-cdc verify: B1, B1b,
CDC, D5, D4, C10, B2 all failed because no row reached the sink.
Inspecting `_dbmazz._raw_dbmazz_slot` showed `created_at` stored as
a string-encoded epoch int, confirming the type-fidelity gap that
mysql-cdc-beta-to-ga is designed to fix properly.

This is the minimum-viable temporal fidelity for MySQL beta. The
proper fix (`Value::Timestamp(i64 micros)` + tz_aware DataType) is
scoped to the beta-to-ga proposal. Today's String-encoded ISO is
forward-compatible: the upcoming sink-side typed binding can
recognise both shapes during the transition period.

Plumbs `col_types: &[u8]` from `BinlogEvent::{Insert,Update,Delete}`
through `decode_row_values` → `map_binlog_value` → `map_mysql_value`,
and dispatches on MYSQL_TYPE_TIMESTAMP (0x07) / MYSQL_TYPE_TIMESTAMP2
(0x11) to format Int values as ISO via chrono. Plain INT/BIGINT
columns are unaffected — verified by `test_int_for_plain_int_column_
stays_int`.

DATETIME (MysqlValue::Date) was already structured and emits the same
"YYYY-MM-DD HH:MM:SS" shape, so it's untouched here.

Tests: 9 pass in source::mysql::converter (3 new tests around
temporal-vs-int dispatch). Full feature suite green.
…_ascii)

Initial fix dispatched only on MysqlValue::Int(epoch_seconds). Empirically
under mysql_common 0.32.4, MySQL 8.0's TIMESTAMP2 columns arrive as
MysqlValue::Bytes containing the ASCII representation of the epoch
(e.g. b"1778368097"), not Int. Verify-time inspection of the PG sink's
raw_table showed `"created_at": "1778368097"` (JSON-quoted = String,
not Number = Int) — confirming the Bytes path was active.

Extends the col-type dispatch in the Bytes arm: when col_type is
TIMESTAMP/TIMESTAMP2 and the bytes parse as an i64, format via the
same chrono helper as the Int path. Otherwise fall back to the
default UTF-8/Bytes handling.

Test: test_bytes_for_timestamp_column_emits_iso covers the new path.
Two follow-up commits (f88b563, 60540ec) added during E2E verification
introduced the minimum-viable temporal fidelity for MySQL beta. The
fix was scope-changed from mysql-cdc-beta-to-ga to pre-merge after the
first MySQL→PG verify run revealed that all data writes failed when
the source schema contained a TIMESTAMP column — making MySQL CDC
unusable for any production-shaped table without this fix.

CHANGELOG note explains both wire-protocol shapes (Int and Bytes paths)
and the forward-compatibility plan toward Value::Timestamp(micros).
The sink side already had a clean factory (`connectors::sinks::create_sink`)
but sources were instantiated by an ad-hoc `init_source` private method
on `CdcEngine`. Asymmetric: a contributor adding a new source had to
hunt through the engine module rather than land everything in
`src/source/mod.rs` next to the existing sources.

Adds `pub async fn create_source(config: &SourceConfig) -> Result<Box<dyn
Source>>` in `src/source/mod.rs`. Same per-variant match dispatch +
feature-gated MySQL arm as `create_sink`. The engine's `init_source`
becomes a one-line wrapper.

Two unit tests parallel the create_sink test suite: PG dispatch
(verified by the connection-error shape since no live PG in unit env),
MySQL dispatch (MysqlSource::new is connection-less so the call
returns Ok). Tests gated on the appropriate features.

Also closes the README MySQL beta section: trims the inline blurb to
a single sentence + link, and moves the prerequisites / configuration /
BETA scope detail into a new docs/mysql-source.md so the README's
supported-pairs table stays scannable. CHANGELOG entry under [2.3.0]
adds the factory pattern alongside the MySQL source itself.

Tests: 245 (mysql-source) + 155 (PG-only) — no regression. Both
clippy gates clean.
@dariomazzitellireplik-coder
dariomazzitellireplik-coder merged commit 4d63d3a into main May 10, 2026
2 checks passed
@dariomazzitellireplik-coder
dariomazzitellireplik-coder deleted the feat/mysql-source branch May 10, 2026 03:09
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants