scripts+migrate: populate denormalized commit_timestamp on the write path

Both query_measurements writers now stamp the denormalized commit_timestamp
(migration 006, the read path's latest-per-series sort key) so new ingests never
depend on the post-deploy re-backfill:

- post-ingest.py --postgres resolves it in the upsert itself via a scalar
  subquery against the commits row the same transaction upserted first, on both
  the INSERT and ON CONFLICT DO UPDATE paths.
- The migrate Rust loader runs the 006 backfill UPDATE after the per-table
  COPYs, inside the same all-or-nothing transaction (the v3 DuckDB source has no
  such column, so it cannot ride the COPY's 1:1 column mapping). The loader's
  schema contract and the e2e container init now include migration 006.

Tests: a new pytest pins stamping on both upsert paths; the Postgres e2e
rehearsal asserts zero NULL and zero drifted commit_timestamp rows after a load.

Signed-off-by: "Connor Tsui" <connor@spiraldb.com>

Author: connortsui20

benchmarks-website/migrate/src/postgres.rs

@@ -223,8 +223,11 @@ impl std::fmt::Display for LoadSummary {
 /// TLS connection (the RDS CA for the prod load); when `None` the connection is
 /// plaintext (`NoTls`, the local rehearsal).
 ///
-/// The target schema (`migrations/001_initial_schema.sql`) must already be
-/// applied: `load` only `COPY`s into the existing tables and never runs DDL.
+/// The target schema must already be applied -- `migrations/001_initial_schema.sql`
+/// plus `006_read_path_perf.sql`, whose denormalized
+/// `query_measurements.commit_timestamp` column the post-COPY denormalization
+/// UPDATE below writes: `load` only `COPY`s into the existing tables and never
+/// runs DDL.
 pub fn load(duckdb_path: &Path, dsn: &str, ca_cert: Option<&Path>) -> Result<LoadSummary> {
     let config = duckdb::Config::default()
         .access_mode(duckdb::AccessMode::ReadOnly)
@@ -250,6 +253,23 @@ pub fn load(duckdb_path: &Path, dsn: &str, ca_cert: Option<&Path>) -> Result<Loa
         per_table.push((spec.name, rows));
     }
 
+    // Populate the denormalized `query_measurements.commit_timestamp` (migration
+    // 006, the read path's latest-per-series sort key). The v3 DuckDB source has
+    // no such column, so it cannot ride the COPY's 1:1 column mapping; it is
+    // derived here from the already-loaded `commits` dim (`TABLE_SPECS` loads
+    // `commits` first), inside the same all-or-nothing transaction.
+    let stamped = tx
+        .execute(
+            "UPDATE query_measurements q
+                SET commit_timestamp = c.timestamp
+               FROM commits c
+              WHERE c.commit_sha = q.commit_sha
+                AND q.commit_timestamp IS NULL",
+            &[],
+        )
+        .context("denormalizing commit_timestamp onto query_measurements")?;
+    info!(rows = stamped, "denormalized commit_timestamp");
+
     tx.commit().context("committing the load transaction")?;
     Ok(LoadSummary { per_table })
 }

benchmarks-website/migrate/tests/postgres_e2e.rs

@@ -31,8 +31,15 @@ use vortex_bench_migrate::verify::run_postgres_value_verify;
 use vortex_bench_server::family;
 use vortex_bench_server::schema::COMMITS_DDL;
 
-/// The authoritative Postgres schema, applied to the container at init.
-const SCHEMA_SQL: &str = include_str!("../../../migrations/001_initial_schema.sql");
+/// The authoritative Postgres schema, applied to the container at init: the base
+/// DDL plus the 006 read-path migration, whose denormalized
+/// `query_measurements.commit_timestamp` column the loader's post-COPY
+/// denormalization UPDATE requires (matching the prod target, where every
+/// migration is applied before a load).
+const SCHEMA_SQL: &str = concat!(
+    include_str!("../../../migrations/001_initial_schema.sql"),
+    include_str!("../../../migrations/006_read_path_perf.sql"),
+);
 
 /// Per-table row counts the fixture loads. Drives the count assertions.
 const FIXTURE_COUNTS: &[(&str, u64)] = &[
@@ -161,6 +168,25 @@ fn rehearsal_load_then_verify_is_clean() -> Result<()> {
             "target count for {table}"
         );
     }
+
+    // The loader denormalized `commit_timestamp` onto every `query_measurements`
+    // row (migration 006, the read path's latest-per-series sort key): no NULLs
+    // remain and each value equals the joined `commits.timestamp`.
+    let unstamped: i64 = client
+        .query_one(
+            "SELECT count(*) FROM query_measurements WHERE commit_timestamp IS NULL",
+            &[],
+        )?
+        .get(0);
+    assert_eq!(unstamped, 0, "rows missing denormalized commit_timestamp");
+    let mismatched: i64 = client
+        .query_one(
+            "SELECT count(*) FROM query_measurements q JOIN commits c USING (commit_sha)
+              WHERE q.commit_timestamp <> c.timestamp",
+            &[],
+        )?
+        .get(0);
+    assert_eq!(mismatched, 0, "denormalized commit_timestamp drifted");
     Ok(())
 }
 

scripts/post-ingest.py

@@ -670,17 +670,19 @@ def _insert_query_measurement(conn, mid_mod, r: dict) -> bool:
             query_idx, storage, engine, format,
             value_ns, all_runtimes_ns,
             peak_physical, peak_virtual, physical_delta, virtual_delta,
-            env_triple
-        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::bigint[], %s, %s, %s, %s, %s)
+            env_triple, commit_timestamp
+        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::bigint[], %s, %s, %s, %s, %s,
+                  (SELECT timestamp FROM commits WHERE commit_sha = %s))
         ON CONFLICT (measurement_id) DO UPDATE SET
-            commit_sha      = excluded.commit_sha,
-            value_ns        = excluded.value_ns,
-            all_runtimes_ns = excluded.all_runtimes_ns,
-            peak_physical   = excluded.peak_physical,
-            peak_virtual    = excluded.peak_virtual,
-            physical_delta  = excluded.physical_delta,
-            virtual_delta   = excluded.virtual_delta,
-            env_triple      = excluded.env_triple
+            commit_sha       = excluded.commit_sha,
+            value_ns         = excluded.value_ns,
+            all_runtimes_ns  = excluded.all_runtimes_ns,
+            peak_physical    = excluded.peak_physical,
+            peak_virtual     = excluded.peak_virtual,
+            physical_delta   = excluded.physical_delta,
+            virtual_delta    = excluded.virtual_delta,
+            env_triple       = excluded.env_triple,
+            commit_timestamp = excluded.commit_timestamp
         RETURNING (xmax = 0) AS inserted
         """,
         (
@@ -700,6 +702,10 @@ def _insert_query_measurement(conn, mid_mod, r: dict) -> bool:
             r.get("physical_delta"),
             r.get("virtual_delta"),
             r.get("env_triple"),
+            # The denormalized `commit_timestamp` (migration 006) is resolved from the
+            # `commits` row this same transaction upserted first, so the read path's
+            # latest-per-series summary never sees a NULL from this writer.
+            r["commit_sha"],
         ),
     )
 

scripts/test_post_ingest_postgres.py

@@ -303,6 +303,29 @@ def test_ingest_inserts_then_updates(schema_conn: psycopg.Connection) -> None:
         assert _count(schema_conn, table) == 1, f"{kind} -> {table} row count drifted"
 
 
+def test_query_measurement_insert_populates_commit_timestamp(schema_conn: psycopg.Connection) -> None:
+    """The denormalized `commit_timestamp` (migration 006, the read path's latest-per-series sort
+    key) is stamped from the envelope's `commits` row on BOTH upsert paths, so rows written by this
+    writer never depend on the post-deploy re-backfill."""
+    commit = _sample_commit()
+    records = [r for r in _sample_records() if r["kind"] == "query_measurement"]
+
+    post_ingest.ingest_postgres(schema_conn, commit, records)
+    row = schema_conn.execute(
+        "SELECT q.commit_timestamp, c.timestamp FROM query_measurements q"
+        " JOIN commits c USING (commit_sha)"
+    ).fetchone()
+    assert row[0] is not None
+    assert row[0] == row[1]
+
+    # The update path re-stamps as well: scrub the column, re-ingest the same envelope (an
+    # ON CONFLICT DO UPDATE), and the timestamp must come back.
+    schema_conn.execute("UPDATE query_measurements SET commit_timestamp = NULL")
+    post_ingest.ingest_postgres(schema_conn, commit, records)
+    restamped = schema_conn.execute("SELECT commit_timestamp FROM query_measurements").fetchone()[0]
+    assert restamped == row[1]
+
+
 # Per-kind mutation of every value/side-counter/env column each table's ON
 # CONFLICT DO UPDATE SET list owns (dim columns deliberately excluded). Mirrors
 # the SET lists in benchmarks-website/server/src/ingest.rs; a stale/incorrect SET