Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 48 additions & 12 deletions crates/applications/cli/tests/e2e_clone_robustness.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
3 changes: 3 additions & 0 deletions crates/extensions/gfs/src/sql/schema.sql
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions crates/extensions/gfs/tpch_validate.sh
Original file line number Diff line number Diff line change
Expand Up @@ -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"; }

Expand Down