supabase · jmqd · Aug 5, 2026 · Aug 6, 2026 · Aug 8, 2026 · Aug 8, 2026
@@ -236,7 +236,8 @@ pub enum BigQueryTimePartitionGranularity {
 ///
 /// `ReplacingMergeTree` (default) gives current-state reads via `FINAL` and
 /// reclaims deleted rows on `OPTIMIZE ... FINAL CLEANUP`. `MergeTree` is an
-/// append-only event-log layout retained for PK-less source tables.
+/// append-only event-log layout retained for PK-less source tables. It stores
+/// source ordering in `cdc_lsn` and `cdc_tx_ordinal`.
 ///
 /// Applied only when a table is created or recreated. ClickHouse cannot
 /// alter a table's engine, so a mismatch against an existing table is a

@@ -480,9 +480,8 @@ impl ClickHouseClient {
     ///
     /// `after_column` controls placement: `Some(name)` inserts the new column
     /// immediately AFTER `name`, `None` inserts it FIRST (used when the table
-    /// has no user columns yet). Either way the new column lands before the
-    /// trailing CDC columns (`cdc_operation`, `cdc_lsn`), which is required
-    /// because RowBinary encoding is positional.
+    /// has no user columns yet). Either placement keeps it before the trailing
+    /// CDC columns, which RowBinary encoding requires.
     pub(crate) async fn add_column(
         &self,
         table_name: &str,

@@ -26,7 +26,7 @@ pub(crate) enum ClickHouseValue {
     /// tombstone.
     UInt8(u8),
     UInt32(u32),
-    /// Unsigned 64-bit integer, used for the MergeTree `cdc_lsn` column.
+    /// Unsigned 64-bit integer, used for MergeTree ordering metadata.
     UInt64(u64),
     /// Unsigned 128-bit integer, used for the ReplacingMergeTree `_etl_version`
     /// column (the packed `EventSequenceKey`).

@@ -12,6 +12,9 @@ use crate::clickhouse::sql::quote_identifier;
 pub(crate) const CDC_OPERATION_COLUMN_NAME: &str = "cdc_operation";
 /// (For MergeTree engine) CDC LSN column (commit_lsn).
 pub(crate) const CDC_LSN_COLUMN_NAME: &str = "cdc_lsn";
+/// (For MergeTree engine) zero-based source event ordinal within its
+/// transaction.
+pub(crate) const CDC_TX_ORDINAL_COLUMN_NAME: &str = "cdc_tx_ordinal";
 /// (For ReplacingMergeTree engine) version column. Holds the packed
 /// `EventSequenceKey` (commit_lsn in the high 64 bits, tx_ordinal in the
 /// low 64 bits) as a UInt128, giving ReplacingMergeTree a total order across
@@ -208,7 +211,9 @@ fn quote_numeric_literal_as_string(expression: &str) -> String {
 /// Trailing CDC column names appended to each replicated row, by engine.
 pub(super) fn trailing_cdc_column_names(engine: ClickHouseEngine) -> &'static [&'static str] {
     match engine {
-        ClickHouseEngine::MergeTree => &[CDC_OPERATION_COLUMN_NAME, CDC_LSN_COLUMN_NAME],
+        ClickHouseEngine::MergeTree => {
+            &[CDC_OPERATION_COLUMN_NAME, CDC_LSN_COLUMN_NAME, CDC_TX_ORDINAL_COLUMN_NAME]
+        }
         ClickHouseEngine::ReplacingMergeTree => &[ETL_VERSION_COLUMN_NAME, ETL_DELETED_COLUMN_NAME],
     }
 }
@@ -231,15 +236,15 @@ where
     }
 }
 
-/// `MergeTree` DDL: appends `cdc_operation String` and `cdc_lsn UInt64`,
-/// `ORDER BY tuple()`.
+/// `MergeTree` DDL: appends `cdc_operation String`, `cdc_lsn UInt64`, and
+/// `cdc_tx_ordinal UInt64`, then uses `ORDER BY tuple()`.
 pub(super) fn create_merge_tree_sql<'a, I>(table_name: &str, column_schemas: I) -> String
 where
     I: IntoIterator<Item = &'a ColumnSchema>,
     I::IntoIter: ExactSizeIterator,
 {
     let iter = column_schemas.into_iter();
-    let mut cols = Vec::with_capacity(iter.len() + 2);
+    let mut cols = Vec::with_capacity(iter.len() + 3);
 
     for col in iter {
         let col_type = clickhouse_column_type(col, false);
@@ -249,6 +254,7 @@ where
 
     cols.push(format!("  {} String", quote_identifier(CDC_OPERATION_COLUMN_NAME)));
     cols.push(format!("  {} UInt64", quote_identifier(CDC_LSN_COLUMN_NAME)));
+    cols.push(format!("  {} UInt64", quote_identifier(CDC_TX_ORDINAL_COLUMN_NAME)));
 
     let col_defs = cols.join(",\n");
     let quoted_table_name = quote_identifier(table_name);
@@ -548,6 +554,7 @@ mod tests {
         let sql = create_merge_tree_sql("public_t", &schemas);
         assert!(sql.contains("\"cdc_operation\" String"), "cdc_operation should be non-nullable");
         assert!(sql.contains("\"cdc_lsn\" UInt64"), "cdc_lsn should be non-nullable UInt64");
+        assert!(sql.contains("\"cdc_tx_ordinal\" UInt64"));
         assert!(sql.contains("ENGINE = MergeTree()"));
         assert!(sql.contains("ORDER BY tuple()"));
     }
@@ -728,7 +735,7 @@ mod tests {
     fn trailing_cdc_column_names_by_engine() {
         assert_eq!(
             trailing_cdc_column_names(ClickHouseEngine::MergeTree),
-            &[CDC_OPERATION_COLUMN_NAME, CDC_LSN_COLUMN_NAME]
+            &[CDC_OPERATION_COLUMN_NAME, CDC_LSN_COLUMN_NAME, CDC_TX_ORDINAL_COLUMN_NAME,]
         );
         assert_eq!(
             trailing_cdc_column_names(ClickHouseEngine::ReplacingMergeTree),

@@ -196,7 +196,7 @@ impl ClickHouseTestDatabase {
 
     /// Returns the column names and ClickHouse type strings in position order,
     /// excluding both engines' trailing CDC columns (`cdc_operation`,
-    /// `cdc_lsn`, `_etl_version`, `_etl_deleted`).
+    /// `cdc_lsn`, `cdc_tx_ordinal`, `_etl_version`, `_etl_deleted`).
     pub async fn column_types(&self, table_name: &str) -> Vec<(String, String)> {
         #[derive(clickhouse::Row, serde::Deserialize)]
         struct Col {
@@ -206,8 +206,8 @@ impl ClickHouseTestDatabase {
         self.db_client
             .query(
                 "SELECT name, type AS type_name FROM system.columns WHERE database = ? AND table \
-                 = ? AND name NOT IN ('cdc_operation', 'cdc_lsn', '_etl_version', '_etl_deleted') \
-                 ORDER BY position",
+                 = ? AND name NOT IN ('cdc_operation', 'cdc_lsn', 'cdc_tx_ordinal', \
+                 '_etl_version', '_etl_deleted') ORDER BY position",
             )
             .bind(&self.database)
             .bind(table_name)

@@ -357,22 +357,22 @@ async fn updates_are_streamed_to_clickhouse_inner(engine: ClickHouseEngine) {
 
     database
         .run_sql(&format!(
-            "UPDATE {} SET value = 'after' WHERE id = 1",
+            "UPDATE {} SET id = 2, value = 'after' WHERE id = 1",
             table_name.as_quoted_identifier(),
         ))
         .await
         .unwrap();
 
     events_notify.notified().await;
 
+    pipeline.shutdown_and_wait().await.unwrap();
+
     let query = current_state_query(engine, UPDATE_FLOW_TABLE, ID_VALUE_PROJECTION, &["id"], "id");
     let rows: Vec<IdValueRow> = clickhouse_db.query(&query).await;
 
-    pipeline.shutdown_and_wait().await.unwrap();
-
     // --- THEN: current state shows the updated value ---
     assert_eq!(rows.len(), 1, "expected one current-state row after UPDATE");
-    assert_eq!(rows[0].id, 1);
+    assert_eq!(rows[0].id, 2);
     assert_eq!(rows[0].value, "after");
 }
 

@@ -1,6 +1,6 @@
 //! MergeTree-only integration tests. These verify event-log semantics
-//! (`cdc_operation` + `cdc_lsn`) that exist only on the MergeTree engine; the
-//! parameterized spine in `pipeline.rs` covers current-state behavior on
+//! (`cdc_operation` + `cdc_lsn` + `cdc_tx_ordinal`) on the MergeTree engine;
+//! the parameterized spine in `pipeline.rs` covers current-state behavior on
 //! both engines.
 
 use etl::{
@@ -19,23 +19,22 @@ use etl_destinations::clickhouse::test_utils::setup_clickhouse_database;
 use etl_telemetry::tracing::init_test_tracing;
 use rand::random;
 
-use crate::support::crypto::install_crypto_provider;
+use crate::support::{clickhouse::current_state_query, crypto::install_crypto_provider};
 
-/// MergeTree event-log row: includes CDC metadata. All three operations in this
-/// test target the same source row, so `id` is asserted on alongside the
-/// CDC columns.
+/// MergeTree event-log row with source CDC ordering metadata.
 #[derive(clickhouse::Row, serde::Deserialize, Debug)]
 struct EventLogRow {
     id: i64,
     value: String,
     cdc_operation: String,
     cdc_lsn: u64,
+    cdc_tx_ordinal: u64,
 }
 
 const TX_ORDER_SELECT: &str = concat!(
-    "SELECT id, value, cdc_operation, cdc_lsn ",
-    "FROM \"test_tx__order\" ",
-    "ORDER BY id, cdc_lsn",
+    "select id, value, cdc_operation, cdc_lsn, cdc_tx_ordinal ",
+    "from \"test_tx__order\" ",
+    "order by id, cdc_lsn, cdc_tx_ordinal",
 );
 
 /// MergeTree-only: verifies that updates from separately committed transactions
@@ -119,10 +118,10 @@ async fn sequential_transactions_preserve_commit_order_merge_tree() {
 
     events_notify.notified().await;
 
-    let rows: Vec<EventLogRow> = clickhouse_db.query(TX_ORDER_SELECT).await;
-
     pipeline.shutdown_and_wait().await.unwrap();
 
+    let rows: Vec<EventLogRow> = clickhouse_db.query(TX_ORDER_SELECT).await;
+
     // --- THEN: three rows on id=1 with strictly increasing LSNs ---
     assert_eq!(rows.len(), 3, "expected INSERT + two UPDATEs");
 
@@ -141,3 +140,137 @@ async fn sequential_transactions_preserve_commit_order_merge_tree() {
     assert_eq!(rows[2].cdc_operation, "UPDATE");
     assert!(rows[2].cdc_lsn > rows[1].cdc_lsn, "update_b must have a higher LSN than update_a");
 }
+
+/// Current user row projected from MergeTree event history.
+#[derive(clickhouse::Row, serde::Deserialize, Debug)]
+struct CurrentRow {
+    id: i64,
+    value: String,
+}
+
+/// MergeTree preserves same-transaction order when primary-key changes move a
+/// row away from a key and then back to it.
+#[tokio::test(flavor = "multi_thread")]
+async fn same_transaction_primary_key_change_preserves_order_merge_tree() {
+    init_test_tracing();
+    install_crypto_provider();
+
+    // --- GIVEN: a source table with one row ready for initial copy ---
+    let mut database = spawn_source_database().await;
+    let table_name = test_table_name("same_tx_pk_change");
+    let table_id = database
+        .create_table(table_name.clone(), true, &[("value", "text not null")])
+        .await
+        .expect("Failed to create same_tx_pk_change test table");
+
+    let publication_name = "test_pub_clickhouse_same_tx_pk_change";
+    database
+        .create_publication(publication_name, std::slice::from_ref(&table_name))
+        .await
+        .expect("Failed to create same_tx_pk_change publication");
+    database
+        .run_sql(&format!(
+            "insert into {} (value) values ('original')",
+            table_name.as_quoted_identifier(),
+        ))
+        .await
+        .expect("Failed to insert initial same_tx_pk_change row");
+
+    let clickhouse_db = setup_clickhouse_database().await;
+    let store = NotifyingStore::new();
+    let destination = TestDestinationWrapper::wrap(
+        clickhouse_db
+            .build_destination_with_engine(store.clone(), ClickHouseEngine::MergeTree)
+            .await,
+    );
+    let table_sync_complete_notify = store.notify_on_table_sync_complete(table_id).await;
+    let mut pipeline = create_pipeline(
+        &database.config,
+        random::<PipelineId>(),
+        publication_name.to_owned(),
+        store,
+        destination.clone(),
+    );
+
+    pipeline.start().await.unwrap();
+    table_sync_complete_notify.notified().await;
+
+    // --- WHEN: one transaction updates the row and moves its key away and back ---
+    let events_notify = destination
+        .wait_for_events(vec![EventCondition::TableCount(EventType::Update, table_id, 3)])
+        .await;
+    let tx = database.begin_transaction().await;
+    tx.run_sql(&format!(
+        "update {} set value = 'intermediate' where id = 1",
+        table_name.as_quoted_identifier(),
+    ))
+    .await
+    .expect("Failed to update the value");
+    tx.run_sql(&format!(
+        "update {} set id = 2, value = 'moved' where id = 1",
+        table_name.as_quoted_identifier(),
+    ))
+    .await
+    .expect("Failed to update the primary key");
+    tx.run_sql(&format!(
+        "update {} set id = 1, value = 'final' where id = 2",
+        table_name.as_quoted_identifier(),
+    ))
+    .await
+    .expect("Failed to reuse the original primary key");
+    tx.commit_transaction().await;
+
+    events_notify.notified().await;
+    pipeline.shutdown_and_wait().await.unwrap();
+
+    // --- THEN: source order and final current state are preserved ---
+    let event_rows: Vec<EventLogRow> = clickhouse_db
+        .query(
+            "select id, value, cdc_operation, cdc_lsn, cdc_tx_ordinal from \
+             \"test_same__tx__pk__change\" order by cdc_lsn, cdc_tx_ordinal, id, cdc_operation",
+        )
+        .await;
+    let current_rows: Vec<CurrentRow> = clickhouse_db
+        .query(&current_state_query(
+            ClickHouseEngine::MergeTree,
+            "test_same__tx__pk__change",
+            "id, value",
+            &["id"],
+            "id",
+        ))
+        .await;
+
+    assert_eq!(event_rows.len(), 6);
+    assert_eq!(event_rows[0].cdc_operation, "INSERT");
+    assert_eq!(event_rows[0].cdc_lsn, 0);
+    assert_eq!(event_rows[0].cdc_tx_ordinal, 0);
+
+    let streaming_lsn = event_rows[1].cdc_lsn;
+    assert!(event_rows[1..].iter().all(|row| row.cdc_lsn == streaming_lsn));
+
+    assert_eq!(event_rows[1].id, 1);
+    assert_eq!(event_rows[1].value, "intermediate");
+    assert_eq!(event_rows[1].cdc_operation, "UPDATE");
+    assert!(event_rows[1].cdc_tx_ordinal < event_rows[2].cdc_tx_ordinal);
+
+    assert_eq!(event_rows[2].id, 1);
+    assert_eq!(event_rows[2].cdc_operation, "DELETE");
+    assert_eq!(event_rows[2].cdc_tx_ordinal, event_rows[3].cdc_tx_ordinal);
+
+    assert_eq!(event_rows[3].id, 2);
+    assert_eq!(event_rows[3].value, "moved");
+    assert_eq!(event_rows[3].cdc_operation, "UPDATE");
+    assert!(event_rows[3].cdc_tx_ordinal < event_rows[4].cdc_tx_ordinal);
+
+    assert_eq!(event_rows[4].id, 1);
+    assert_eq!(event_rows[4].value, "final");
+    assert_eq!(event_rows[4].cdc_operation, "UPDATE");
+    assert_eq!(event_rows[4].cdc_tx_ordinal, event_rows[5].cdc_tx_ordinal);
+
+    assert_eq!(event_rows[5].id, 2);
+    assert_eq!(event_rows[5].cdc_operation, "DELETE");
+
+    assert_eq!(current_rows.len(), 1);
+    assert_eq!(current_rows[0].id, 1);
+    assert_eq!(current_rows[0].value, "final");
+}
@@ -69,10 +69,10 @@ pub(crate) struct DateBoundariesRow {
 /// engine-specific dedup + tombstone filter and applies the caller's
 /// `ORDER BY` for deterministic test reads.
 ///
-/// MergeTree path: take the latest event per PK with `LIMIT 1 BY`, then drop
-/// any whose latest event is a DELETE. The drop-DELETE filter must come
-/// AFTER the dedup, otherwise a deleted PK whose latest event is a DELETE
-/// would surface its prior INSERT instead of being absent.
+/// MergeTree path: take the latest event per PK with `LIMIT 1 BY`, ordered by
+/// commit LSN and transaction ordinal, then drop any latest DELETE. The
+/// drop-DELETE filter must come after deduplication, or a deleted PK would
+/// surface its prior row.
 ///
 /// ReplacingMergeTree path: `FINAL` + `_etl_deleted = 0`.
 pub(crate) fn current_state_query(
@@ -84,8 +84,9 @@ pub(crate) fn current_state_query(
 ) -> String {
     match engine {
         ClickHouseEngine::MergeTree => format!(
-            "SELECT {projection} FROM (SELECT * FROM \"{table}\" ORDER BY cdc_lsn DESC LIMIT 1 BY \
-             ({pks})) AS current WHERE cdc_operation != 'DELETE' ORDER BY {order_by}",
+            "SELECT {projection} FROM (SELECT * FROM \"{table}\" ORDER BY cdc_lsn DESC, \
+             cdc_tx_ordinal DESC LIMIT 1 BY ({pks})) AS current WHERE cdc_operation != 'DELETE' \
+             ORDER BY {order_by}",
             pks = pk_cols.join(", ")
         ),
         ClickHouseEngine::ReplacingMergeTree => format!(

@@ -239,23 +239,25 @@ Read patterns:
 
 #### MergeTree
 
-Each replicated table is created as `MergeTree() ORDER BY tuple()` with two CDC metadata
-columns appended to every row:
+Each replicated table is created as `MergeTree() ORDER BY tuple()` with three CDC
+metadata columns appended to every row:
 
 - `cdc_operation`: `INSERT`, `UPDATE`, or `DELETE`
 - `cdc_lsn`: the Postgres commit LSN at the time of the change
+- `cdc_tx_ordinal`: the zero-based event position within the Postgres transaction
 
 Read patterns:
 
-- Current state per primary key: take the latest event by `cdc_lsn` with `LIMIT 1 BY`,
-  then filter out tombstones. Example:
+- Current state per primary key: take the latest event by `cdc_lsn` and
+  `cdc_tx_ordinal` with `LIMIT 1 BY`, then filter out tombstones. Example:
 
   ```sql
-  SELECT <user columns> FROM (
-      SELECT * FROM "public_orders"
-      ORDER BY cdc_lsn DESC LIMIT 1 BY (id)
+  select <user columns> from (
+      select * from "public_orders"
+      order by cdc_lsn desc, cdc_tx_ordinal desc
+      limit 1 by (id)
   )
-  WHERE cdc_operation != 'DELETE'
+  where cdc_operation != 'DELETE'
   ```
 
 - Event log queries: read the table directly; every CDC event is preserved.