diff --git a/crates/pg2iceberg-snapshot/src/lib.rs b/crates/pg2iceberg-snapshot/src/lib.rs index 6258b9c..efe75d2 100644 --- a/crates/pg2iceberg-snapshot/src/lib.rs +++ b/crates/pg2iceberg-snapshot/src/lib.rs @@ -351,7 +351,7 @@ impl Snapshotter { let new_key = pk_key(last_in_chunk, &pk_cols); if let Some(prev) = last_pk_key.as_deref() { - if new_key.as_str() <= prev { + if new_key.as_str() == prev { return Err(SnapshotError::Source(format!( "snapshot source did not advance past PK {prev} for table {}; got {new_key}", schema.ident diff --git a/crates/pg2iceberg-snapshot/tests/snapshot_e2e.rs b/crates/pg2iceberg-snapshot/tests/snapshot_e2e.rs index 6e7fbc2..e3c87b5 100644 --- a/crates/pg2iceberg-snapshot/tests/snapshot_e2e.rs +++ b/crates/pg2iceberg-snapshot/tests/snapshot_e2e.rs @@ -3,13 +3,14 @@ //! live replication and verify Iceberg state == PG ground truth at every //! point. +use async_trait::async_trait; use pg2iceberg_coord::schema::CoordSchema; use pg2iceberg_coord::Coordinator; use pg2iceberg_core::typemap::IcebergType; use pg2iceberg_core::{ - ColumnName, ColumnSchema, Namespace, PgValue, Row, TableIdent, TableSchema, Timestamp, + ColumnName, ColumnSchema, Lsn, Namespace, PgValue, Row, TableIdent, TableSchema, Timestamp, }; -use pg2iceberg_iceberg::read_materialized_state; +use pg2iceberg_iceberg::{pk_key, read_materialized_state}; use pg2iceberg_logical::pipeline::CounterBlobNamer; use pg2iceberg_logical::{CounterMaterializerNamer, Materializer, Pipeline}; use pg2iceberg_sim::blob::MemoryBlobStore; @@ -17,7 +18,9 @@ use pg2iceberg_sim::catalog::MemoryCatalog; use pg2iceberg_sim::clock::TestClock; use pg2iceberg_sim::coord::MemoryCoordinator; use pg2iceberg_sim::postgres::{SimPostgres, SimReplicationStream}; -use pg2iceberg_snapshot::{run_snapshot, run_snapshot_chunked, Snapshotter}; +use pg2iceberg_snapshot::{ + run_snapshot, run_snapshot_chunked, Result as SnapResult, SnapshotSource, Snapshotter, +}; use pollster::block_on; use std::collections::BTreeMap; use std::sync::Arc; @@ -527,6 +530,96 @@ fn chunked_snapshot_with_zero_chunk_size_panics() { assert!(result.is_err(), "chunk_size=0 must panic"); } +// ---------- regression: PK-advance guard must not be lexicographic ---------- + +/// A `SnapshotSource` that returns rows ordered by **numeric** PK ASC, +/// mirroring real Postgres (`ORDER BY pk ASC` with a row-value +/// `WHERE (pk) > (cursor)`). `SimPostgres` can't reproduce the bug below +/// because it stores/filters rows by the lexicographic `pk_key` string, +/// so source order and the guard agree; real PG orders numerically. +struct NumericPkSource { + /// Pre-sorted by numeric PK ASC. + rows: Vec, + pk_cols: Vec, + lsn: Lsn, +} + +#[async_trait] +impl SnapshotSource for NumericPkSource { + async fn snapshot_lsn(&self) -> SnapResult { + Ok(self.lsn) + } + + async fn read_chunk( + &self, + _ident: &TableIdent, + chunk_size: usize, + after_pk_key: Option<&str>, + ) -> SnapResult> { + // Resume strictly after the row whose canonical key == cursor, + // preserving numeric order (rows are already numerically sorted). + let start = match after_pk_key { + None => 0, + Some(after) => self + .rows + .iter() + .position(|r| pk_key(r, &self.pk_cols) == after) + .map(|i| i + 1) + .unwrap_or(self.rows.len()), + }; + Ok(self + .rows + .iter() + .skip(start) + .take(chunk_size) + .cloned() + .collect()) + } +} + +#[test] +fn snapshot_advances_across_integer_pk_digit_width_boundary() { + // Regression for the snapshot no-progress guard. It used to compare + // canonical PK keys lexicographically (`new_key <= prev`). With a + // numerically-ordered source, a chunk boundary that crosses a + // digit-width boundary produces keys where, as strings, + // `[{"Int4":10}]` < `[{"Int4":9}]` ('1' < '9'). The guard then + // falsely concluded the snapshot "did not advance past PK 9" and + // aborted — breaking every integer-PK table larger than chunk_size. + // The guard now compares for equality (true no-progress) only. + let mut h = boot(&[]); // PG side unused; the mock drives the snapshot. + + let rows: Vec = (1..=12).map(|i| row(i, i * 10)).collect(); + let src = NumericPkSource { + rows, + pk_cols: vec![col("id")], + lsn: Lsn(1), + }; + + // chunk_size 9 => chunk 1 ends at PK 9, chunk 2 ends at PK 12 — the + // boundary that broke the old lexicographic comparison. + let snap_lsn = block_on(run_snapshot_chunked( + &src, + h.coord.clone() as Arc, + &[schema()], + &mut h.pipeline, + 9, + )) + .expect("snapshot must advance across the 9->10..12 digit-width boundary"); + + h.stream.send_standby(snap_lsn); + block_on(h.materializer.cycle()).unwrap(); + + let iceberg = read_iceberg(&h); + let expected: Vec = (1..=12).map(|i| row(i, i * 10)).collect(); + assert_eq!( + iceberg.len(), + 12, + "every row across the digit-width boundary must materialize" + ); + assert_eq!(iceberg, expected); +} + #[test] fn snapshot_xid_does_not_collide_with_real_pg_xids() { // Run snapshot, then commit many real txs. The synthetic snapshot xid