pg-parity-harness: shared PostgreSQL parity harness (D-QSP-H)

.github/workflows/rust-test.yml

@@ -168,6 +168,12 @@ jobs:
       # the persisted OrderIndex (S3) and its refusals.
       - name: Run dr256-ternlog-lab tests
         run: cargo test --manifest-path crates/dr256-ternlog-lab/Cargo.toml
+      # pg-parity-harness (D-QSP-H): its 13 unit tests are the comparison
+      # rules every parity example relies on (multiset, ties, NULL vs empty,
+      # SQLSTATE, vacuous mutation). The live test prints SKIP here: CI has no
+      # PostgreSQL server.
+      - name: Run pg-parity-harness tests
+        run: cargo test --manifest-path crates/pg-parity-harness/Cargo.toml
       # lance-graph-report: no workflow ran its tests before D-LANESOCK-2. The
       # lane-socket suite is a `#[cfg(test)]` unit module (its constructor
       # takes an unwitnessed reader scope), so the plain run covers it.

Cargo.toml

@@ -13,6 +13,9 @@ members = [
     # `ogar-r2il`). The contract may never depend on OGAR (`ogar-loco` depends on
     # the contract), but members other than the contract may.
     "crates/lance-graph-r2il",
+    # Shared PostgreSQL parity harness (D-QSP-H): drives psql, compares
+    # multisets / ordered sequences / types / SQLSTATEs. No dependencies.
+    "crates/pg-parity-harness",
     # ONNX mathematical compatibility facade over existing ndarray functions
     # (D-ONNX-NATIVE-0). Depends on ndarray only; nothing native depends on it.
     "crates/lance-graph-onnx",

crates/lance-graph-r2il/Cargo.toml

@@ -34,3 +34,6 @@ lance-graph-contract = { path = "../lance-graph-contract" }
 # by Quack's names and checks each answer against Quack's own lowering, run
 # through mask-risc. Quack is zero-OGAR and does not depend on this crate.
 lance-graph-quack = { path = "../lance-graph-quack" }
+# The shared PostgreSQL parity harness (D-QSP-H), for
+# examples/null_aggs_pg_parity.rs. Lab, no dependencies of its own.
+pg-parity-harness = { path = "../pg-parity-harness" }

crates/lance-graph-r2il/examples/null_aggs_pg_parity.rs

@@ -12,8 +12,11 @@
 //! validity plane replaced by every row (the defect #1681 shipped) and must
 //! differ from PostgreSQL somewhere.
 //!
-//! Needs `psql` and a server, connection from the usual `PG*` variables.
-//! Without one it prints `SKIP` and exits 0: a lab instrument, not CI parity.
+//! Needs `psql` and a server; the connection goes through `pg-parity-harness`
+//! and reads the usual `PG*` variables. Without a server it prints `SKIP` and
+//! exits 0: a lab instrument, not CI parity. A failure after a successful
+//! connect (the server rejecting the script, an unparsable answer) is a panic,
+//! not a skip. NULL reaches this example as NULL, never as an empty field.
 //!
 //! ```text
 //! PGHOST=<socket dir> PGPORT=<port> PGUSER=<user> PGDATABASE=postgres \
@@ -23,8 +26,7 @@
 use lance_graph_mask_risc::{words_for, Foreign, LaneRef, Planes};
 use lance_graph_r2il::procs::{bool_and, bool_or, count_distinct_sorted, select_lt, Select};
 use lance_graph_r2il::{FoldDialect, Val};
-use std::io::Write;
-use std::process::{Command, Stdio};
+use pg_parity_harness::{Cell, Pg, Skip};
 
 const ROWS: usize = 120;
 const TABLES: usize = 60;
@@ -130,43 +132,23 @@ fn aggs(t: &Table, sel: i32, vb: u16, vk: u16) -> [Option<i64>; 3] {
     ]
 }
 
-fn pg_value(s: &str) -> Option<i64> {
-    match s {
-        "" => None,
-        "t" => Some(1),
-        "f" => Some(0),
-        n => Some(n.parse().expect("number")),
-    }
-}
-
-fn psql(script: String) -> Result<Vec<u8>, String> {
-    let out = Command::new("psql")
-        .args(["-X", "-q", "-tA", "-F,", "-v", "ON_ERROR_STOP=1", "-f", "-"])
-        .stdin(Stdio::piped())
-        .stdout(Stdio::piped())
-        .stderr(Stdio::piped())
-        .spawn()
-        .and_then(|mut child| {
-            let mut stdin = child.stdin.take().expect("piped stdin");
-            let writer = std::thread::spawn(move || {
-                let _ = stdin.write_all(script.as_bytes());
-            });
-            let out = child.wait_with_output();
-            let _ = writer.join();
-            out
-        })
-        .map_err(|e| format!("could not run psql ({e})"))?;
-    if out.status.success() {
-        Ok(out.stdout)
-    } else {
-        Err(format!(
-            "psql failed: {}",
-            String::from_utf8_lossy(&out.stderr).trim()
-        ))
+fn pg_value(c: &Cell) -> Option<i64> {
+    match c.as_deref() {
+        None => None,
+        Some("t") => Some(1),
+        Some("f") => Some(0),
+        Some(n) => Some(n.parse().expect("number")),
     }
 }
 
 fn main() {
+    let pg = match Pg::connect() {
+        Ok(pg) => pg,
+        Err(Skip(why)) => {
+            println!("SKIP: {why}");
+            return;
+        }
+    };
     let mut rng = Rng(0x00D1_5EED_0011_0001);
     let tables: Vec<Table> = (0..TABLES).map(|_| table(&mut rng)).collect();
 
@@ -198,40 +180,31 @@ fn main() {
             ));
         }
     }
-    let stdout = match psql(script) {
-        Ok(o) => o,
-        Err(why) => {
-            println!("SKIP: {why}");
-            return;
-        }
-    };
-    let lines: Vec<&str> = std::str::from_utf8(&stdout)
-        .expect("utf-8")
-        .lines()
-        .filter(|l| !l.is_empty())
-        .collect();
-    assert_eq!(lines.len(), TABLES * SELECTIONS.len(), "one row per query");
+    let rows = pg
+        .script(&script)
+        .unwrap_or_else(|e| panic!("PostgreSQL rejected the script: {e}"));
+    assert_eq!(rows.len(), TABLES * SELECTIONS.len(), "one row per query");
 
     let (mut nulls, mut trues_with_null, mut naive_wrong) = (0usize, 0usize, 0usize);
-    let mut line = lines.iter();
+    let mut row_iter = rows.iter();
     for t in &tables {
         for sel in SELECTIONS {
-            let cols: Vec<&str> = line.next().expect("row").split(',').collect();
-            let pg = [pg_value(cols[0]), pg_value(cols[1]), pg_value(cols[2])];
+            let cols = row_iter.next().expect("row");
+            assert_eq!(cols.len(), 3);
+            let pg_row = [pg_value(&cols[0]), pg_value(&cols[1]), pg_value(&cols[2])];
             let ours = aggs(t, sel, VB, VK);
-            assert_eq!(ours, pg, "x < {sel}: r2il {ours:?} vs PostgreSQL {pg:?}");
-            nulls += usize::from(pg[0].is_none());
+            assert_eq!(
+                ours, pg_row,
+                "x < {sel}: r2il {ours:?} vs PostgreSQL {pg_row:?}"
+            );
+            nulls += usize::from(pg_row[0].is_none());
             let has_null_selected =
                 (0..ROWS).any(|r| t.x[r] < sel && !bit(&t.planes[VB as usize], r));
-            trues_with_null += usize::from(pg[1] == Some(1) && has_null_selected);
-            naive_wrong += usize::from(aggs(t, sel, ALL, ALL) != pg);
+            trues_with_null += usize::from(pg_row[1] == Some(1) && has_null_selected);
+            naive_wrong += usize::from(aggs(t, sel, ALL, ALL) != pg_row);
         }
     }
-    let version = Command::new("psql")
-        .args(["-X", "-tA", "-c", "SHOW server_version"])
-        .output()
-        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
-        .unwrap_or_default();
+    let version = pg.version();
     let total = TABLES * SELECTIONS.len();
     println!(
         "PostgreSQL {version}: {total} queries ({nulls} with NULL bool aggregates, \

crates/pg-parity-harness/Cargo.toml

@@ -0,0 +1,10 @@
+[package]
+name = "pg-parity-harness"
+version = "0.1.0"
+edition = "2021"
+publish = false
+description = "LAB. Shared PostgreSQL parity harness (D-QSP-H): drives psql against a real server, returns rows with NULL kept distinct, result column types and SQLSTATE, and compares them with an implementation's answer (multiset, ordered-with-ties, types, errors). Prints SKIP without a server. No dependencies."
+
+# A workspace member with no dependencies. Other crates take it as a
+# dev-dependency only; nothing native depends on it.
+[dependencies]