diff --git a/crates/applications/cli/tests/e2e_clone_robustness.rs b/crates/applications/cli/tests/e2e_clone_robustness.rs index 8b203dd2..739971af 100644 --- a/crates/applications/cli/tests/e2e_clone_robustness.rs +++ b/crates/applications/cli/tests/e2e_clone_robustness.rs @@ -439,25 +439,61 @@ fn clone_range_and_temporal_hydration_elide() { ); psql(&clone, "postgres", "TRUNCATE orders"); // fresh local state for the window + // A temporal window is answered by federating to the source, while the + // background worker copies it; the worker's copy skips clone_stats, so + // rows_fetched does not move here and federate_calls is what to watch. let tq = "SELECT id FROM orders WHERE placed_at BETWEEN date '2024-02-01' AND date '2024-02-15' ORDER BY id"; - let t0 = sum_fetched(&clone); - psql(&clone, "postgres", tq); - let t1 = sum_fetched(&clone); - assert!(t1 > t0, "temporal window hydrated (t0={t0} t1={t1})"); - psql(&clone, "postgres", tq); + let expected = psql(remote, "shop", tq); + assert!(!expected.is_empty(), "the fixture has rows in the window"); + let federated = || -> i64 { + psql( + &clone, + "postgres", + "SELECT federate_calls FROM gfs.clones WHERE clone = 'orders'", + ) + .parse() + .unwrap_or(-1) + }; + let t0 = federated(); assert_eq!( - sum_fetched(&clone), - t1, - "temporal window elided — no new source fetch" + psql(&clone, "postgres", tq), + expected, + "temporal result matches source" ); + let t1 = federated(); assert_eq!( + t1, + t0 + 1, + "an uncovered temporal window is federated (t0={t0} t1={t1})" + ); + + let deadline = Instant::now() + Duration::from_secs(60); + let copied = || { psql( &clone, "postgres", - &format!("SELECT count(*) FROM ({tq}) t") - ), - psql(remote, "shop", &format!("SELECT count(*) FROM ({tq}) t")), - "temporal result matches source" + "SELECT NOT EXISTS (SELECT 1 FROM gfs.copy_queue WHERE relid = 'orders'::regclass) \ + AND EXISTS (SELECT 1 FROM gfs.clones WHERE clone = 'orders' \ + AND (whole_cached OR cached_ranges > 0))", + ) == "t" + }; + while !copied() { + assert!( + Instant::now() < deadline, + "the background copy of the window never landed" + ); + thread::sleep(Duration::from_millis(500)); + } + + assert_eq!( + psql(&clone, "postgres", tq), + expected, + "copied window matches source" + ); + assert_eq!( + federated(), + t1, + "temporal window served locally once copied" ); drop(cl); } diff --git a/crates/extensions/gfs/src/sql/schema.sql b/crates/extensions/gfs/src/sql/schema.sql index a9cb88d8..c92ca5b5 100644 --- a/crates/extensions/gfs/src/sql/schema.sql +++ b/crates/extensions/gfs/src/sql/schema.sql @@ -1600,6 +1600,9 @@ COMMENT ON FUNCTION gfs.note_tombstone() IS 'AFTER DELETE trigger: record the de CREATE TABLE gfs.clone_stats ( relid regclass PRIMARY KEY REFERENCES gfs.clone_source(relid) ON DELETE CASCADE, fetch_calls bigint NOT NULL DEFAULT 0, + -- Rows a backend copies for its own query, plus rows gfs.warm copies. The + -- background worker's copies (temporal windows, whole-table and predicate + -- copies) are not counted: it skips this table (DEFER_BOOKKEEPING in hydrate.rs). rows_fetched bigint NOT NULL DEFAULT 0, federate_calls bigint NOT NULL DEFAULT 0, -- times this table was pushed to the source last_fetch timestamptz diff --git a/crates/extensions/gfs/tpch_validate.sh b/crates/extensions/gfs/tpch_validate.sh index 2c4ac74e..2dd37b05 100755 --- a/crates/extensions/gfs/tpch_validate.sh +++ b/crates/extensions/gfs/tpch_validate.sh @@ -39,6 +39,8 @@ PASS=0; FAIL=0 src() { PGPASSWORD=pw "$PSQL" "postgresql://app:pw@localhost:${SRC_PORT}/tpch" -tAqc "$1" 2>/dev/null; } cln() { PGPASSWORD=postgres "$PSQL" "postgresql://postgres:postgres@localhost:${CLONE_PORT}/postgres" -tAqc "$1" 2>/dev/null; } ck() { echo "SELECT COALESCE(md5(string_agg(t::text,'|' ORDER BY t::text)),'NONE') FROM ($1) t"; } +# rows_fetched leaves out the background worker's copies, so pulled() +# under-reports whenever a query federates and the worker copies afterwards. counters() { cln "SELECT COALESCE(sum(rows_fetched),0)||' '||COALESCE(sum(federate_calls),0) FROM gfs.clones"; } pulled() { cln "SELECT COALESCE(sum(rows_fetched),0) FROM gfs.clones"; }