feat: MySQL CDC GA — type fidelity, bootstrap, cursor chunker, non-integer PKs

CHANGELOG.md

@@ -4,6 +4,80 @@ All notable changes to dbmazz will be documented here.
 
 ## [Unreleased]
 
+## [2.5.0] - 2026-05-16
+
+### Changed
+
+- **MySQL source graduates from BETA to STABLE.** The remaining type-
+  fidelity and bootstrap-UX gaps from v2.3.0 (BETA) are closed in this
+  release.
+
+### Added
+
+- **`Value::UInt64(u64)` and `DataType::UInt64`** for accurate
+  `BIGINT UNSIGNED` handling. Previously the MySQL source silently
+  wrapped values ≥ 2^63 to negative `i64`. Sinks map to `NUMERIC(20,0)`
+  (PostgreSQL / Snowflake) or `LARGEINT` (StarRocks); `value_to_json`
+  stringifies for sinks where the JSON encoder would lose precision.
+- **First-run binlog bootstrap (H5).** On a fresh start with no prior
+  checkpoint, dbmazz now captures the server's current binlog position
+  via `SHOW MASTER STATUS` **before** the snapshot worker spawns and
+  persists it as a `PROVISIONAL` checkpoint. The post-snapshot CDC
+  stream resumes from that point — no more replaying days of binlogs
+  after first start, and no more hard error when old binlogs were
+  purged on long-running databases. The first commit-boundary
+  checkpoint promotes the row to `ACTIVE`.
+- **Cursor-based snapshot chunker (M3).** Replaces the
+  `MIN(pk) / MAX(pk)` linear partitioner with keyset paging
+  (`SELECT pk FROM t WHERE pk > ? ORDER BY pk LIMIT chunk_size + 1`).
+  Each chunk has bounded row count regardless of PK density — sparse
+  distributions (gaps from DELETEs, auto-increment skips after
+  rollbacks) no longer produce empty / oversized chunks. Trade-off:
+  one bound-discovery SELECT per chunk plus the chunk-data SELECT,
+  roughly 2× the SELECT count of linear partitioning. Negligible at
+  production chunk sizes.
+- **Non-integer primary key support (M4).** Tables with `VARCHAR`,
+  `CHAR`, `UUID`, `BINARY`, or `VARBINARY` primary keys are now
+  snapshot-able. String PKs use `COLLATE utf8mb4_bin` to guarantee
+  deterministic byte-wise ordering for cursor correctness.
+
+### Fixed
+
+- **MySQL `TIMESTAMP` and `DATETIME` columns emit `Value::Timestamp(micros)`
+  directly.** The temporary ISO-string shim from PR #97 (v2.3.0) is
+  removed; sinks consume the canonical typed value. `DATETIME(p)` micros
+  are now preserved end-to-end — previously the `us` parameter of
+  `MysqlValue::Date(y,m,d,h,m,s,us)` was dropped.
+- **MySQL DECIMAL precision and scale are propagated** from
+  `information_schema.columns.NUMERIC_PRECISION` /
+  `NUMERIC_SCALE`. Previously dbmazz hardcoded `Decimal{38, 9}`
+  regardless of source. Precision > 38 clamps with a WARN log
+  (sinks reject anything wider). The converter routes
+  `MYSQL_TYPE_NEWDECIMAL` / `MYSQL_TYPE_DECIMAL` to `Value::Decimal`
+  instead of falling through to `Value::String`.
+
+### Removed
+
+- **`Source` trait dead methods**: `start_replication`,
+  `checkpoint_position`, `cleanup` — engine uses `create_loop`
+  exclusively. Internal-only breaking change.
+- **`SinkResult::last_position`** field — populated by all three
+  sinks but never consumed by the pipeline. LSN flows through
+  `PipelineEvent::lsn`. Internal-only breaking change.
+
+### Migration
+
+- **`dbmazz_snapshot_state` schema** auto-migrates from v1 to v2 on
+  first start with v2.5.0+: adds nullable `start_pk_text TEXT`,
+  `end_pk_text TEXT`, `pk_kind VARCHAR(16)`; drops `NOT NULL` on the
+  legacy `start_pk` / `end_pk` BIGINT columns; backfills the typed
+  columns for legacy Int-kinded rows. Idempotent; re-running on v2
+  is a no-op. Forward-compatible with v1 binaries that read only
+  Int rows.
+- **`dbmazz_checkpoints` schema** adds a nullable `status VARCHAR(16)`
+  column to mark bootstrap-only rows as `PROVISIONAL`. Existing rows
+  default to `ACTIVE`. Forward-compatible.
+
 ## [2.4.0] - 2026-05-16
 
 ### Changed

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dbmazz"
-version = "2.4.0"
+version = "2.5.0"
 edition = "2021"
 license = "Elastic-2.0"
 # Minimum Supported Rust Version.

README.md

@@ -123,7 +123,7 @@ Full Docker reference (env vars, healthcheck, persisting state, build-from-sourc
 | PostgreSQL 12+ | **PostgreSQL 15+** | ✅ Stable | Binary `COPY` → raw table → `MERGE` normalizer; supports hard delete |
 | PostgreSQL 12+ | **Snowflake** | ✅ Stable | Parquet → PUT (stage) → `COPY INTO` → background `MERGE`; JWT key-pair auth supported; requires `ALTER TABLE` privilege on target schema |
 | PostgreSQL 12+ | S3 / Iceberg | 🚧 Roadmap | Tracked in [issues](https://github.com/ez-cdc/dbmazz/issues) |
-| MySQL 5.7+ / 8.0+ | All sinks | 🧪 Beta | Binlog-based with GTID-aware checkpointing. See [`docs/mysql-source.md`](docs/mysql-source.md). |
+| MySQL 5.7+ / 8.0+ | All sinks | ✅ Stable | Binlog-based with GTID-aware checkpointing, BIGINT UNSIGNED, microsecond DATETIME, non-integer PK support, cursor-based snapshot chunker. See [`docs/mysql-source.md`](docs/mysql-source.md). |
 
 Adding a new sink is intentionally small: implement a 6-method `Sink` trait and CDC + snapshot work automatically. See [`docs/contributing-connectors.md`](docs/contributing-connectors.md).
 

docs/mysql-source.md

@@ -1,8 +1,10 @@
-# MySQL source (BETA)
+# MySQL source
 
 dbmazz can stream changes from MySQL 5.7+ / 8.0+ to any of the supported sinks (StarRocks, PostgreSQL, Snowflake) via the `mysql-source` cargo feature.
 
-This document covers MySQL-specific setup, configuration, and the BETA scope. For general dbmazz architecture, see [architecture.md](architecture.md).
+**Status**: ✅ Stable as of v2.5.0 (was BETA in v2.3.0 – v2.4.x). All correctness gaps from the beta scope are closed: BIGINT UNSIGNED, microsecond DATETIME, real DECIMAL precision/scale, first-run binlog bootstrap via `SHOW MASTER STATUS`, cursor-based snapshot chunking, non-integer primary keys.
+
+This document covers MySQL-specific setup, configuration, and what's in / out of scope. For general dbmazz architecture, see [architecture.md](architecture.md).
 
 ## Prerequisites
 
@@ -75,22 +77,41 @@ dbmazz follows Debezium's MySQL connector design, adapted to dbmazz's single-bin
 - **Incremental snapshot** — read-only DBLog (Andreakis & Papapanagiotou 2020), adapted to use the binlog consumer's live GTID set as low/high watermarks instead of a writable signal table. Concurrent CDC + snapshot is supported; per-PK reconciliation drops snapshot rows that lost a race to a concurrent binlog event.
 - **Schema introspection** — `information_schema` query at startup; the schema is cached in-memory and used to map binlog row payloads to the engine's generic `CdcRecord` shape.
 
-## BETA scope
-
-What works today:
+## What's supported
 
 - Binlog streaming (INSERT / UPDATE / DELETE) for ROW format, GTID mode.
-- Incremental snapshot for tables with single-column integer primary keys.
-- Concurrent snapshot + CDC reconciliation.
-- Crash-recovery startup probe (validates persisted GTID set vs `@@global.gtid_executed`).
-- Verified end-to-end against PostgreSQL and StarRocks sinks via `ez-cdc verify` (Tier 1: A1-A4, B1/B1b/B2/B3, CDC, D4, D5, C10, H1).
-
-What's known to be incomplete (tracked in `openspec/changes/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; full JSON typed roundtrip. The `C11` (Type roundtrip) verify check is expected to FAIL until this lands — use `ez-cdc verify --skip C11` in CI.
-- **Snapshot bootstrap from arbitrary binlog position**: today the daemon resumes from the persisted checkpoint or starts at the binlog tail; no first-run bootstrap that walks backwards.
-- **Cursor-based snapshot chunker**: today's chunker is range-based; cursor-based gives more uniform chunk sizes for non-contiguous PK distributions.
-- **Non-integer primary keys**: composite, UUID, and string PKs are not yet supported in the snapshot path.
+- Incremental snapshot for tables with single-column primary keys of type
+  `INT*` / `BIGINT` / `BIGINT UNSIGNED` / `VARCHAR` / `CHAR` / `BINARY` /
+  `VARBINARY` / `UUID`. Composite PK and no-PK tables are skipped with a
+  WARN log.
+- Concurrent snapshot + CDC reconciliation (read-only DBLog watermarks).
+- Cursor-based snapshot chunker — row-balanced chunks for sparse PK
+  distributions.
+- First-run binlog bootstrap via `SHOW MASTER STATUS` — no replay of
+  historical binlogs on the first start; works on long-running databases
+  with binlog purge.
+- Type fidelity: `BIGINT UNSIGNED` (full u64 range), `DATETIME(p)` with
+  microseconds, `TIMESTAMP` UTC-canonical micros, `DECIMAL(p, s)` with
+  real precision/scale, JSON pass-through.
+- Crash-recovery startup probe (validates persisted GTID set vs
+  `@@global.gtid_executed`).
+- Verified end-to-end against PostgreSQL and StarRocks sinks via
+  `ez-cdc verify` (full Tier 1 including C11 type-roundtrip).
+
+## Out of scope
+
+- **Composite primary keys** in incremental snapshot. The active-chunks
+  PK-range model is single-column today; tracked as future work.
+- **DDL replication**: MySQL `QueryEvent` → `CdcRecord::SchemaChange`
+  parsing requires a MySQL DDL parser; separate change.
+- **TINYINT / SMALLINT / MEDIUMINT / INT UNSIGNED**: they fit losslessly
+  in `i64` and stay on the existing `Int*` variants — no `UInt32`
+  variant.
+- **GEOMETRY / POINT / POLYGON** typed parsing — WKB bytes pass through
+  as `Value::Bytes`.
+- **Read-only source DB support**: dbmazz writes `dbmazz_checkpoints`
+  and `dbmazz_snapshot_state` to the source; read-only replicas without
+  write access remain unsupported.
 
 ## Verifying
 

src/connectors/sinks/postgres/mod.rs

@@ -270,7 +270,9 @@ impl Sink for PostgresSink {
             return Ok(SinkResult::default());
         }
 
-        // Track last position for checkpoint
+        // Extract last position to compute LSN for the metadata write
+        // below. Not surfaced on SinkResult (the pipeline reads LSN
+        // from PipelineEvent, not from sinks).
         let last_position = records.iter().rev().find_map(|r| match r {
             CdcRecord::Insert { position, .. }
             | CdcRecord::Update { position, .. }
@@ -334,7 +336,6 @@ impl Sink for PostgresSink {
         Ok(SinkResult {
             records_written,
             bytes_written,
-            last_position,
             schema_evolution_skipped: 0,
         })
     }

src/connectors/sinks/postgres/raw_table.rs

@@ -333,6 +333,7 @@ fn value_to_json(value: &Value) -> serde_json::Value {
         Value::Null => serde_json::Value::Null,
         Value::Bool(b) => serde_json::json!(*b),
         Value::Int64(n) => serde_json::json!(*n),
+        Value::UInt64(n) => serde_json::json!(*n),
         Value::Float64(f) => serde_json::json!(*f),
         Value::String(s) => serde_json::json!(s),
         Value::Bytes(b) => {

src/connectors/sinks/postgres/types.rs

@@ -40,6 +40,9 @@ pub fn data_type_to_pg(dt: &DataType) -> &'static str {
         DataType::Int16 => "smallint",
         DataType::Int32 => "integer",
         DataType::Int64 => "bigint",
+        // u64::MAX = 18_446_744_073_709_551_615 → 20 digits. NUMERIC(20, 0)
+        // fits the full unsigned range exactly. BIGINT would overflow.
+        DataType::UInt64 => "numeric(20, 0)",
         DataType::Float32 => "real",
         DataType::Float64 => "double precision",
         DataType::Decimal { .. } => "numeric",

src/connectors/sinks/snowflake/mod.rs

@@ -422,16 +422,6 @@ impl Sink for SnowflakeSink {
 
         // Phase 2 — data writes.
 
-        // Track last position
-        let last_position = records.iter().rev().find_map(|r| match r {
-            CdcRecord::Insert { position, .. }
-            | CdcRecord::Update { position, .. }
-            | CdcRecord::Delete { position, .. }
-            | CdcRecord::Commit { position, .. }
-            | CdcRecord::Heartbeat { position, .. } => Some(position.clone()),
-            _ => None,
-        });
-
         // Generate batch_id
         let batch_id = self.next_batch_id();
 
@@ -443,7 +433,6 @@ impl Sink for SnowflakeSink {
             return Ok(SinkResult {
                 records_written: 0,
                 bytes_written: 0,
-                last_position,
                 schema_evolution_skipped: 0,
             });
         }
@@ -479,7 +468,6 @@ impl Sink for SnowflakeSink {
         Ok(SinkResult {
             records_written: data_count,
             bytes_written: file_size,
-            last_position,
             schema_evolution_skipped: 0,
         })
     }

src/connectors/sinks/snowflake/types.rs

@@ -29,6 +29,8 @@ impl TypeMapper {
             DataType::Int16 => "SMALLINT".to_string(),
             DataType::Int32 => "INT".to_string(),
             DataType::Int64 => "BIGINT".to_string(),
+            // NUMBER(20, 0) fits u64::MAX (20 digits) exactly.
+            DataType::UInt64 => "NUMBER(20,0)".to_string(),
             DataType::Float32 => "FLOAT".to_string(),
             DataType::Float64 => "DOUBLE".to_string(),
             DataType::Decimal { precision, scale } => {
@@ -85,6 +87,11 @@ impl TypeMapper {
             Value::Null => serde_json::Value::Null,
             Value::Bool(b) => serde_json::json!(b),
             Value::Int64(i) => serde_json::json!(i),
+            // Stringify u64 to avoid JSON-number precision loss in
+            // downstream parsers that read as i64 / f64. Snowflake's
+            // VARIANT parser tolerates both forms when casting to
+            // NUMBER(20,0).
+            Value::UInt64(u) => serde_json::json!(u.to_string()),
             Value::Float64(f) => serde_json::json!(f),
             Value::String(s) => serde_json::json!(s),
             Value::Bytes(b) => {

src/connectors/sinks/starrocks/mod.rs

@@ -436,16 +436,6 @@ impl Sink for StarRocksSink {
         // Cache timestamp for entire batch
         let synced_at = Utc::now().format("%Y-%m-%d %H:%M:%S").to_string();
 
-        // Track last position from records
-        let last_position = records.iter().rev().find_map(|r| match r {
-            CdcRecord::Insert { position, .. }
-            | CdcRecord::Update { position, .. }
-            | CdcRecord::Delete { position, .. }
-            | CdcRecord::Commit { position, .. }
-            | CdcRecord::Heartbeat { position, .. } => Some(position.clone()),
-            _ => None,
-        });
-
         // Convert records to JSON batches grouped by table
         let batches = self.records_to_json_batches(&records, &synced_at)?;
 
@@ -477,7 +467,6 @@ impl Sink for StarRocksSink {
         Ok(SinkResult {
             records_written: total_written as usize,
             bytes_written: total_bytes,
-            last_position,
             schema_evolution_skipped,
         })
     }

src/connectors/sinks/starrocks/types.rs

@@ -72,6 +72,9 @@ impl TypeMapper {
             DataType::Int16 => "SMALLINT".to_string(),
             DataType::Int32 => "INT".to_string(),
             DataType::Int64 => "BIGINT".to_string(),
+            // LARGEINT is a 128-bit signed integer in StarRocks; fits
+            // the full u64 range easily.
+            DataType::UInt64 => "LARGEINT".to_string(),
             DataType::Float32 => "FLOAT".to_string(),
             DataType::Float64 => "DOUBLE".to_string(),
             DataType::Decimal { precision, scale } => {
@@ -145,6 +148,9 @@ impl TypeMapper {
             Value::Null => serde_json::Value::Null,
             Value::Bool(b) => serde_json::json!(b),
             Value::Int64(i) => serde_json::json!(i),
+            // Stringify u64 so Stream Load parses through LARGEINT
+            // without JSON-number precision loss.
+            Value::UInt64(u) => serde_json::json!(u.to_string()),
             Value::Float64(f) => serde_json::json!(f),
             Value::String(s) => serde_json::json!(s),
             Value::Bytes(b) => {

src/core/record.rs

@@ -120,6 +120,14 @@ pub enum Value {
     Null,
     Bool(bool),
     Int64(i64),
+    /// Unsigned 64-bit integer. Used for MySQL `BIGINT UNSIGNED` values
+    /// where the upper half (>= 2^63) does not fit in `Int64`. Sinks
+    /// dispatch on `DataType::UInt64` (when present) to emit the
+    /// appropriate wide-integer DDL (`NUMERIC(20,0)` for PG/Snowflake,
+    /// `LARGEINT` for StarRocks). Smaller unsigned integer types
+    /// (TINY/SMALL/MEDIUM/INT UNSIGNED) still fit in `Int64` and are
+    /// emitted as `Int64`.
+    UInt64(u64),
     Float64(f64),
     String(String),
     Bytes(Vec<u8>),
@@ -149,9 +157,16 @@ pub enum DataType {
     Int16,
     Int32,
     Int64,
+    /// Unsigned 64-bit integer (MySQL `BIGINT UNSIGNED`). Sinks map to
+    /// a wide-integer or numeric type that fits the full u64 range.
+    /// Smaller unsigned integer types stay on `Int16/Int32/Int64`.
+    UInt64,
     Float32,
     Float64,
-    Decimal { precision: u8, scale: u8 },
+    Decimal {
+        precision: u8,
+        scale: u8,
+    },
     String,
     Text,
     Bytes,
@@ -172,6 +187,7 @@ impl DataType {
             DataType::Int16
                 | DataType::Int32
                 | DataType::Int64
+                | DataType::UInt64
                 | DataType::Float32
                 | DataType::Float64
                 | DataType::Decimal { .. }

src/core/traits.rs

@@ -48,7 +48,6 @@ pub enum StageFormat {
 pub struct SinkResult {
     pub records_written: usize,
     pub bytes_written: u64,
-    pub last_position: Option<SourcePosition>,
     /// Number of `CdcRecord::SchemaChange` events that the sink could not
     /// auto-apply during this batch (e.g. StarRocks degraded mode where
     /// the target table lacks `fast_schema_evolution=true`). The pipeline
@@ -120,18 +119,6 @@ pub trait Source: Send + Sync {
     /// Setup source-side resources (slots for PG, GTID verification for MySQL)
     async fn setup(&mut self, tables: &[String]) -> Result<()>;
 
-    /// Start replication stream from given position
-    async fn start_replication(
-        &mut self,
-        position: Option<SourcePosition>,
-    ) -> Result<Box<dyn ReplicationStream>>;
-
-    /// Get current replication position for checkpointing
-    fn checkpoint_position(&self) -> Option<SourcePosition>;
-
-    /// Cleanup source-side resources
-    async fn cleanup(&mut self) -> Result<()>;
-
     /// Create the appropriate replication loop for this source type.
     /// This replaces the need for downcasting. Each implementation
     /// creates the loop variant it needs (PgReplicationLoop or MysqlReplicationLoop).