diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index d5c7b0165..9f85d0574 100644 --- a/.github/workflows/rust-test.yml +++ b/.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. diff --git a/Cargo.toml b/Cargo.toml index 9d0018622..f4cc5c517 100644 --- a/Cargo.toml +++ b/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", diff --git a/crates/lance-graph-r2il/Cargo.toml b/crates/lance-graph-r2il/Cargo.toml index f098ec063..b8668cacd 100644 --- a/crates/lance-graph-r2il/Cargo.toml +++ b/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" } diff --git a/crates/lance-graph-r2il/examples/null_aggs_pg_parity.rs b/crates/lance-graph-r2il/examples/null_aggs_pg_parity.rs index bc874a729..c87ddac53 100644 --- a/crates/lance-graph-r2il/examples/null_aggs_pg_parity.rs +++ b/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= PGPORT= PGUSER= 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; 3] { ] } -fn pg_value(s: &str) -> Option { - match s { - "" => None, - "t" => Some(1), - "f" => Some(0), - n => Some(n.parse().expect("number")), - } -} - -fn psql(script: String) -> Result, 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 { + 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 = (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, \ diff --git a/crates/pg-parity-harness/Cargo.toml b/crates/pg-parity-harness/Cargo.toml new file mode 100644 index 000000000..20d7ee088 --- /dev/null +++ b/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] diff --git a/crates/pg-parity-harness/src/compare.rs b/crates/pg-parity-harness/src/compare.rs new file mode 100644 index 000000000..927b1dd4a --- /dev/null +++ b/crates/pg-parity-harness/src/compare.rs @@ -0,0 +1,410 @@ +//! Comparators over PostgreSQL's answer (`expected`) and an implementation's. + +use crate::pg::PgError; +use crate::{Row, Rows}; +use std::collections::BTreeMap; + +/// A typed refusal: the implementation does not support a feature. `feature` +/// is a stable identifier (e.g. "window-function"), `detail` free text. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Refusal { + pub feature: &'static str, + pub detail: String, +} + +/// The implementation's answer to one query. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Answer { + Rows(Rows), + Error { sqlstate: String }, + Unsupported(Refusal), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Order<'a> { + /// No ORDER BY: results are multisets. + Unordered, + /// ORDER BY over these OUTPUT column indices, in key order. + By(&'a [usize]), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Diff { + Types { + expected: Vec, + actual: Vec, + }, + /// Rows whose multiplicity differs, with (expected, actual) counts. + /// At most the first 5 in row order are kept; `total` is the full number. + Multiset { + rows: Vec<(Row, usize, usize)>, + total: usize, + }, + /// First position where the sequences differ beyond an allowed tie + /// reordering (`None` = that side ended). + Ordered { + index: usize, + expected: Option, + actual: Option, + }, + Sqlstate { + expected: String, + actual: String, + }, + /// One side returned rows and the other an error. Values: "rows" / "error". + Kind { + expected: &'static str, + actual: &'static str, + }, + /// A mutation test whose mutation did not change PostgreSQL's answer. + VacuousMutation, + Harness(String), +} + +impl std::fmt::Display for Diff { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Diff::Types { expected, actual } => { + write!( + f, + "column types: PostgreSQL {expected:?}, actual {actual:?}" + ) + } + Diff::Multiset { rows, total } => { + write!(f, "{total} row(s) differ in multiplicity; first:")?; + for (row, e, a) in rows { + write!(f, " {row:?} (PostgreSQL {e}, actual {a});")?; + } + Ok(()) + } + Diff::Ordered { + index, + expected, + actual, + } => write!( + f, + "order differs at position {index}: PostgreSQL {expected:?}, actual {actual:?}" + ), + Diff::Sqlstate { expected, actual } => { + write!(f, "SQLSTATE: PostgreSQL {expected}, actual {actual}") + } + Diff::Kind { expected, actual } => { + write!( + f, + "PostgreSQL returned {expected}, actual returned {actual}" + ) + } + Diff::VacuousMutation => { + write!(f, "the mutation did not change PostgreSQL's answer") + } + Diff::Harness(m) => write!(f, "harness: {m}"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Outcome { + Pass, + Fail(Diff), + Unsupported(Refusal), + Skip(String), +} + +/// `None` when both sides hold the same rows with the same multiplicities, +/// in any order. +pub fn multiset_diff(expected: &[Row], actual: &[Row]) -> Option { + let mut counts: BTreeMap<&Row, (usize, usize)> = BTreeMap::new(); + for row in expected { + let e = counts.entry(row).or_insert((0, 0)); + e.0 += 1; + } + for row in actual { + let e = counts.entry(row).or_insert((0, 0)); + e.1 += 1; + } + let differing: Vec<(Row, usize, usize)> = counts + .into_iter() + .filter(|(_, c)| c.0 != c.1) + .map(|(row, c)| (row.clone(), c.0, c.1)) + .collect(); + if differing.is_empty() { + return None; + } + let total = differing.len(); + Some(Diff::Multiset { + rows: differing.into_iter().take(5).collect(), + total, + }) +} + +/// Order check for an ORDER BY over output columns `keys`: rows whose key +/// cells are equal form a tie, and a tie may appear in any internal order. +/// +/// Keys compare as text, so two key values PostgreSQL treats as equal but +/// prints differently are treated as distinct; that can only make the check +/// stricter, never looser. An ORDER BY expression that is not an output +/// column must be projected by the caller. A key index past a row's end is a +/// caller bug and panics. +pub fn ordered_diff(expected: &[Row], actual: &[Row], keys: &[usize]) -> Option { + let same_key = |a: &Row, b: &Row| keys.iter().all(|&k| a[k] == b[k]); + let mut i = 0; + while i < expected.len() { + let mut j = i + 1; + while j < expected.len() && same_key(&expected[i], &expected[j]) { + j += 1; + } + if actual.len() < j { + let index = actual.len().max(i); + return Some(Diff::Ordered { + index, + expected: expected.get(index).cloned(), + actual: None, + }); + } + let mut e = expected[i..j].to_vec(); + let mut a = actual[i..j].to_vec(); + e.sort(); + a.sort(); + if e != a { + let k = (i..j).find(|&k| actual[k] != expected[k]).unwrap_or(i); + return Some(Diff::Ordered { + index: k, + expected: Some(expected[k].clone()), + actual: Some(actual[k].clone()), + }); + } + i = j; + } + if actual.len() > expected.len() { + return Some(Diff::Ordered { + index: expected.len(), + expected: None, + actual: Some(actual[expected.len()].clone()), + }); + } + None +} + +/// `None` when equal, else `Diff::Types`. +pub fn types_diff(expected: &[String], actual: &[String]) -> Option { + if expected == actual { + None + } else { + Some(Diff::Types { + expected: expected.to_vec(), + actual: actual.to_vec(), + }) + } +} + +/// Compares PostgreSQL's result with the implementation's answer. +pub fn compare(pg: &Result, ours: &Answer, order: Order<'_>) -> Outcome { + if let Answer::Unsupported(r) = ours { + return Outcome::Unsupported(r.clone()); + } + match (pg, ours) { + (Err(PgError::Harness(m)), _) => Outcome::Fail(Diff::Harness(m.clone())), + (Err(PgError::Sql { sqlstate: e, .. }), Answer::Error { sqlstate: a }) => { + if e == a { + Outcome::Pass + } else { + Outcome::Fail(Diff::Sqlstate { + expected: e.clone(), + actual: a.clone(), + }) + } + } + (Err(PgError::Sql { .. }), Answer::Rows(_)) => Outcome::Fail(Diff::Kind { + expected: "error", + actual: "rows", + }), + (Ok(_), Answer::Error { .. }) => Outcome::Fail(Diff::Kind { + expected: "rows", + actual: "error", + }), + (Ok(e), Answer::Rows(a)) => { + if let Some(d) = types_diff(&e.types, &a.types) { + return Outcome::Fail(d); + } + let rows = match order { + Order::Unordered => multiset_diff(&e.rows, &a.rows), + Order::By(keys) => ordered_diff(&e.rows, &a.rows, keys), + }; + match rows { + Some(d) => Outcome::Fail(d), + None => Outcome::Pass, + } + } + (_, Answer::Unsupported(r)) => Outcome::Unsupported(r.clone()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn r(cells: &[Option<&str>]) -> Row { + cells.iter().map(|c| c.map(str::to_string)).collect() + } + + fn row2(a: &str, b: &str) -> Row { + r(&[Some(a), Some(b)]) + } + + #[test] + fn multiset_counts_duplicates() { + let a = row2("1", "a"); + let b = row2("2", "b"); + let expected = [a.clone(), a.clone(), b.clone()]; + let actual = [a.clone(), b.clone(), b.clone()]; + match multiset_diff(&expected, &actual) { + Some(Diff::Multiset { rows, total }) => { + assert_eq!(total, 2); + assert_eq!(rows, vec![(a, 2, 1), (b, 1, 2)]); + } + other => panic!("expected a multiset diff, got {other:?}"), + } + } + + #[test] + fn multiset_ignores_order() { + let a = row2("1", "a"); + let b = row2("2", "b"); + let c = row2("3", "c"); + assert_eq!( + multiset_diff( + &[a.clone(), a.clone(), b.clone(), c.clone()], + &[c.clone(), a.clone(), b.clone(), a.clone()] + ), + None + ); + assert_eq!( + multiset_diff( + &[a.clone(), a.clone(), b.clone()], + &[a.clone(), a.clone(), b.clone()] + ), + None + ); + } + + #[test] + fn ordered_allows_reorder_inside_a_tie() { + let expected = [row2("1", "x"), row2("1", "y"), row2("2", "z")]; + let actual = [row2("1", "y"), row2("1", "x"), row2("2", "z")]; + assert_eq!(ordered_diff(&expected, &actual, &[0]), None); + } + + #[test] + fn ordered_rejects_reorder_across_keys() { + let expected = [row2("1", "x"), row2("2", "y"), row2("2", "z")]; + let swapped = [row2("2", "y"), row2("1", "x"), row2("2", "z")]; + assert!(matches!( + ordered_diff(&expected, &swapped, &[0]), + Some(Diff::Ordered { index: 0, .. }) + )); + let short = [row2("1", "x"), row2("2", "y")]; + assert!(matches!( + ordered_diff(&expected, &short, &[0]), + Some(Diff::Ordered { actual: None, .. }) + )); + let long = [ + row2("1", "x"), + row2("2", "y"), + row2("2", "z"), + row2("3", "w"), + ]; + assert!(matches!( + ordered_diff(&expected, &long, &[0]), + Some(Diff::Ordered { + index: 3, + expected: None, + actual: Some(_) + }) + )); + } + + #[test] + fn null_is_not_the_empty_string() { + let null = [r(&[None])]; + let empty = [r(&[Some("")])]; + assert!(multiset_diff(&null, &empty).is_some()); + assert!(ordered_diff(&null, &empty, &[0]).is_some()); + } + + fn rows(types: &[&str], rows: Vec) -> Rows { + Rows { + types: types.iter().map(|t| t.to_string()).collect(), + rows, + } + } + + #[test] + fn compare_orders_its_checks() { + let data = vec![row2("1", "a"), row2("2", "b")]; + let pg_ok = Ok(rows(&["integer", "text"], data.clone())); + + // (a) types differ, rows equal + let ours = Answer::Rows(rows(&["bigint", "text"], data.clone())); + assert!(matches!( + compare(&pg_ok, &ours, Order::Unordered), + Outcome::Fail(Diff::Types { .. }) + )); + + // (b) error codes + let pg_err = Err(PgError::Sql { + sqlstate: "22012".into(), + stderr: String::new(), + }); + let same = Answer::Error { + sqlstate: "22012".into(), + }; + let other = Answer::Error { + sqlstate: "22003".into(), + }; + assert_eq!(compare(&pg_err, &same, Order::Unordered), Outcome::Pass); + assert!(matches!( + compare(&pg_err, &other, Order::Unordered), + Outcome::Fail(Diff::Sqlstate { .. }) + )); + + // (c) kind mismatch both ways + let ours_rows = Answer::Rows(rows(&["integer", "text"], data.clone())); + assert_eq!( + compare(&pg_err, &ours_rows, Order::Unordered), + Outcome::Fail(Diff::Kind { + expected: "error", + actual: "rows" + }) + ); + assert_eq!( + compare(&pg_ok, &same, Order::Unordered), + Outcome::Fail(Diff::Kind { + expected: "rows", + actual: "error" + }) + ); + + // (d) a refusal is recorded, even when PostgreSQL also failed + let refusal = Refusal { + feature: "window-function", + detail: "not yet".into(), + }; + let unsupported = Answer::Unsupported(refusal); + for pg in [&pg_ok, &pg_err] { + match compare(pg, &unsupported, Order::Unordered) { + Outcome::Unsupported(r) => assert_eq!(r.feature, "window-function"), + other => panic!("expected Unsupported, got {other:?}"), + } + } + + // (e) a harness failure + let pg_harness = Err(PgError::Harness("boom".into())); + assert_eq!( + compare(&pg_harness, &ours_rows, Order::Unordered), + Outcome::Fail(Diff::Harness("boom".into())) + ); + + // (f) equal rows + assert_eq!(compare(&pg_ok, &ours_rows, Order::Unordered), Outcome::Pass); + } +} diff --git a/crates/pg-parity-harness/src/lib.rs b/crates/pg-parity-harness/src/lib.rs new file mode 100644 index 000000000..949d4fd38 --- /dev/null +++ b/crates/pg-parity-harness/src/lib.rs @@ -0,0 +1,73 @@ +//! D-QSP-H: the shared PostgreSQL parity harness. +//! +//! A lab instrument. It drives `psql` against a real PostgreSQL server and +//! hands back what PostgreSQL printed: the rows, with SQL NULL kept distinct +//! from the empty string, the result column types as `\gdesc` names them, and +//! the SQLSTATE of a rejected statement. It then compares that with an +//! implementation's answer: as a multiset when the query has no ORDER BY, in +//! order (a tie may reorder) when it has one, plus column types and error +//! codes. Without a server `Pg::connect` returns a `Skip` and callers print +//! `SKIP: ...` and return. +//! +//! The harness holds **final result sets only**: PostgreSQL's printed answer +//! and the implementation's printed answer, as text cells. That is an +//! external reference's output and the compared result, bounded by the result +//! size. It is not an execution path and answers no count, exists or sum +//! itself. It reports **no timings**. +//! +//! # Wire format +//! +//! `psql` runs with `-X -q -At -F -0 -P null=`: +//! fields are separated by U+001F, records are NUL-terminated, and a NULL +//! prints as the marker `\u{1}NULL\u{1}`. +//! +//! Fixture constraint: text values must not contain U+001F and must not equal +//! the NULL marker. NUL cannot occur in PostgreSQL text. A U+001F inside a +//! value shows up as a row whose cell count differs from the `\gdesc` column +//! count and is reported as `PgError::Harness`. +//! +//! # Values +//! +//! Values are compared as PostgreSQL text. The caller renders implementation +//! values in PostgreSQL's output form (`t` / `f` for booleans, decimal +//! integers, ...). No float helper is provided: a caller that cannot render a +//! type returns `Answer::Unsupported`. +//! +//! # Connection +//! +//! The connection comes from the usual `PG*` variables (`PGHOST`, `PGPORT`, +//! `PGUSER`, `PGDATABASE`, ...). `PG_PARITY_PSQL` names a different `psql` +//! program. Each `Pg` works in its own scratch schema, dropped on drop. + +mod compare; +mod pg; +mod report; + +pub use compare::{ + compare, multiset_diff, ordered_diff, types_diff, Answer, Diff, Order, Outcome, Refusal, +}; +pub use pg::{parse_rows, parse_sqlstate, Pg, PgError, Skip}; +pub use report::{check, check_mutation, mutation_outcomes, Case, Report}; + +/// One result cell as PostgreSQL prints it. `None` is SQL NULL; `Some("")` is +/// the empty string. The two are never merged. +pub type Cell = Option; +pub type Row = Vec; + +/// A result: column types as `\gdesc` names them (`integer`, `bigint`, +/// `text`, `boolean`, `double precision`, `numeric`, ...) and the rows. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Rows { + pub types: Vec, + pub rows: Vec, +} + +/// `Some(v.to_string())`. +pub fn text(v: impl ToString) -> Cell { + Some(v.to_string()) +} + +/// PostgreSQL's boolean output: `Some("t")` / `Some("f")`. +pub fn boolean(b: bool) -> Cell { + Some(if b { "t" } else { "f" }.to_string()) +} diff --git a/crates/pg-parity-harness/src/pg.rs b/crates/pg-parity-harness/src/pg.rs new file mode 100644 index 000000000..813ddd0b1 --- /dev/null +++ b/crates/pg-parity-harness/src/pg.rs @@ -0,0 +1,280 @@ +//! The `psql` runner and the two pure parsers. + +use crate::{Row, Rows}; +use std::ffi::OsString; +use std::io::Write; +use std::process::{Command, Stdio}; + +const FIELD_SEP: char = '\u{1f}'; // psql -F +const NULL_MARK: &str = "\u{1}NULL\u{1}"; // psql -P null= + // records are NUL-terminated: psql -0 + +/// No usable server. Callers print `SKIP: {0}` and return. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Skip(pub String); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PgError { + /// The server rejected a statement. `sqlstate` is the five-character code. + Sql { sqlstate: String, stderr: String }, + /// psql failed without a SQLSTATE, or its output could not be parsed. + /// Never a skip: the server was reachable at `connect`. + Harness(String), +} + +impl std::fmt::Display for PgError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + PgError::Sql { sqlstate, stderr } => { + write!(f, "SQLSTATE {sqlstate}: {}", stderr.trim()) + } + PgError::Harness(m) => write!(f, "harness: {m}"), + } + } +} + +pub struct Pg { + program: OsString, + schema: String, + version: String, +} + +impl Pg { + /// Checks that psql runs and a server answers, then creates a scratch + /// schema `pgh__`. Any failure here is a `Skip` with the + /// reason (psql missing, connection refused, no CREATE privilege). + pub fn connect() -> Result { + let program = std::env::var_os("PG_PARITY_PSQL").unwrap_or_else(|| OsString::from("psql")); + let out = Command::new(&program) + .args(["-X", "-q", "-At", "-c", "SHOW server_version"]) + .output() + .map_err(|e| Skip(format!("could not run psql ({e})")))?; + if !out.status.success() { + return Err(Skip(format!( + "psql failed: {}", + String::from_utf8_lossy(&out.stderr).trim() + ))); + } + let version = String::from_utf8_lossy(&out.stdout).trim().to_string(); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let pg = Pg { + program, + schema: format!("pgh_{}_{}", std::process::id(), nanos), + version, + }; + pg.run(format!("CREATE SCHEMA {};\n", pg.schema)) + .map_err(|e| Skip(e.to_string()))?; + Ok(pg) + } + + /// `SHOW server_version`, read once at connect. + pub fn version(&self) -> &str { + &self.version + } + + pub fn schema(&self) -> &str { + &self.schema + } + + /// Runs a whole script in ONE psql session (setup, mutation, or a batch + /// of queries; `CREATE TEMP TABLE` lives for this call only). Returns the + /// rows of every statement, concatenated in order. + pub fn script(&self, sql: &str) -> Result, PgError> { + let out = self.run(sql.to_string())?; + parse_rows(&out).map_err(PgError::Harness) + } + + /// Runs one query. Column types come from `\gdesc` (first session), rows + /// from executing it (second session). A trailing `;` and whitespace on + /// `sql` are trimmed. Tables it reads must be non-temporary (created via + /// `script`; they land in the scratch schema). + pub fn query(&self, sql: &str) -> Result { + let q = sql.trim_end(); + let q = q.strip_suffix(';').unwrap_or(q).trim_end(); + let described = + parse_rows(&self.run(format!("{q}\n\\gdesc\n"))?).map_err(PgError::Harness)?; + let mut types = Vec::with_capacity(described.len()); + for row in described { + match <[Option; 2]>::try_from(row) { + Ok([_name, ty]) => types.push(ty.unwrap_or_default()), + Err(row) => { + return Err(PgError::Harness(format!( + "\\gdesc row has {} cells, expected 2", + row.len() + ))) + } + } + } + let rows = parse_rows(&self.run(format!("{q};\n"))?).map_err(PgError::Harness)?; + for row in &rows { + if row.len() != types.len() { + return Err(PgError::Harness(format!( + "row has {} cells, \\gdesc reported {} columns", + row.len(), + types.len() + ))); + } + } + Ok(Rows { types, rows }) + } + + /// The one place psql runs; shared by `script`, `query` and `Drop`. + fn run(&self, script: String) -> Result, PgError> { + let mut options = std::env::var("PGOPTIONS").unwrap_or_default(); + options.push_str(&format!( + " -c search_path={} -c client_min_messages=warning", + self.schema + )); + let mut child = Command::new(&self.program) + .args(["-X", "-q", "-At", "-F"]) + .arg(FIELD_SEP.to_string()) + .args(["-0", "-P"]) + .arg(format!("null={NULL_MARK}")) + .args([ + "-v", + "ON_ERROR_STOP=1", + "-v", + "VERBOSITY=sqlstate", + "-f", + "-", + ]) + .env("LC_MESSAGES", "C") + .env("PGCLIENTENCODING", "UTF8") + .env("PGOPTIONS", options) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|e| PgError::Harness(format!("could not run psql ({e})")))?; + let mut stdin = child.stdin.take().expect("piped stdin"); + // psql answers while it reads. Writing the whole script before reading + // the output deadlocks once the stdout pipe fills, so the script goes + // in on its own thread while this one drains the output. + let writer = std::thread::spawn(move || { + let _ = stdin.write_all(script.as_bytes()); + }); + let out = child.wait_with_output(); + let _ = writer.join(); + let out = out.map_err(|e| PgError::Harness(format!("could not run psql ({e})")))?; + if out.status.success() { + return Ok(out.stdout); + } + let stderr = String::from_utf8_lossy(&out.stderr).to_string(); + match parse_sqlstate(&stderr) { + Some(sqlstate) => Err(PgError::Sql { sqlstate, stderr }), + None => Err(PgError::Harness(format!("psql failed: {}", stderr.trim()))), + } + } +} + +impl Drop for Pg { + fn drop(&mut self) { + let _ = self.run(format!("DROP SCHEMA IF EXISTS {} CASCADE;\n", self.schema)); + } +} + +/// Parses psql output produced with the flags `run` uses. Pure. +pub fn parse_rows(stdout: &[u8]) -> Result, String> { + if stdout.is_empty() { + return Ok(vec![]); + } + if stdout[stdout.len() - 1] != 0 { + return Err("unterminated record".to_string()); + } + let mut records: Vec<&[u8]> = stdout.split(|b| *b == 0).collect(); + records.pop(); // the empty piece after the final terminator + let mut rows = Vec::with_capacity(records.len()); + for rec in records { + let rec = std::str::from_utf8(rec).map_err(|e| format!("record is not UTF-8 ({e})"))?; + rows.push( + rec.split(FIELD_SEP) + .map(|f| { + if f == NULL_MARK { + None + } else { + Some(f.to_string()) + } + }) + .collect(), + ); + } + Ok(rows) +} + +/// First SQLSTATE in psql's stderr under `VERBOSITY=sqlstate`. Pure. +pub fn parse_sqlstate(stderr: &str) -> Option { + for line in stderr.lines() { + let found = ["ERROR:", "FATAL:", "PANIC:"] + .iter() + .filter_map(|tag| line.find(tag).map(|p| p + tag.len())) + .min(); + let Some(start) = found else { continue }; + let rest = line[start..].trim_start_matches(' '); + let run = rest + .bytes() + .take_while(|b| b.is_ascii_uppercase() || b.is_ascii_digit()) + .count(); + let after = rest[run..].chars().next(); + if run == 5 && matches!(after, None | Some(':')) { + return Some(rest[..run].to_string()); + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + fn r(cells: &[Option<&str>]) -> Row { + cells.iter().map(|c| c.map(str::to_string)).collect() + } + + #[test] + fn parse_rows_keeps_null_distinct() { + let input = "1\u{1f}\u{1}NULL\u{1}\u{0}2\u{1f}\u{0}3\u{1f}a|b\nc\0".as_bytes(); + assert_eq!( + parse_rows(input).unwrap(), + vec![ + r(&[Some("1"), None]), + r(&[Some("2"), Some("")]), + r(&[Some("3"), Some("a|b\nc")]), + ] + ); + assert_eq!(parse_rows(b"").unwrap(), Vec::::new()); + assert_eq!(parse_rows(b"\0").unwrap(), vec![r(&[Some("")])]); + } + + #[test] + fn parse_rows_rejects_an_unterminated_record() { + assert!(parse_rows("1\u{1f}2".as_bytes()).is_err()); + assert_eq!( + parse_rows("1\u{1f}2\0".as_bytes()).unwrap(), + vec![r(&[Some("1"), Some("2")])] + ); + } + + #[test] + fn parse_sqlstate_reads_the_code() { + assert_eq!( + parse_sqlstate("psql::1: ERROR: 22012\n"), + Some("22012".to_string()) + ); + assert_eq!( + parse_sqlstate("NOTICE: x\npsql::3: ERROR: 42P01\n"), + Some("42P01".to_string()) + ); + assert_eq!( + parse_sqlstate( + "psql: error: connection to server on socket \"/x/.s.PGSQL.5\" failed: No such file or directory\n" + ), + None + ); + assert_eq!(parse_sqlstate("ERROR: division by zero\n"), None); + assert_eq!(parse_sqlstate("FATAL: 57P01\n"), Some("57P01".to_string())); + assert_eq!(parse_sqlstate("ERROR: XX0001\n"), None); + } +} diff --git a/crates/pg-parity-harness/src/report.rs b/crates/pg-parity-harness/src/report.rs new file mode 100644 index 000000000..c03cefc30 --- /dev/null +++ b/crates/pg-parity-harness/src/report.rs @@ -0,0 +1,256 @@ +//! Recording outcomes, the `check` driver and the mutation test. + +use crate::compare::{compare, Answer, Diff, Order, Outcome, Refusal}; +use crate::pg::{Pg, PgError}; +use crate::Rows; + +pub struct Case<'a> { + pub name: &'a str, + pub sql: &'a str, + pub order: Order<'a>, +} + +#[derive(Debug, Default)] +pub struct Report { + entries: Vec<(String, Outcome)>, +} + +impl Report { + pub fn new() -> Report { + Report::default() + } + + pub fn record(&mut self, name: &str, outcome: Outcome) { + self.entries.push((name.to_string(), outcome)); + } + + pub fn outcomes(&self) -> &[(String, Outcome)] { + &self.entries + } + + /// (pass, fail, unsupported, skip) + pub fn counts(&self) -> (usize, usize, usize, usize) { + let mut c = (0, 0, 0, 0); + for (_, o) in &self.entries { + match o { + Outcome::Pass => c.0 += 1, + Outcome::Fail(_) => c.1 += 1, + Outcome::Unsupported(_) => c.2 += 1, + Outcome::Skip(_) => c.3 += 1, + } + } + c + } + + /// Every recorded refusal, in order: (case name, refusal). + pub fn unsupported(&self) -> Vec<(&str, &Refusal)> { + self.entries + .iter() + .filter_map(|(n, o)| match o { + Outcome::Unsupported(r) => Some((n.as_str(), r)), + _ => None, + }) + .collect() + } + + /// One line: `pg-parity: N cases, P pass, F fail, U unsupported, S skip`. + pub fn summary(&self) -> String { + let (p, f, u, s) = self.counts(); + format!( + "pg-parity: {} cases, {p} pass, {f} fail, {u} unsupported, {s} skip", + self.entries.len() + ) + } + + /// `summary()` first, then one line per Fail (`FAIL : `) and + /// one per Unsupported (`UNSUPPORTED : : `). + pub fn render(&self) -> String { + let mut out = self.summary(); + for (name, o) in &self.entries { + match o { + Outcome::Fail(d) => out.push_str(&format!("\nFAIL {name}: {d}")), + Outcome::Unsupported(r) => out.push_str(&format!( + "\nUNSUPPORTED {name}: {}: {}", + r.feature, r.detail + )), + _ => {} + } + } + out + } +} + +/// Runs `case.sql` on PostgreSQL, compares with `ours()`, records and returns +/// the outcome. +pub fn check( + pg: &Pg, + report: &mut Report, + case: &Case<'_>, + ours: impl FnOnce() -> Answer, +) -> Outcome { + let expected = pg.query(case.sql); + let outcome = compare(&expected, &ours(), case.order); + report.record(case.name, outcome.clone()); + outcome +} + +/// Pure core of the mutation test. Returns (before, after). +pub fn mutation_outcomes( + pg_before: &Result, + ours_before: &Answer, + pg_after: &Result, + ours_after: &Answer, + order: Order<'_>, +) -> (Outcome, Outcome) { + let before = compare(pg_before, ours_before, order); + // "Unchanged" uses the same semantics as the parity comparison: a row + // reorder that the query leaves unspecified is not a change. + let unchanged = match (pg_before, pg_after) { + (Ok(b), Ok(a)) => compare(&Ok(b.clone()), &Answer::Rows(a.clone()), order) == Outcome::Pass, + (b, a) => b == a, + }; + let after = if unchanged { + Outcome::Fail(Diff::VacuousMutation) + } else { + compare(pg_after, ours_after, order) + }; + (before, after) +} + +/// Mutation test: answer, mutate both sides, answer again. +/// `answer` reads the implementation state; `mutate` changes it. Records +/// ` [before]` and ` [after]`; returns (before, after). +/// +/// A value change must show the new values through an unchanged mapping, and +/// a key change must be visible after the caller's `mutate` invalidates or +/// recomputes its preparation. A stale preparation shows up as `after = Fail`. +pub fn check_mutation( + pg: &Pg, + report: &mut Report, + case: &Case<'_>, + pg_mutation: &str, + state: &mut S, + answer: impl Fn(&S) -> Answer, + mutate: impl FnOnce(&mut S), +) -> (Outcome, Outcome) { + let pg_before = pg.query(case.sql); + let ours_before = answer(state); + let mutation = pg.script(pg_mutation); + mutate(state); + let (before, after) = match mutation { + Err(e) => ( + compare(&pg_before, &ours_before, case.order), + Outcome::Fail(Diff::Harness(format!("mutation failed: {e}"))), + ), + Ok(_) => { + let pg_after = pg.query(case.sql); + let ours_after = answer(state); + mutation_outcomes(&pg_before, &ours_before, &pg_after, &ours_after, case.order) + } + }; + report.record(&format!("{} [before]", case.name), before.clone()); + report.record(&format!("{} [after]", case.name), after.clone()); + (before, after) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::Row; + + fn row2(a: &str, b: &str) -> Row { + vec![Some(a.to_string()), Some(b.to_string())] + } + + fn rows(data: Vec) -> Rows { + Rows { + types: vec!["integer".into(), "text".into()], + rows: data, + } + } + + #[test] + fn mutation_guard_fires_on_a_no_op() { + let data = vec![row2("1", "a"), row2("2", "b")]; + let pg = Ok(rows(data)); + let ours = Answer::Rows(pg.clone().unwrap()); + let (before, after) = mutation_outcomes(&pg, &ours, &pg, &ours, Order::Unordered); + assert_eq!(before, Outcome::Pass); + assert_eq!(after, Outcome::Fail(Diff::VacuousMutation)); + } + + #[test] + fn mutation_guard_fires_when_postgresql_only_reorders_rows() { + let pg_before = Ok(rows(vec![row2("1", "a"), row2("2", "b")])); + let pg_after = Ok(rows(vec![row2("2", "b"), row2("1", "a")])); + let ours = Answer::Rows(pg_before.clone().unwrap()); + let (before, after) = + mutation_outcomes(&pg_before, &ours, &pg_after, &ours, Order::Unordered); + assert_eq!(before, Outcome::Pass); + assert_eq!(after, Outcome::Fail(Diff::VacuousMutation)); + } + + #[test] + fn mutation_guard_stays_silent_on_a_real_change() { + let before_rows = rows(vec![row2("1", "a"), row2("2", "b")]); + let after_rows = rows(vec![row2("1", "a"), row2("2", "c")]); + let pg_before = Ok(before_rows.clone()); + let pg_after = Ok(after_rows.clone()); + let ours_before = Answer::Rows(before_rows.clone()); + let ours_after = Answer::Rows(after_rows); + assert_eq!( + mutation_outcomes( + &pg_before, + &ours_before, + &pg_after, + &ours_after, + Order::Unordered + ), + (Outcome::Pass, Outcome::Pass) + ); + // A stale preparation: the implementation still answers the old rows. + let stale = Answer::Rows(before_rows); + let (b, a) = mutation_outcomes( + &pg_before, + &ours_before, + &pg_after, + &stale, + Order::Unordered, + ); + assert_eq!(b, Outcome::Pass); + assert!(matches!(a, Outcome::Fail(Diff::Multiset { .. }))); + } + + #[test] + fn report_counts_every_outcome() { + let mut report = Report::new(); + report.record("p1", Outcome::Pass); + report.record("f1", Outcome::Fail(Diff::VacuousMutation)); + report.record( + "u1", + Outcome::Unsupported(Refusal { + feature: "window-function", + detail: "not yet".into(), + }), + ); + report.record("s1", Outcome::Skip("no server".into())); + report.record("p2", Outcome::Pass); + assert_eq!(report.counts(), (2, 1, 1, 1)); + assert_eq!( + report.summary(), + "pg-parity: 5 cases, 2 pass, 1 fail, 1 unsupported, 1 skip" + ); + let un = report.unsupported(); + assert_eq!(un.len(), 1); + assert_eq!(un[0].0, "u1"); + assert_eq!(un[0].1.feature, "window-function"); + let text = report.render(); + assert_eq!(text.lines().filter(|l| l.starts_with("FAIL ")).count(), 1); + assert_eq!( + text.lines() + .filter(|l| l.starts_with("UNSUPPORTED ")) + .count(), + 1 + ); + } +} diff --git a/crates/pg-parity-harness/tests/pg_live.rs b/crates/pg-parity-harness/tests/pg_live.rs new file mode 100644 index 000000000..b4b375281 --- /dev/null +++ b/crates/pg-parity-harness/tests/pg_live.rs @@ -0,0 +1,191 @@ +//! The one live-server test. Without a server it prints `SKIP` and passes. + +use pg_parity_harness::{ + check, check_mutation, multiset_diff, text, Answer, Case, Diff, Order, Outcome, Pg, PgError, + Refusal, Report, Row, Rows, Skip, +}; + +fn row(k: Option<&str>, v: Option<&str>) -> Row { + vec![k.map(str::to_string), v.map(str::to_string)] +} + +fn int_text(rows: Vec) -> Answer { + Answer::Rows(Rows { + types: vec!["integer".into(), "text".into()], + rows, + }) +} + +fn sqlstate_of(pg: &Pg, sql: &str) -> String { + match pg.query(sql) { + Err(PgError::Sql { sqlstate, .. }) => sqlstate, + other => panic!("{sql}: expected an SQL error, got {other:?}"), + } +} + +#[test] +fn live_server_roundtrip() { + let pg = match Pg::connect() { + Ok(pg) => pg, + Err(Skip(why)) => { + println!("SKIP: {why}"); + return; + } + }; + let version = pg.version().to_string(); + let mut report = Report::new(); + + // 1. setup prints nothing + let setup = pg + .script( + "CREATE TABLE t (k int, v text); \ + INSERT INTO t VALUES (1,'a'),(1,'a'),(2,NULL),(2,''),(3,E'x|y\\nz');", + ) + .expect("setup"); + assert_eq!(setup.len(), 0); + + // 2. NULL, empty string and delimiter characters survive the wire + let all = pg.query("SELECT k, v FROM t").expect("select"); + assert_eq!(all.types, ["integer", "text"]); + assert_eq!(all.rows.len(), 5); + let expected = vec![ + row(Some("1"), Some("a")), + row(Some("1"), Some("a")), + row(Some("2"), None), + row(Some("2"), Some("")), + row(Some("3"), Some("x|y\nz")), + ]; + assert_eq!(multiset_diff(&expected, &all.rows), None); + assert!(all.rows.contains(&row(Some("2"), None))); + assert!(all.rows.contains(&row(Some("2"), Some("")))); + assert!(all.rows.contains(&row(Some("3"), Some("x|y\nz")))); + + // 3. an empty result keeps its column types; trailing `;` is trimmed + let none = pg.query("SELECT k, v FROM t WHERE k > 99;").expect("empty"); + assert_eq!(none.rows.len(), 0); + assert_eq!(none.types.len(), 2); + + // 4. aggregates over no rows: count 0, sum NULL + let agg = pg + .query("SELECT count(*), sum(k) FROM t WHERE false") + .expect("agg"); + assert_eq!(agg.types, ["bigint", "bigint"]); + assert_eq!(agg.rows, vec![vec![text(0), None]]); + + // 5. SQLSTATE + assert_eq!(sqlstate_of(&pg, "SELECT 1/0"), "22012"); + assert_eq!(sqlstate_of(&pg, "SELEC 1"), "42601"); + assert_eq!(sqlstate_of(&pg, "SELECT * FROM no_such_table"), "42P01"); + + // 6. ordered comparison: a tie may reorder, a key may not + let ordered_sql = "SELECT k, v FROM t ORDER BY k"; + let pg_rows = pg.query(ordered_sql).expect("ordered").rows; + let case = Case { + name: "ordered tie", + sql: ordered_sql, + order: Order::By(&[0]), + }; + let mut tie_flipped = pg_rows.clone(); + tie_flipped.swap(0, 1); + tie_flipped.swap(2, 3); + assert_ne!(tie_flipped, pg_rows, "the NULL/empty tie must be reordered"); + let out = check(&pg, &mut report, &case, || int_text(tie_flipped)); + assert_eq!(out, Outcome::Pass); + let mut key_flipped = pg_rows.clone(); + key_flipped.rotate_right(1); // the k = 3 row first + let case = Case { + name: "ordered key", + sql: ordered_sql, + order: Order::By(&[0]), + }; + let out = check(&pg, &mut report, &case, || int_text(key_flipped)); + assert!( + matches!(out, Outcome::Fail(Diff::Ordered { .. })), + "{out:?}" + ); + + // 7. a refusal is recorded, not failed + let case = Case { + name: "refused", + sql: "SELECT k, v FROM t", + order: Order::Unordered, + }; + let out = check(&pg, &mut report, &case, || { + Answer::Unsupported(Refusal { + feature: "demo", + detail: "not implemented".into(), + }) + }); + assert!(matches!(out, Outcome::Unsupported(_))); + assert_eq!(report.unsupported().len(), 1); + + // 8. mutation test against a toy implementation holding the table + let mut state: Vec = pg.query("SELECT k, v FROM t").expect("toy").rows; + let answer = |s: &Vec| int_text(s.clone()); + let case = Case { + name: "mutation", + sql: "SELECT k, v FROM t", + order: Order::Unordered, + }; + // (a) the same change on both sides + let outs = check_mutation( + &pg, + &mut report, + &case, + "UPDATE t SET v = 'b' WHERE k = 3", + &mut state, + answer, + |s| { + for r in s.iter_mut().filter(|r| r[0].as_deref() == Some("3")) { + r[1] = text("b"); + } + }, + ); + assert_eq!(outs, (Outcome::Pass, Outcome::Pass)); + // (b) a key change the toy side never sees: a stale preparation + let outs = check_mutation( + &pg, + &mut report, + &case, + "UPDATE t SET k = 7 WHERE k = 3", + &mut state, + answer, + |_| {}, + ); + assert_eq!(outs.0, Outcome::Pass); + assert!( + matches!(outs.1, Outcome::Fail(Diff::Multiset { .. })), + "{:?}", + outs.1 + ); + for r in state.iter_mut().filter(|r| r[0].as_deref() == Some("3")) { + r[0] = text(7); // re-sync the toy state + } + // (c) a mutation PostgreSQL cannot see + let outs = check_mutation( + &pg, + &mut report, + &case, + "UPDATE t SET v = v WHERE false", + &mut state, + answer, + |_| {}, + ); + assert_eq!(outs.0, Outcome::Pass); + assert_eq!(outs.1, Outcome::Fail(Diff::VacuousMutation)); + + // 9. the scratch schema is dropped with the session object + let schema = pg.schema().to_string(); + drop(pg); + let again = Pg::connect().expect("second connect"); + let left = again + .query(&format!( + "SELECT count(*) FROM pg_namespace WHERE nspname = '{schema}'" + )) + .expect("namespace count"); + assert_eq!(left.rows, vec![vec![text(0)]]); + + // 10. + println!("{}", report.render()); + println!("PostgreSQL {version}: live_server_roundtrip ok"); +}