diff --git a/.claude/memory/reference_roborev_severity_threshold_met_is_clean.md b/.claude/memory/reference_roborev_severity_threshold_met_is_clean.md index 221da8c0..6ca0268e 100644 --- a/.claude/memory/reference_roborev_severity_threshold_met_is_clean.md +++ b/.claude/memory/reference_roborev_severity_threshold_met_is_clean.md @@ -5,7 +5,7 @@ metadata: node_type: memory type: reference originSessionId: 05a4b981-1ec9-4115-91f7-2cdcd554c7ee - modified: 2026-07-29T17:33:37.873Z + modified: 2026-07-31T00:37:19.351Z --- When `roborev show ` prints only `SEVERITY_THRESHOLD_MET` as the review body, the commit **passed**. The string reads like "the severity threshold was met, so there are findings" — it means the opposite: the reviewer produced findings that all fell *below* `review_min_severity`, which `.roborev.toml` sets to `'medium'`, so they were suppressed and never surfaced. @@ -14,4 +14,6 @@ The authoritative field is `roborev show --json | jq .verdict_bool` — ** So: an empty `kata list --label cha-NNN` alongside a `SEVERITY_THRESHOLD_MET` job is consistent, not a dropped finding. Don't go hunting for the suppressed text — check `verdict_bool` and move on. +**Timing gotcha when polling for the verdict.** For a window after a job finishes, `roborev show --json` exits non-zero with `Error: no review found for job ` while the plain `roborev show ` already renders the body — the review row lands after the job row. So a wait loop shaped like `until [ "$(roborev show N --json | jq -r '.status // "running"')" != "running" ]` **terminates immediately on that error**: jq gets no input, prints the empty string, and `"" != "running"` is true, which reads as "the job finished" when nothing was actually observed. Poll plain `roborev show` / `roborev list` for completion, then read `--json .verdict_bool` once. See [[feedback_absence_is_not_evidence]]. + Related: [[feedback_poll_roborev_after_any_commits]], and [[reference_roborev_kata_bridge_needs_cha_branch]] for the other direction — an empty queue is *not* evidence of a clean review when the bridge never ran at all. diff --git a/crates/penca-api/src/lifecycle/compact.rs b/crates/penca-api/src/lifecycle/compact.rs index 9273b49c..ef5c1264 100644 --- a/crates/penca-api/src/lifecycle/compact.rs +++ b/crates/penca-api/src/lifecycle/compact.rs @@ -357,12 +357,10 @@ where LifecycleManager::commit_compact_segment(&tx, &catalog_str, &branch_str, &merged_uri).await?; // Delete-set LAST, per the ordering invariant on - // `insert_segment_delete_set_rows`. This tx already holds a lock on the - // segment-metadata parent — its very first statement is - // `enumerate_unsealed_segments`, a `SELECT ... FOR UPDATE OF seg` against the - // catalog-wide parent — and the defer set is derived from that read, so - // compact cannot take a delete-set row lock before a parent lock even in - // principle. That is what fixes the global order for every other writer. + // `insert_segment_delete_set_rows`. Since CHA-546 every statement above + // names this branch's partitions, so the only parent lock this tx holds is + // the `AccessShare` its leaf INSERTs imply — compatible with the sweep's, + // so the invariant costs this path nothing. // // Still inside `tx`, which is what ADR 0019 §"Four-part mechanism" item 3 // requires: the row must commit atomically with the URI swap. diff --git a/crates/penca-api/src/lifecycle/retire.rs b/crates/penca-api/src/lifecycle/retire.rs index 330d67ab..dfbaab54 100644 --- a/crates/penca-api/src/lifecycle/retire.rs +++ b/crates/penca-api/src/lifecycle/retire.rs @@ -178,11 +178,12 @@ impl LifecycleManager { ) .await?; - // Delete-set LAST, after every segment-metadata parent this tx touches, - // per the ordering invariant on `insert_segment_delete_set_rows`. The - // sidecar URIs are read above (before their rows are deleted) but - // enqueued here, so the parent locks are all taken before any delete-set - // row lock. Position within the tx is free for ADR 0019 item 3 — it + // Delete-set LAST, per the ordering invariant on + // `insert_segment_delete_set_rows`. Since CHA-546 every statement above + // names this branch's partitions, and they are all SELECTs and DELETEs + // — which on a leaf take nothing on the parent — so this tx holds no + // segment-metadata parent lock at all and the invariant costs it + // nothing. Position within the tx is free for ADR 0019 item 3 — it // requires the rows to commit atomically with the retirement, not to // precede it. penca_storage_meta::LifecycleManager::insert_segment_delete_set_rows( diff --git a/crates/penca-api/src/query/meta_plan.rs b/crates/penca-api/src/query/meta_plan.rs index 7fdd6fd1..c72e864e 100644 --- a/crates/penca-api/src/query/meta_plan.rs +++ b/crates/penca-api/src/query/meta_plan.rs @@ -369,7 +369,8 @@ impl QueryManager { w_snap: i64, ) -> Result<(i64, bool)> { let catalog = parse_uuid(catalog_uuid); - let purge_table = naming::table_purge_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let purge_table = naming::table_purge_metadata_partition(&catalog, &branch); let sql = format!( "WITH fence AS (\ SELECT GREATEST(\ @@ -389,7 +390,7 @@ impl QueryManager { delete = qi(delete_table_name), ); let params = vec![ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(w_snap), ]; @@ -435,9 +436,10 @@ impl QueryManager { Option, )> { let catalog = parse_uuid(catalog_uuid); + let branch = parse_uuid(branch_uuid); let table = parse_meta_uuid(table_uuid, "table_uuid")?; - let persist_name = naming::table_persist_metadata_table(&catalog); - let snap_name = naming::table_snapshot_metadata_table(&catalog); + let persist_name = naming::table_persist_metadata_partition(&catalog, &branch); + let snap_name = naming::table_snapshot_metadata_partition(&catalog, &branch); // Fold the child's fork lineage into this already-per-read query (a // branch_store PK join) so a non-forked read pays no extra round-trip // for the base-source gate. @@ -449,7 +451,7 @@ impl QueryManager { }; let mut params: Vec = vec![ SqlValue::Uuid(table), - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::Int64(as_of_micros), ]; if let Some(seq) = commit_seq_upper { @@ -705,22 +707,22 @@ impl QueryManager { pinned_snapshot_uuid: Option, ) -> Result { let catalog = parse_uuid(catalog_uuid); + let branch = parse_uuid(branch_uuid); // Parse fallibly (unlike the panicking `parse_uuid` above): a malformed // `table_uuid` surfaces as the same typed protocol error the // `meta_resolve` getters produce. let table = parse_meta_uuid(table_uuid, "table_uuid")?; - let snap_name = naming::table_snapshot_metadata_table(&catalog); - let seg_name = naming::table_snapshot_segment_metadata_table(&catalog); + let snap_name = naming::table_snapshot_metadata_partition(&catalog, &branch); + let seg_name = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); // The internal row_uuid index parent/child, joined in below so a // planned snapshot segment carries its sidecar inline. - let idx_parent = naming::table_snapshot_index_metadata_table(&catalog); - let idx_child = naming::table_snapshot_segment_index_metadata_table(&catalog); + let idx_parent = naming::table_snapshot_index_metadata_partition(&catalog, &branch); + let idx_child = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); // Params `$1` = table, `$2` = branch (fixed in the JOINs / WHERE); the // rest are bound in push order below, each `$N` computed from // `params.len()`, so the pinned-uuid and as_of/seq picks share one // numbering scheme. - let mut params: Vec = - vec![SqlValue::Uuid(table), SqlValue::uuid_str(branch_uuid)?]; + let mut params: Vec = vec![SqlValue::Uuid(table), SqlValue::Uuid(branch)]; let snapshot_selection = if let Some(uuid) = pinned_snapshot_uuid { params.push(SqlValue::Uuid(uuid)); format!("snap.table_snapshot_uuid = ${}", params.len()) @@ -1027,7 +1029,8 @@ impl QueryManager { fork_commit_seq_num: i64, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let seg = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let seg = naming::table_persist_segment_metadata_partition(&catalog, &branch); let rows = driver .execute_params( &format!( @@ -1038,7 +1041,7 @@ impl QueryManager { seg = qi(&seg), ), &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(fork_commit_seq_num), ], @@ -1248,8 +1251,15 @@ impl QueryManager { // `commit_seq_num` already exceeds the cutoff. Composes with the // committed_at tier fence; absent for the micros / OpenTx axes // (`commit_seq_upper = None`). - let seg_table = naming::table_persist_segment_metadata_table(catalog_uuid); - let tfm_table = naming::table_persist_metadata_table(catalog_uuid); + // + // Both relations are named as the `branch_uuid` ARGUMENT's partitions, + // never an ambient current-branch value: `enumerate_base_cold_source` + // calls this with a fork's PARENT branch to resolve inherited cold, + // while the ordinary read path calls it with the reading branch. Taking + // the name from anywhere else compiles and silently reads the wrong + // branch's segments. + let seg_table = naming::table_persist_segment_metadata_partition(catalog_uuid, branch_uuid); + let tfm_table = naming::table_persist_metadata_partition(catalog_uuid, branch_uuid); let mut log_sql = format!( "SELECT tfm.log_kind, \ seg.table_persist_segment_uuid AS segment_uuid, \ @@ -1370,8 +1380,8 @@ impl QueryManager { let catalog = parse_uuid(catalog_uuid); let branch = parse_uuid(branch_uuid); let table = parse_uuid(table_uuid); - let seg_table = naming::table_persist_segment_metadata_table(&catalog); - let tfm_table = naming::table_persist_metadata_table(&catalog); + let seg_table = naming::table_persist_segment_metadata_partition(&catalog, &branch); + let tfm_table = naming::table_persist_metadata_partition(&catalog, &branch); let mut sql = format!( "SELECT MAX(seg.max_commit_seq_num) AS max_seq \ FROM {seg} seg \ diff --git a/crates/penca-api/src/write/mod.rs b/crates/penca-api/src/write/mod.rs index e837ba8e..5f8d59fd 100644 --- a/crates/penca-api/src/write/mod.rs +++ b/crates/penca-api/src/write/mod.rs @@ -1086,28 +1086,38 @@ impl WriteManager { // safe — what the caller needs is to TELL a lock loss from a real failure, // which is why this maps to `Aborted` rather than the default `Internal`. // - // One conflict ordering cannot remove, TODO(CHA-546): a lifecycle write - // names the catalog-wide PARENT, so it holds ROW EXCLUSIVE there while - // waiting on this branch's leaf, and the drops below need ACCESS - // EXCLUSIVE on that same parent — a cycle, measured, which Postgres - // resolves by killing teardown. Rare (it needs a metadata write on this - // branch inside teardown's window) and clean (full rollback, reported as - // `Aborted`, succeeds on reissue). CHA-546 converts those writes to name - // the partition, as the tx-log family already does, which removes it. + // One conflict ordering cannot remove: the drops below need ACCESS + // EXCLUSIVE on the catalog-wide parents, which conflicts with EVERY + // mode, so any concurrent holder of any mode on a parent delays them. + // Two such holders survive CHA-546 by design. CHA-531's refcount gate + // reads those parents catalog-wide, and — measured, not assumed — an + // INSERT or UPDATE naming a LEAF still takes `AccessShare` on its + // parent to evaluate the partition constraint, held to commit. So any + // in-flight writer on any branch in the catalog delays these drops. + // + // Both are waits, not cycles — neither holder takes a lock this + // transaction holds — so they surface as a `lock_timeout` rather than a + // deadlock kill, and are clean either way (full rollback, reported as + // `Aborted`, succeeds on reissue). What CHA-546 bought here is the + // EXCLUSIVE step above rather than these drops: `AccessShare` clears + // it, the pre-CHA-546 `RowExclusive` did not. // // The branch's HOT data tables (`schema_uuid = None` = catalog-wide), // resolved BEFORE the teardown transaction and used only for their drops. // - // It cannot move inside. `list_table_uuids_for_branch` plans through - // `QueryManager::plan`, which reads the catalog-wide metadata parents BY - // NAME — so running it in the transaction takes `ACCESS SHARE` on every - // one of them, which is precisely what - // `lock_branch_teardown_partitions` is built to avoid: it stalls plans on - // every branch in the catalog while this cold-capable read waits on - // object storage, and it sets up the `ACCESS SHARE` -> `ACCESS EXCLUSIVE` - // upgrade at the drops that deadlocks two concurrent teardowns of - // DIFFERENT branches. Partition-scoping the enumerations bought exactly - // that property; planning inside the lock gives it back. + // It cannot move inside, and since CHA-546 the reason is duration + // rather than lock footprint. `list_table_uuids_for_branch` resolves + // `sys_tables` through the full read path, which names only this + // branch's partitions — so it would take no parent lock. What it does + // take is unbounded time: it is a cold-capable read that waits on + // object storage, and NOTHING bounds it. `lock_timeout` governs lock + // acquisition, not execution, and no `statement_timeout` is set on + // this path — so inside the transaction that wait runs to completion + // while this branch's 14 leaves are held EXCLUSIVE, blocking every + // writer on the branch for the length of an S3 fetch. It also widens + // the window before the drops, which must then win ACCESS EXCLUSIVE on + // the catalog-wide parents within the 5s bound — the one place that + // bound does bite (see above). // // The cost is a real leak, stated plainly rather than filed under // "best effort": a `CreateTable` committing between this read and the @@ -1117,8 +1127,8 @@ impl WriteManager { // delete-set consequence — but they are permanent. // // Pre-existing, not introduced here: `main` resolves this list the same - // way. Closing it needs a branch-scoped table enumeration that does not - // plan through the parents, which does not exist today — the hot data + // way. Closing it needs a branch-scoped table enumeration that cannot + // reach object storage, which does not exist today — the hot data // relations are named `hash(table_uuid, branch_uuid)`, so they cannot be // recovered from `pg_class` by branch either. Needs its own ticket. let table_uuid_strs = self @@ -1241,12 +1251,13 @@ impl WriteManager { LifecycleManager::drop_branch_partitions(tx, catalog_str, branch_str).await?; // Delete-set LAST, after the partition drops, per the ordering - // invariant on `insert_segment_delete_set_rows`. Dropping a - // partition takes ACCESS EXCLUSIVE on the catalog-wide parent; a - // concurrent compact holds ROW SHARE on that same parent from its - // opening `SELECT ... FOR UPDATE` and cannot release it, so - // teardown must not be holding a delete-set row while it waits - // for the parent. Still one transaction, so removing the + // invariant on `insert_segment_delete_set_rows`. Teardown is the + // one writer still subject to it: dropping a partition takes + // ACCESS EXCLUSIVE on the catalog-wide parent, and the sweep's + // refcount gate holds AccessShare on that same parent — CHA-531 + // keeps those probes catalog-wide — while it locks delete-set + // rows. So teardown must not be holding a delete-set row while it + // waits for the parent. Still one transaction, so removing the // references and queueing the files remain one atomic fact. LifecycleManager::insert_segment_delete_set_rows(tx, catalog_str, &queued_uris) .await?; diff --git a/crates/penca-core/src/naming/tables.rs b/crates/penca-core/src/naming/tables.rs index d7f55ef5..bf02c27b 100644 --- a/crates/penca-core/src/naming/tables.rs +++ b/crates/penca-core/src/naming/tables.rs @@ -551,4 +551,17 @@ mod tests { "6830ca7e-5210-6616-91cf-34c0e0d7c612_tx_table_log_partition" ); } + + #[test] + fn test_parity_table_snapshot_segment_metadata_partition() { + // CHA-546: the first METADATA leaf a Python caller has to name. + // Integration tests count PG statements by relation name, so the + // client must compute byte-identical leaf names to the server — + // a mismatch makes a `== 0` statement-count assertion pass + // vacuously rather than fail. Golden mirrors the Python suite. + assert_eq!( + table_snapshot_segment_metadata_partition(&CAT, &BR), + "1a93a047-8229-c30f-2de8-08a4482c2051_table_snapshot_segment_metadata_partition" + ); + } } diff --git a/crates/penca-db/src/dialect/pg.rs b/crates/penca-db/src/dialect/pg.rs index 82609b65..634f521a 100644 --- a/crates/penca-db/src/dialect/pg.rs +++ b/crates/penca-db/src/dialect/pg.rs @@ -1380,21 +1380,26 @@ impl PgDialect { .join(", "); // `SET LOCAL` is TRANSACTION-scoped, not statement-scoped, so this bound // governs every later statement too — including the 14 `DROP TABLE`s, - // which is where it actually bites. That is deliberate but worth stating, - // because the dominant conflict is not another teardown: `compact` opens - // with `SELECT ... FOR UPDATE OF seg` naming the catalog-wide - // `table_persist_segment_metadata` PARENT, so it holds `ROW SHARE` there - // across its whole cold read and merged write. A drop needs `ACCESS - // EXCLUSIVE` on that parent, so a compact running anywhere in the catalog - // for longer than this bound fails the teardown. + // which is where it actually bites, since `ACCESS EXCLUSIVE` on a parent + // conflicts with every mode while the `EXCLUSIVE` taken here does not + // conflict with `AccessShare`. + // + // On this branch, its own lifecycle work holds `ROW SHARE` / + // `ROW EXCLUSIVE` on the very leaves locked here — deleting a branch + // while writing it is the caller's own race, and the `EXCLUSIVE` below + // is what catches it. Catalog-wide, two holders can still delay the + // drops: CHA-531's refcount gate, which probes the segment-metadata + // parents in a single statement, and any in-flight writer on any branch, + // because an `INSERT`/`UPDATE` naming a LEAF still takes `AccessShare` + // on its parent to evaluate the partition constraint (measured — reads + // and `DELETE`s take nothing). Both are bounded by a statement or a + // transaction rather than by a whole lifecycle wave, which is the + // difference CHA-546 made. // // Failing is the right end of the trade — waiting instead would queue // every subsequent lock request on that parent behind us, turning one - // slow compact into a catalog-wide stall. The caller reports it as - // `Aborted` and reissues. TODO(CHA-546): the root cause is compact naming - // the parent rather than the branch's partition; once reads and writes - // both target partitions, teardown's drops contend with nothing outside - // the branch and this stops being reachable in steady state. + // slow writer into a catalog-wide stall. The caller reports it as + // `Aborted` and reissues. // // Both errors propagate UNWRAPPED. Re-wrapping as `sqlx::Error::Protocol` // would erase the `Error::Database` variant, and `as_database_error()` diff --git a/crates/penca-storage-meta/src/compact.rs b/crates/penca-storage-meta/src/compact.rs index 5735f658..0a3080e5 100644 --- a/crates/penca-storage-meta/src/compact.rs +++ b/crates/penca-storage-meta/src/compact.rs @@ -53,7 +53,8 @@ impl LifecycleManager { object_uri: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::compact_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::compact_segment_metadata_partition(&catalog, &branch); let sql = format!( "INSERT INTO {table} \ (object_uri, branch_uuid, table_uuid, commit_micros) \ @@ -65,7 +66,7 @@ impl LifecycleManager { &sql, &[ SqlValue::Text(object_uri.to_string()), - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, ], ) @@ -89,7 +90,8 @@ impl LifecycleManager { object_uri: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::compact_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::compact_segment_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND object_uri = $2", @@ -100,7 +102,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::Text(object_uri.to_string()), ], ) @@ -132,7 +134,8 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); let arr = format_sql_uuid_array(segment_uuids); let sql = format!( "UPDATE {table} SET is_sealed = TRUE \ @@ -141,7 +144,7 @@ impl LifecycleManager { table = qi(&table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -168,7 +171,7 @@ impl LifecycleManager { table = qi(&table), ); let rows = driver - .execute_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(rows .iter() @@ -217,28 +220,43 @@ impl LifecycleManager { /// same with its segment-row deletes). /// /// **Lock-ordering invariant — call this LAST, after every - /// segment-metadata parent the transaction touches.** Three writers - /// mutate both this table and those parents: the compact merge - /// (`lifecycle::compact`), snapshot retirement (`lifecycle::retire`), - /// and branch teardown (`write::delete_branch`). A writer holding a - /// delete-set row while it waits for a parent lock deadlocks against - /// one holding that parent while it waits for the row, and PG breaks - /// the cycle by aborting a side — a nondeterministic user-visible - /// failure or a killed lifecycle wave. A URI shared across a fork - /// edge makes it reachable with no misuse, since carry-forward and - /// the CHA-539 fork copy both leave one file referenced from several - /// branches. + /// segment-metadata parent the transaction touches.** Since CHA-546 + /// made every branch-scoped statement name its partition, branch + /// teardown (`write::delete_branch`) is the only writer left that can + /// CONFLICT on a parent: `DROP TABLE` on a leaf needs `ACCESS + /// EXCLUSIVE` on the parent to rewrite its partition descriptor, and + /// teardown enqueues here in the same transaction. + /// + /// The other two callers clear the invariant for different reasons, + /// and the difference is worth keeping straight. Naming a leaf + /// removes the parent lock outright for `SELECT`, `SELECT ... FOR + /// UPDATE` and `DELETE`, but an `INSERT` or `UPDATE` evaluates the + /// leaf's partition constraint, which opens the parent for its + /// partition key and leaves `AccessShare` held to commit (measured; + /// see the table in + /// `tests/integration/integration_cha546_partition_lock_footprint_test.py`). + /// So the compact merge (`lifecycle::compact`) does hold a parent + /// `AccessShare` — a harmless one, since it cannot conflict with the + /// `AccessShare` the sweep's gate takes. Snapshot retirement + /// (`lifecycle::retire`) issues only leaf `SELECT`s and `DELETE`s + /// before arriving here, so it holds no parent lock at all. Don't + /// collapse the two into "they touch no parent": that is true of + /// retire and false of compact, and a future reader who believed it + /// of both might drop the rule. /// - /// The direction is forced, not chosen. `compact_one_scope`'s FIRST - /// statement is `enumerate_unsealed_segments` — a - /// `SELECT ... FOR UPDATE OF seg` against the catalog-wide - /// `table_persist_segment_metadata` parent, taking `ROW SHARE` held - /// to commit — and the URIs it defers are derived from that read. So - /// compact cannot reach a delete-set row before a parent lock even in - /// principle, and `ROW SHARE` conflicts with the `ACCESS EXCLUSIVE` - /// that teardown's `DROP PARTITION` needs on the same parent. - /// Parent-locks-first is therefore the only order all three can - /// honor; the other two conform to compact. + /// The direction is forced, not chosen. CHA-531's refcount gate + /// ([`Self::eligible_segment_delete_set_rows`], + /// [`Self::reap_referenced_segment_delete_set_rows`]) probes those + /// three parents catalog-wide — deliberately, since carry-forward + /// crosses fork edges — in the same statements that read and delete + /// these rows, and a statement takes its table locks before any row + /// lock. Parents-first is therefore a property of the sweep's + /// statements rather than a choice it makes, and teardown is the side + /// that conforms. Reversed, teardown's `ON CONFLICT` grace refresh + /// would hold a delete-set row — one a fork-shared URI makes likely + /// to already exist — while waiting for a parent the sweep holds, and + /// PG would break the cycle by aborting a side: a nondeterministic + /// user-visible failure or a killed lifecycle wave. /// /// Position within the transaction is free as far as ADR 0019 /// §"Four-part mechanism" item 3 is concerned: it requires these rows @@ -390,6 +408,12 @@ impl LifecycleManager { /// fourth referencing table were added to the gate but not to the reaper, the /// reaper would drop delete-set rows for URIs that table still references. /// + /// All three names must be the catalog-wide PARENTS — the one place CHA-546's + /// partition-direct rule deliberately does not apply, for the reason spelled + /// out on [`Self::eligible_segment_delete_set_rows`]. A partition here would + /// narrow the refcount to a single branch and collect files a sibling still + /// reads. + /// /// `committed_only` is where the two callers deliberately differ, and the /// asymmetry is load-bearing in both directions: /// diff --git a/crates/penca-storage-meta/src/fork_copy.rs b/crates/penca-storage-meta/src/fork_copy.rs index e4afdfaf..b73cc8f6 100644 --- a/crates/penca-storage-meta/src/fork_copy.rs +++ b/crates/penca-storage-meta/src/fork_copy.rs @@ -130,10 +130,19 @@ impl LifecycleManager { fork_commit_micros: i64, commit_micros: i64, ) -> Result { - let snap_meta = naming::table_snapshot_metadata_table(catalog); - let snap_seg = naming::table_snapshot_segment_metadata_table(catalog); - let idx_meta = naming::table_snapshot_index_metadata_table(catalog); - let idx_seg = naming::table_snapshot_segment_index_metadata_table(catalog); + // Every statement below spans the fork edge, so each table needs both + // sides named: rows are read from the parent's partition and written + // into the child's. + let parent = parse_uuid(parent_branch_uuid); + let snap_meta_child = naming::table_snapshot_metadata_partition(catalog, child); + let snap_meta_parent = naming::table_snapshot_metadata_partition(catalog, &parent); + let snap_seg_child = naming::table_snapshot_segment_metadata_partition(catalog, child); + let snap_seg_parent = naming::table_snapshot_segment_metadata_partition(catalog, &parent); + let idx_meta_child = naming::table_snapshot_index_metadata_partition(catalog, child); + let idx_meta_parent = naming::table_snapshot_index_metadata_partition(catalog, &parent); + let idx_seg_child = naming::table_snapshot_segment_index_metadata_partition(catalog, child); + let idx_seg_parent = + naming::table_snapshot_segment_index_metadata_partition(catalog, &parent); // Bounded on BOTH axes. Seq is the authority for the ceiling — a // same-micros higher-seq parent commit must not be inherited — and the @@ -149,10 +158,10 @@ impl LifecycleManager { AND snapshotted_at_micros <= $4 \ ORDER BY snapshotted_at_micros DESC, commit_seq_num DESC \ LIMIT 1", - snap = qi(&snap_meta), + snap = qi(&snap_meta_parent), ), &[ - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Uuid(*table), SqlValue::Int64(fork_commit_seq_num), SqlValue::Int64(fork_commit_micros), @@ -179,15 +188,16 @@ impl LifecycleManager { SELECT $1, $2, old.table_uuid, old.snapshotted_at_micros, \ old.commit_seq_num, old.durable, old.partition_keys, \ old.clustering_keys, $5 \ - FROM {snap} old \ + FROM {snap_old} old \ WHERE old.branch_uuid = $3 AND old.table_snapshot_uuid = $4 \ ON CONFLICT (branch_uuid, table_snapshot_uuid) DO NOTHING", - snap = qi(&snap_meta), + snap = qi(&snap_meta_child), + snap_old = qi(&snap_meta_parent), ), &[ SqlValue::Uuid(new_snap), SqlValue::Uuid(*child), - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Uuid(parent_snap), SqlValue::Int64(commit_micros), ], @@ -204,12 +214,9 @@ impl LifecycleManager { WHERE branch_uuid = $1 AND table_snapshot_uuid = $2 \ AND commit_micros IS NOT NULL \ ORDER BY chunk_idx, \"offset\"", - seg = qi(&snap_seg), + seg = qi(&snap_seg_parent), ), - &[ - SqlValue::uuid_str(parent_branch_uuid)?, - SqlValue::Uuid(parent_snap), - ], + &[SqlValue::Uuid(parent), SqlValue::Uuid(parent_snap)], ) .await?; let mut seg_map: Vec<(Uuid, Uuid, u32)> = Vec::with_capacity(segs.len()); @@ -250,13 +257,14 @@ impl LifecycleManager { SELECT m.new_uuid, $1, $2, old.table_uuid, m.chunk_idx, old.object_uri, \ old.\"offset\", old.length, old.size_bytes, old.format, \ old.metadata, old.statistics, old.row_count, $4 \ - FROM {seg} old \ + FROM {seg_old} old \ JOIN unnest({new_arr}, {old_arr}, {chunk_arr}) \ AS m(new_uuid, old_uuid, chunk_idx) \ ON old.table_snapshot_segment_uuid = m.old_uuid \ WHERE old.branch_uuid = $3 \ ON CONFLICT (branch_uuid, table_snapshot_segment_uuid) DO NOTHING", - seg = qi(&snap_seg), + seg = qi(&snap_seg_child), + seg_old = qi(&snap_seg_parent), new_arr = format_sql_uuid_array(&new_refs), old_arr = format_sql_uuid_array(&old_refs), chunk_arr = chunk_arr, @@ -264,7 +272,7 @@ impl LifecycleManager { &[ SqlValue::Uuid(new_snap), SqlValue::Uuid(*child), - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Int64(commit_micros), ], ) @@ -279,12 +287,9 @@ impl LifecycleManager { "SELECT table_snapshot_index_uuid, index_uuid FROM {idx} \ WHERE branch_uuid = $1 AND table_snapshot_uuid = $2 \ AND commit_micros IS NOT NULL", - idx = qi(&idx_meta), + idx = qi(&idx_meta_parent), ), - &[ - SqlValue::uuid_str(parent_branch_uuid)?, - SqlValue::Uuid(parent_snap), - ], + &[SqlValue::Uuid(parent), SqlValue::Uuid(parent_snap)], ) .await?; let mut parent_map: Vec<(Uuid, Uuid, String)> = Vec::with_capacity(idx_parents.len()); @@ -309,19 +314,20 @@ impl LifecycleManager { (table_snapshot_index_uuid, branch_uuid, table_snapshot_uuid, \ index_uuid, key_columns, commit_micros) \ SELECT m.new_parent, $1, $2, old.index_uuid, old.key_columns, $4 \ - FROM {idx} old \ + FROM {idx_old} old \ JOIN unnest({new_arr}, {old_arr}) AS m(new_parent, old_parent) \ ON old.table_snapshot_index_uuid = m.old_parent \ WHERE old.branch_uuid = $3 \ ON CONFLICT (branch_uuid, table_snapshot_index_uuid) DO NOTHING", - idx = qi(&idx_meta), + idx = qi(&idx_meta_child), + idx_old = qi(&idx_meta_parent), new_arr = format_sql_uuid_array(&new_refs), old_arr = format_sql_uuid_array(&old_refs), ), &[ SqlValue::Uuid(*child), SqlValue::Uuid(new_snap), - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Int64(commit_micros), ], ) @@ -367,14 +373,15 @@ impl LifecycleManager { SELECT m.new_sidecar, $1, m.new_seg, m.new_parent, old.object_uri, \ old.\"offset\", old.length, old.format, old.size_bytes, \ old.statistics, $3 \ - FROM {sidecar} old \ + FROM {sidecar_old} old \ JOIN unnest({a1}, {a2}, {a3}, {a4}, {a5}) \ AS m(new_sidecar, new_seg, new_parent, old_seg, old_parent) \ ON old.segment_uuid = m.old_seg \ AND old.table_snapshot_index_uuid = m.old_parent \ WHERE old.branch_uuid = $2 AND old.commit_micros IS NOT NULL \ ON CONFLICT (branch_uuid, segment_index_uuid) DO NOTHING", - sidecar = qi(&idx_seg), + sidecar = qi(&idx_seg_child), + sidecar_old = qi(&idx_seg_parent), a1 = a(&sc_new), a2 = a(&sc_new_seg), a3 = a(&sc_new_parent), @@ -383,7 +390,7 @@ impl LifecycleManager { ), &[ SqlValue::Uuid(*child), - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Int64(commit_micros), ], ) @@ -407,8 +414,12 @@ impl LifecycleManager { baseline_watermark: Option, commit_micros: i64, ) -> Result<()> { - let persist_meta = naming::table_persist_metadata_table(catalog); - let persist_seg = naming::table_persist_segment_metadata_table(catalog); + // Both sides named, as in `copy_inherited_snapshot`. + let parent = parse_uuid(parent_branch_uuid); + let persist_meta_child = naming::table_persist_metadata_partition(catalog, child); + let persist_meta_parent = naming::table_persist_metadata_partition(catalog, &parent); + let persist_seg_child = naming::table_persist_segment_metadata_partition(catalog, child); + let persist_seg_parent = naming::table_persist_segment_metadata_partition(catalog, &parent); // Headers from the baseline INCLUSIVE; segments only strictly above it. // // The inclusive header is load-bearing and easy to lose. The read plan's @@ -452,10 +463,10 @@ impl LifecycleManager { AND commit_micros IS NOT NULL \ AND persisted_at_micros >= $3 \ ORDER BY persisted_at_micros", - meta = qi(&persist_meta), + meta = qi(&persist_meta_parent), ), &[ - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Uuid(*table), SqlValue::Int64(header_from), ], @@ -500,10 +511,10 @@ impl LifecycleManager { AND commit_micros IS NOT NULL \ AND min_commit_seq_num <= $3 \ ORDER BY chunk_idx", - seg = qi(&persist_seg), + seg = qi(&persist_seg_parent), ), &[ - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Uuid(old_header), SqlValue::Int64(fork_commit_seq_num), ], @@ -547,15 +558,16 @@ impl LifecycleManager { commit_seq_num, log_kind, commit_micros) \ SELECT $1, $2, old.table_uuid, old.persisted_at_micros, \ LEAST(old.commit_seq_num, $5), old.log_kind, $6 \ - FROM {meta} old \ + FROM {meta_old} old \ WHERE old.branch_uuid = $3 AND old.table_persist_uuid = $4 \ ON CONFLICT (branch_uuid, table_persist_uuid) DO NOTHING", - meta = qi(&persist_meta), + meta = qi(&persist_meta_child), + meta_old = qi(&persist_meta_parent), ), &[ SqlValue::Uuid(new_header), SqlValue::Uuid(*child), - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Uuid(old_header), SqlValue::Int64(fork_commit_seq_num), SqlValue::Int64(commit_micros), @@ -606,19 +618,20 @@ impl LifecycleManager { old.object_uri, old.\"offset\", old.length, old.row_count, \ old.format, old.size_bytes, old.metadata, old.statistics, \ TRUE, $5 \ - FROM {seg} old \ + FROM {seg_old} old \ JOIN unnest({new_arr}, {old_arr}) AS m(new_uuid, old_uuid) \ ON old.table_persist_segment_uuid = m.old_uuid \ WHERE old.branch_uuid = $3 \ ON CONFLICT (branch_uuid, table_persist_segment_uuid) DO NOTHING", - seg = qi(&persist_seg), + seg = qi(&persist_seg_child), + seg_old = qi(&persist_seg_parent), new_arr = format_sql_uuid_array(&new_refs), old_arr = format_sql_uuid_array(&old_refs), ), &[ SqlValue::Uuid(new_header), SqlValue::Uuid(*child), - SqlValue::uuid_str(parent_branch_uuid)?, + SqlValue::Uuid(parent), SqlValue::Int64(fork_commit_seq_num), SqlValue::Int64(commit_micros), ], diff --git a/crates/penca-storage-meta/src/lifecycle.rs b/crates/penca-storage-meta/src/lifecycle.rs index 59b054f1..6e723c64 100644 --- a/crates/penca-storage-meta/src/lifecycle.rs +++ b/crates/penca-storage-meta/src/lifecycle.rs @@ -223,7 +223,7 @@ impl LifecycleManager { page_size: i64, offset: i64, ) -> Result> { - let table = naming::table_persist_metadata_table(catalog_uuid); + let table = naming::table_persist_metadata_partition(catalog_uuid, branch_uuid); let mut params: Vec = Vec::new(); params.push(SqlValue::Uuid(*branch_uuid)); @@ -323,7 +323,7 @@ impl LifecycleManager { let tx_table_part = naming::tx_table_log_partition(catalog_uuid, branch_uuid); let commit_tx_log_part = naming::commit_tx_log_partition(catalog_uuid, branch_uuid); let abort_part = naming::abort_tx_log_partition(catalog_uuid, branch_uuid); - let purge_table = naming::table_purge_metadata_table(catalog_uuid); + let purge_table = naming::table_purge_metadata_partition(catalog_uuid, branch_uuid); // The committed_at filter pins the `Pu`/`Pa` view as-of // `cleanup_started_at`, against a concurrent Purge mid-pass. let sql = format!( diff --git a/crates/penca-storage-meta/src/persist.rs b/crates/penca-storage-meta/src/persist.rs index 9825bd73..4e881fbb 100644 --- a/crates/penca-storage-meta/src/persist.rs +++ b/crates/penca-storage-meta/src/persist.rs @@ -44,7 +44,8 @@ impl LifecycleManager { commit_seq_num: Option, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_metadata_partition(&catalog, &branch); let sql = format!( "INSERT INTO {table} \ (table_persist_uuid, branch_uuid, table_uuid, \ @@ -61,7 +62,7 @@ impl LifecycleManager { &sql, &[ SqlValue::uuid_str(table_persist_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(persisted_at_micros), SqlValue::Text(log_kind.as_str().to_string()), @@ -82,7 +83,8 @@ impl LifecycleManager { table_persist_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND table_persist_uuid = $2", @@ -93,7 +95,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_uuid)?, ], ) @@ -111,7 +113,8 @@ impl LifecycleManager { table_persist_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {table} \ WHERE branch_uuid = $1 AND table_persist_uuid = $2 \ @@ -122,7 +125,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_uuid)?, ], ) @@ -146,7 +149,8 @@ impl LifecycleManager { table_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_metadata_partition(&catalog, &branch); let sql = format!( "SELECT MAX(persisted_at_micros) AS watermark FROM {table} \ WHERE branch_uuid = $1 \ @@ -157,10 +161,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(rows @@ -194,7 +195,8 @@ impl LifecycleManager { table_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_metadata_partition(&catalog, &branch); let sql = format!( "SELECT MAX(commit_seq_num) AS watermark FROM {table} \ WHERE branch_uuid = $1 \ @@ -205,10 +207,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(rows @@ -265,8 +264,9 @@ impl LifecycleManager { // `latest_committed_table_persist_watermark` — lifecycle ops call that // without a floor, so overloading it would ripple. let catalog = parse_uuid(catalog_uuid); - let persist_name = naming::table_persist_metadata_table(&catalog); - let snap_name = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let persist_name = naming::table_persist_metadata_partition(&catalog, &branch); + let snap_name = naming::table_snapshot_metadata_partition(&catalog, &branch); let window_start = crate::retention_window_start_expr("$3"); let floor_select = crate::retention_floor_select(&qi(&snap_name), "$1", "$2", &window_start); @@ -284,7 +284,7 @@ impl LifecycleManager { .execute_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(duration_seconds), ], @@ -334,7 +334,8 @@ impl LifecycleManager { statistics: &[u8], ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); // min/max_commit_seq_num are stamped alongside the committed_at bounds // and, like them, are NOT refreshed by the DO UPDATE: compact re-points // storage location only and preserves the original commit-order bounds. @@ -358,7 +359,7 @@ impl LifecycleManager { &[ SqlValue::uuid_str(table_persist_segment_uuid)?, SqlValue::uuid_str(table_persist_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(chunk_idx as i64), SqlValue::Int64(min_tx_commit_micros), @@ -415,7 +416,8 @@ impl LifecycleManager { seal_now: bool, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); let seal_clause = if seal_now { ", is_sealed = TRUE" } else { "" }; let sql = format!( "UPDATE {table} SET \ @@ -439,7 +441,7 @@ impl LifecycleManager { SqlValue::Int64(size_bytes), SqlValue::Text(format_text.to_string()), SqlValue::Bytes(statistics.to_vec()), - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_segment_uuid)?, ], ) @@ -458,7 +460,8 @@ impl LifecycleManager { size_bytes: i64, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET size_bytes = $1 \ WHERE branch_uuid = $2 AND table_persist_segment_uuid = $3", @@ -469,7 +472,7 @@ impl LifecycleManager { &sql, &[ SqlValue::Int64(size_bytes), - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_segment_uuid)?, ], ) @@ -487,7 +490,8 @@ impl LifecycleManager { table_persist_segment_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND table_persist_segment_uuid = $2", @@ -498,7 +502,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_segment_uuid)?, ], ) @@ -516,7 +520,8 @@ impl LifecycleManager { table_persist_segment_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {table} \ WHERE branch_uuid = $1 AND table_persist_segment_uuid = $2 \ @@ -527,7 +532,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_persist_segment_uuid)?, ], ) @@ -556,8 +561,9 @@ impl LifecycleManager { max_persisted_at_micros: Option, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let seg_table = naming::table_persist_segment_metadata_table(&catalog); - let tfm_table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let seg_table = naming::table_persist_segment_metadata_partition(&catalog, &branch); + let tfm_table = naming::table_persist_metadata_partition(&catalog, &branch); let mut sql = format!( "SELECT DISTINCT seg.table_uuid, tfm.log_kind \ FROM {seg} seg \ @@ -570,7 +576,7 @@ impl LifecycleManager { seg = qi(&seg_table), tfm = qi(&tfm_table), ); - let mut params: Vec = vec![SqlValue::uuid_str(branch_uuid)?]; + let mut params: Vec = vec![SqlValue::Uuid(branch)]; if let Some(min) = min_persisted_at_micros { params.push(SqlValue::Int64(min)); sql.push_str(&format!(" AND seg.commit_micros >= ${}", params.len())); @@ -619,8 +625,9 @@ impl LifecycleManager { max_persisted_at_micros: Option, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let seg_table = naming::table_persist_segment_metadata_table(&catalog); - let tfm_table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let seg_table = naming::table_persist_segment_metadata_partition(&catalog, &branch); + let tfm_table = naming::table_persist_metadata_partition(&catalog, &branch); let mut sql = format!( "SELECT DISTINCT tfm.log_kind \ FROM {seg} seg \ @@ -634,10 +641,8 @@ impl LifecycleManager { seg = qi(&seg_table), tfm = qi(&tfm_table), ); - let mut params: Vec = vec![ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ]; + let mut params: Vec = + vec![SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?]; if let Some(min) = min_persisted_at_micros { params.push(SqlValue::Int64(min)); sql.push_str(&format!(" AND seg.commit_micros >= ${}", params.len())); @@ -689,8 +694,9 @@ impl LifecycleManager { for_update: bool, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let seg_table = naming::table_persist_segment_metadata_table(&catalog); - let tfm_table = naming::table_persist_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let seg_table = naming::table_persist_segment_metadata_partition(&catalog, &branch); + let tfm_table = naming::table_persist_metadata_partition(&catalog, &branch); let mut sql = format!( "SELECT seg.table_persist_segment_uuid, seg.table_persist_uuid, seg.object_uri, \ seg.\"offset\", seg.length, seg.format, seg.row_count, seg.size_bytes, \ @@ -709,7 +715,7 @@ impl LifecycleManager { tfm = qi(&tfm_table), ); let mut params: Vec = vec![ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Text(log_kind.as_str().to_string()), ]; @@ -755,7 +761,7 @@ impl LifecycleManager { table = qi(&table), ); let rows = driver - .execute_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(rows .iter() @@ -783,7 +789,8 @@ impl LifecycleManager { table_uuids: &[&str], ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); if table_uuids.is_empty() { return Ok(Vec::new()); } @@ -796,7 +803,7 @@ impl LifecycleManager { table = qi(&table), ); let rows = driver - .execute_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(rows .iter() @@ -821,7 +828,8 @@ impl LifecycleManager { table_persist_segment_uuids: &[String], ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_persist_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_persist_segment_metadata_partition(&catalog, &branch); let segment_uuid_refs: Vec<&str> = table_persist_segment_uuids .iter() .map(String::as_str) @@ -834,7 +842,7 @@ impl LifecycleManager { table = qi(&table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } diff --git a/crates/penca-storage-meta/src/purge.rs b/crates/penca-storage-meta/src/purge.rs index fb28046f..1c4d0ff6 100644 --- a/crates/penca-storage-meta/src/purge.rs +++ b/crates/penca-storage-meta/src/purge.rs @@ -37,7 +37,8 @@ impl LifecycleManager { last_purged_aborted_seq_num: Option, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_purge_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_purge_metadata_partition(&catalog, &branch); let sql = format!( "INSERT INTO {table} \ (table_purge_uuid, branch_uuid, table_uuid, \ @@ -53,7 +54,7 @@ impl LifecycleManager { &sql, &[ SqlValue::uuid_str(table_purge_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, last_purged_commit_seq_num .map_or(SqlValue::Null(SqlType::Int64), SqlValue::Int64), @@ -75,7 +76,8 @@ impl LifecycleManager { table_purge_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_purge_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_purge_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND table_purge_uuid = $2", @@ -86,7 +88,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_purge_uuid)?, ], ) @@ -104,7 +106,8 @@ impl LifecycleManager { table_purge_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_purge_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_purge_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {table} \ WHERE branch_uuid = $1 AND table_purge_uuid = $2 \ @@ -115,7 +118,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_purge_uuid)?, ], ) @@ -180,7 +183,8 @@ impl LifecycleManager { col: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_purge_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_purge_metadata_partition(&catalog, &branch); let sql = format!( "SELECT MAX({col}) AS watermark FROM {table} \ WHERE branch_uuid = $1 \ @@ -191,10 +195,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(rows @@ -231,7 +232,8 @@ impl LifecycleManager { table_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "SELECT MAX(commit_seq_num) AS watermark FROM {table} \ WHERE branch_uuid = $1 \ @@ -242,10 +244,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(rows diff --git a/crates/penca-storage-meta/src/segment_index.rs b/crates/penca-storage-meta/src/segment_index.rs index 68bf75eb..0378008c 100644 --- a/crates/penca-storage-meta/src/segment_index.rs +++ b/crates/penca-storage-meta/src/segment_index.rs @@ -57,7 +57,8 @@ impl LifecycleManager { key_columns: Option<&[String]>, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_index_metadata_partition(&catalog, &branch); let index_uuid_val = match index_uuid { Some(u) => SqlValue::uuid_str(u)?, None => SqlValue::Null(SqlType::Uuid), @@ -82,7 +83,7 @@ impl LifecycleManager { &sql, &[ SqlValue::uuid_str(table_snapshot_index_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_uuid)?, index_uuid_val, key_columns_val, @@ -104,7 +105,8 @@ impl LifecycleManager { table_snapshot_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_index_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND table_snapshot_uuid = $2 \ @@ -116,7 +118,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_uuid)?, ], ) @@ -136,7 +138,8 @@ impl LifecycleManager { table_snapshot_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_index_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {table} \ WHERE branch_uuid = $1 AND table_snapshot_uuid = $2 \ @@ -147,7 +150,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_uuid)?, ], ) @@ -169,8 +172,9 @@ impl LifecycleManager { branch_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let index_table = naming::table_snapshot_index_metadata_table(&catalog); - let snap_table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let index_table = naming::table_snapshot_index_metadata_partition(&catalog, &branch); + let snap_table = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {index_table} tsi \ WHERE tsi.branch_uuid = $1 \ @@ -183,7 +187,7 @@ impl LifecycleManager { snap_table = qi(&snap_table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -200,7 +204,8 @@ impl LifecycleManager { table_snapshot_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_index_metadata_partition(&catalog, &branch); let sql = format!( "SELECT table_snapshot_index_uuid, index_uuid \ FROM {table} \ @@ -212,7 +217,7 @@ impl LifecycleManager { .execute_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_uuid)?, ], ) @@ -252,7 +257,8 @@ impl LifecycleManager { statistics: &[u8], ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); let sql = format!( "INSERT INTO {table} \ (segment_index_uuid, branch_uuid, segment_uuid, table_snapshot_index_uuid, \ @@ -274,7 +280,7 @@ impl LifecycleManager { &sql, &[ SqlValue::uuid_str(segment_index_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(segment_uuid)?, SqlValue::uuid_str(table_snapshot_index_uuid)?, SqlValue::Text(object_uri.to_string()), @@ -304,7 +310,8 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = segment_uuids.iter().map(String::as_str).collect(); let arr = format_sql_uuid_array(&uuid_refs); let sql = format!( @@ -315,7 +322,7 @@ impl LifecycleManager { epoch = epoch(), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -337,7 +344,8 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = segment_uuids.iter().map(String::as_str).collect(); let arr = format_sql_uuid_array(&uuid_refs); let sql = format!( @@ -347,7 +355,7 @@ impl LifecycleManager { table = qi(&table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -368,7 +376,8 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = segment_uuids.iter().map(String::as_str).collect(); let arr = format_sql_uuid_array(&uuid_refs); let sql = format!( @@ -376,7 +385,7 @@ impl LifecycleManager { table = qi(&table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -397,7 +406,8 @@ impl LifecycleManager { return Ok(Vec::new()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_index_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = segment_uuids.iter().map(String::as_str).collect(); let arr = format_sql_uuid_array(&uuid_refs); let sql = format!( @@ -409,7 +419,7 @@ impl LifecycleManager { table = qi(&table), ); let rows = driver - .execute_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(rows .iter() @@ -475,7 +485,7 @@ impl LifecycleManager { table = qi(&table), ); let rows = driver - .execute_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(rows.iter().map(|r| r.get("object_uri")).collect()) } @@ -528,7 +538,15 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_index_metadata_table(&catalog); + // Two branches in one statement: `old` rows come from the source + // branch's partition, the insert lands in this branch's. On a + // non-fork carry-forward both names resolve to the same relation, + // which the `old` alias keeps unambiguous. + let branch = parse_uuid(branch_uuid); + let source_branch = parse_uuid(source_branch_uuid); + let table = naming::table_snapshot_segment_index_metadata_partition(&catalog, &branch); + let source_table = + naming::table_snapshot_segment_index_metadata_partition(&catalog, &source_branch); // Resolve every id in Rust (xxh3 via row_uuid_for_pk) — never md5 in SQL. // The new sidecar id matches a fresh build of new_seg for this index; the // prior sidecar is found by its own id. @@ -556,7 +574,7 @@ impl LifecycleManager { old.size_bytes, old.statistics \ FROM UNNEST({new_seg_arr}, {new_sidecar_arr}, {prior_sidecar_arr}) \ AS n(new_seg, new_sidecar, prior_sidecar) \ - JOIN {table} old \ + JOIN {source_table} old \ ON old.branch_uuid = $3 \ AND old.segment_index_uuid = n.prior_sidecar \ AND old.commit_micros IS NOT NULL \ @@ -570,14 +588,15 @@ impl LifecycleManager { size_bytes = EXCLUDED.size_bytes, \ statistics = EXCLUDED.statistics", table = qi(&table), + source_table = qi(&source_table), ); driver .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(new_parent_index_uuid)?, - SqlValue::uuid_str(source_branch_uuid)?, + SqlValue::Uuid(source_branch), ], ) .await?; diff --git a/crates/penca-storage-meta/src/snapshot.rs b/crates/penca-storage-meta/src/snapshot.rs index a285398b..f1b3e179 100644 --- a/crates/penca-storage-meta/src/snapshot.rs +++ b/crates/penca-storage-meta/src/snapshot.rs @@ -33,7 +33,8 @@ impl LifecycleManager { durable: bool, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_metadata_partition(&catalog, &branch); // `table_snapshot_uuid` is deterministic from // `(catalog, branch, table, snapshotted_at)`, so retries collapse via // `DO UPDATE`. @@ -63,7 +64,7 @@ impl LifecycleManager { &sql, &[ SqlValue::uuid_str(table_snapshot_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(snapshotted_at_micros), SqlValue::TextArray(partition_keys.to_vec()), @@ -89,7 +90,8 @@ impl LifecycleManager { table_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "SELECT MAX(snapshotted_at_micros) AS last_durable FROM {table} \ WHERE branch_uuid = $1 AND table_uuid = $2 \ @@ -99,10 +101,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(rows.first().and_then(|row| { @@ -135,7 +134,8 @@ impl LifecycleManager { statistics: &[u8], ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let sql = format!( "INSERT INTO {table} \ (table_snapshot_segment_uuid, table_snapshot_uuid, branch_uuid, table_uuid, \ @@ -156,7 +156,7 @@ impl LifecycleManager { &[ SqlValue::uuid_str(table_snapshot_segment_uuid)?, SqlValue::uuid_str(table_snapshot_uuid)?, - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(chunk_idx as i64), SqlValue::Text(object_uri.to_string()), @@ -182,7 +182,8 @@ impl LifecycleManager { size_bytes: i64, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET size_bytes = $1 \ WHERE branch_uuid = $2 AND table_snapshot_segment_uuid = $3", @@ -193,7 +194,7 @@ impl LifecycleManager { &sql, &[ SqlValue::Int64(size_bytes), - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_segment_uuid)?, ], ) @@ -211,7 +212,8 @@ impl LifecycleManager { table_snapshot_segment_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND table_snapshot_segment_uuid = $2", @@ -222,7 +224,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_segment_uuid)?, ], ) @@ -240,7 +242,8 @@ impl LifecycleManager { table_snapshot_segment_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {table} \ WHERE branch_uuid = $1 AND table_snapshot_segment_uuid = $2 \ @@ -251,7 +254,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_segment_uuid)?, ], ) @@ -269,7 +272,8 @@ impl LifecycleManager { table_snapshot_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "DELETE FROM {table} \ WHERE branch_uuid = $1 AND table_snapshot_uuid = $2 \ @@ -280,7 +284,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_uuid)?, ], ) @@ -298,7 +302,8 @@ impl LifecycleManager { table_snapshot_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "UPDATE {table} SET commit_micros = {epoch} \ WHERE branch_uuid = $1 AND table_snapshot_uuid = $2", @@ -309,7 +314,7 @@ impl LifecycleManager { .execute_no_result_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_snapshot_uuid)?, ], ) @@ -338,7 +343,7 @@ impl LifecycleManager { seg_table = qi(&seg_name), ); let rows = driver - .execute_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(rows .iter() @@ -361,8 +366,9 @@ impl LifecycleManager { table_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let seg_name = naming::table_snapshot_segment_metadata_table(&catalog); - let snap_name = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let seg_name = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); + let snap_name = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "SELECT seg.table_snapshot_segment_uuid, seg.object_uri \ FROM {seg_table} seg \ @@ -378,10 +384,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(rows @@ -416,8 +419,9 @@ impl LifecycleManager { table_uuid: &str, ) -> Result> { let catalog = parse_uuid(catalog_uuid); - let seg_name = naming::table_snapshot_segment_metadata_table(&catalog); - let snap_name = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let seg_name = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); + let snap_name = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = format!( "SELECT seg.table_snapshot_segment_uuid, seg.object_uri \ FROM {seg_table} seg \ @@ -440,10 +444,7 @@ impl LifecycleManager { let rows = driver .execute_params( &sql, - &[ - SqlValue::uuid_str(table_uuid)?, - SqlValue::uuid_str(branch_uuid)?, - ], + &[SqlValue::uuid_str(table_uuid)?, SqlValue::Uuid(branch)], ) .await?; Ok(rows @@ -469,7 +470,8 @@ impl LifecycleManager { table_snapshot_segment_uuids: &[String], ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = table_snapshot_segment_uuids .iter() .map(String::as_str) @@ -482,7 +484,7 @@ impl LifecycleManager { table = qi(&table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -542,7 +544,13 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + // The one two-branch statement in this module: rows are read from the + // source branch's partition and written into this branch's. + let branch = parse_uuid(branch_uuid); + let source_branch = parse_uuid(source_branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); + let source_table = + naming::table_snapshot_segment_metadata_partition(&catalog, &source_branch); let new_uuids: Vec<&str> = specs.iter().map(|s| s.new_seg_uuid_str.as_str()).collect(); let prior_uuids: Vec<&str> = specs @@ -564,7 +572,7 @@ impl LifecycleManager { old.size_bytes, old.format, old.metadata, old.statistics \ FROM UNNEST({new_arr}, {idx_arr}, {prior_arr}) \ AS new(uuid, idx, old_uuid) \ - JOIN {table} old \ + JOIN {source_table} old \ ON old.table_snapshot_segment_uuid = new.old_uuid \ AND old.branch_uuid = $3 \ ON CONFLICT (branch_uuid, table_snapshot_segment_uuid) DO UPDATE \ @@ -580,6 +588,7 @@ impl LifecycleManager { statistics = EXCLUDED.statistics \ RETURNING table_snapshot_segment_uuid", table = qi(&table), + source_table = qi(&source_table), ); // RETURNING + a row-count check turns a non-joining prior uuid // (a stale/wrong spec, or a prior row retired between the read @@ -592,8 +601,8 @@ impl LifecycleManager { &sql, &[ SqlValue::uuid_str(table_snapshot_uuid)?, - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(source_branch_uuid)?, + SqlValue::Uuid(branch), + SqlValue::Uuid(source_branch), ], ) .await?; @@ -623,7 +632,8 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = table_snapshot_segment_uuids .iter() .map(String::as_str) @@ -637,7 +647,7 @@ impl LifecycleManager { epoch = epoch(), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -662,7 +672,8 @@ impl LifecycleManager { return Ok(()); } let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); let uuid_refs: Vec<&str> = table_snapshot_segment_uuids .iter() .map(String::as_str) @@ -676,7 +687,7 @@ impl LifecycleManager { table = qi(&table), ); driver - .execute_no_result_params(&sql, &[SqlValue::uuid_str(branch_uuid)?]) + .execute_no_result_params(&sql, &[SqlValue::Uuid(branch)]) .await?; Ok(()) } @@ -694,8 +705,9 @@ impl LifecycleManager { table_uuid: &str, ) -> Result<()> { let catalog = parse_uuid(catalog_uuid); - let snap_name = naming::table_snapshot_metadata_table(&catalog); - let seg_name = naming::table_snapshot_segment_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let snap_name = naming::table_snapshot_metadata_partition(&catalog, &branch); + let seg_name = naming::table_snapshot_segment_metadata_partition(&catalog, &branch); // NOT EXISTS, not NOT IN: the segment table's // `table_snapshot_uuid` is nullable, and one NULL in a NOT IN // subquery NULLs the whole predicate — silently turning this @@ -716,10 +728,7 @@ impl LifecycleManager { driver .execute_no_result_params( &sql, - &[ - SqlValue::uuid_str(branch_uuid)?, - SqlValue::uuid_str(table_uuid)?, - ], + &[SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?], ) .await?; Ok(()) @@ -752,13 +761,14 @@ impl LifecycleManager { }; let window_start = now_micros - duration_seconds * 1_000_000; let catalog = parse_uuid(catalog_uuid); - let table = naming::table_snapshot_metadata_table(&catalog); + let branch = parse_uuid(branch_uuid); + let table = naming::table_snapshot_metadata_partition(&catalog, &branch); let sql = retention_floor_select(&qi(&table), "$1", "$2", "$3"); let rows = driver .execute_params( &sql, &[ - SqlValue::uuid_str(branch_uuid)?, + SqlValue::Uuid(branch), SqlValue::uuid_str(table_uuid)?, SqlValue::Int64(window_start), ], diff --git a/crates/penca-storage-meta/tests/retention_floor.rs b/crates/penca-storage-meta/tests/retention_floor.rs index 743f576c..fc064b01 100644 --- a/crates/penca-storage-meta/tests/retention_floor.rs +++ b/crates/penca-storage-meta/tests/retention_floor.rs @@ -46,7 +46,7 @@ async fn seed_snapshot( durable: bool, committed: bool, ) { - let table = naming::table_snapshot_metadata_table(catalog_uuid); + let table = naming::table_snapshot_metadata_partition(catalog_uuid, branch_uuid); let sql = format!( "INSERT INTO {tbl} \ (table_snapshot_uuid, branch_uuid, table_uuid, snapshotted_at_micros, \ diff --git a/packages/penca-client/src/penca_client/naming.py b/packages/penca-client/src/penca_client/naming.py index ec1b3832..13e9f4b7 100644 --- a/packages/penca-client/src/penca_client/naming.py +++ b/packages/penca-client/src/penca_client/naming.py @@ -44,7 +44,8 @@ without state. See :func:`system_schema_uuid`, :func:`system_schemas_table_uuid`, :func:`system_tables_table_uuid`. -- **Per-branch partition leaves**: the tx-log family. Each leaf's +- **Per-branch partition leaves**: the tx-log family and the metadata + family (CHA-546 made both resolve by leaf name). Each leaf's ``partition_uuid`` derives directly from ``(catalog_uuid, branch_uuid, partition_tag)``, where ``partition_tag`` is the fixed PG-name suffix (e.g. ``"commit_tx_log"``). @@ -222,6 +223,22 @@ def tx_table_log_partition(catalog_uuid: str, branch_uuid: str) -> str: return f"{partition_uuid}_tx_table_log_partition" +def table_snapshot_segment_metadata_partition( + catalog_uuid: str, branch_uuid: str +) -> str: + """Partition of table_snapshot_segment_metadata for a branch (CHA-546). + + Same leaf shape as the tx-log family above. Tests that count PG + statements by relation name need this rather than the parent: since + CHA-546 the read path names the leaf, so a parent-name needle matches + nothing and an ``== 0`` assertion passes vacuously. + """ + partition_uuid = row_uuid_for_pk( + catalog_uuid, [branch_uuid, TABLE_SNAPSHOT_SEGMENT_METADATA] + ) + return f"{partition_uuid}_{TABLE_SNAPSHOT_SEGMENT_METADATA}_partition" + + def deterministic_uuid_from(*parts: str) -> str: """Generic deterministic UUID combiner. diff --git a/tests/integration/integration_cha546_partition_lock_footprint_test.py b/tests/integration/integration_cha546_partition_lock_footprint_test.py new file mode 100644 index 00000000..7c04b735 --- /dev/null +++ b/tests/integration/integration_cha546_partition_lock_footprint_test.py @@ -0,0 +1,363 @@ +"""CHA-546 red test: branch ops must take no lock on a metadata parent. + +The 8 metadata tables are LIST-partitioned by ``branch_uuid``, but their +call sites named the catalog-wide parent and let Postgres route. Naming a +parent takes a lock on the parent; naming a partition takes none. That +difference costs twice: + +* a writer holding ``RowExclusiveLock`` on the parent deadlocks branch + teardown, which walks leaf-to-parent while the writer walks + parent-to-leaf; +* compact's ``enumerate_unsealed_persist_segments_for_scope`` opens with + ``SELECT ... FOR UPDATE OF seg`` on the ``table_persist_segment_metadata`` + parent and holds ``ROW SHARE`` there across its whole cold read and + merged write. Teardown's ``DROP TABLE`` needs ``ACCESS EXCLUSIVE`` on + that same parent under a transaction-scoped 5s ``lock_timeout``, so + ``DeleteBranch`` fails ``Aborted`` whenever any compact in the catalog + has been mid-merge for more than five seconds. That is ordinary + operation, not a race — and it is a *read* that causes it. + +The fixture holds locks on ``ONLY`` the 8 parents from an out-of-band +session — what teardown holds against every branch other than the one it +is deleting — and requires branch-scoped operations to finish anyway. + +**Two lock modes, because the fix does not reach equally far.** Naming a +partition removes the parent lock for reads outright, but not for writes. +Measured directly against PG 17 (``pg_locks`` for the issuing backend, +inside its transaction, statement issued against the LEAF): + +=============================== =========================== +statement on a leaf lock taken on its parent +=============================== =========================== +``SELECT`` none +``SELECT ... FOR UPDATE`` none +``DELETE`` none +``INSERT`` ``AccessShare``, held to commit +``UPDATE`` ``AccessShare``, held to commit +=============================== =========================== + +The write rows are not avoidable and not a leftover parent-naming bug: +evaluating the leaf's partition constraint opens the parent for its +partition key, so any statement that produces a new row tuple takes +``AccessShare`` there. ``DELETE`` and every read produce none, so they take +nothing — which is why the read half of the fix is total and the write half +is a downgrade from ``RowExclusive`` to ``AccessShare``. + +That downgrade is the whole point for teardown, because the two conflict +differently. ``lock_branch_teardown_partitions`` takes ``EXCLUSIVE``, which +conflicts with ``RowExclusive`` but **not** with ``AccessShare``. So: + +* ``EXCLUSIVE`` held on the parents — writes must proceed. Red before the + fix (the writer's parent ``RowExclusive`` conflicts), green after. +* ``ACCESS EXCLUSIVE`` held on the parents — reads must proceed. Red before + the fix (the reader's parent ``AccessShare`` conflicts), green after, + since reads now take nothing on the parent at all. + +Asserting writes under ``ACCESS EXCLUSIVE`` would be asserting something +Postgres cannot give, and is why that residual is documented on +``write::delete_branch`` rather than tested away here: a ``DROP TABLE`` +still needs ``ACCESS EXCLUSIVE`` on the parent and still waits behind a +concurrent writer's ``AccessShare``. + +Rejected alternatives, so a later reader does not "fix" this back into one: + +* **Race a real compact against DeleteBranch.** Needs a compact to stay + mid-merge for over five seconds to trip the timeout. Inherently flaky. +* **Simulate compact's lock with a raw ``SELECT ... FOR UPDATE`` on the + partition.** That passes before the fix, so it is not a red test. +* **One fixture at ``ACCESS EXCLUSIVE`` for both halves.** What this file + did first. It fails on an idle stack for the write half, for the + Postgres reason above — a green that no correct implementation can earn. + +Setup runs to completion *before* the locks are taken: catalog, schema, +table, and branch creation are DDL, and DDL names parents by design. + +Run via ``just integration-test cha546_partition_lock_footprint``. +""" + +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import TimeoutError as FutureTimeout + +import pyarrow as pa +import pytest +from penca_client.naming import ( + COMPACT_SEGMENT_METADATA, + TABLE_PERSIST_METADATA, + TABLE_PERSIST_SEGMENT_METADATA, + TABLE_PURGE_METADATA, + TABLE_SNAPSHOT_INDEX_METADATA, + TABLE_SNAPSHOT_METADATA, + TABLE_SNAPSHOT_SEGMENT_INDEX_METADATA, + TABLE_SNAPSHOT_SEGMENT_METADATA, +) +from psycopg.sql import SQL, Identifier + +from .integration_helpers import ( + USER_SCHEMA, + make_lock_driver, + setup_partitioned_table, + write_and_persist, + write_cycle, +) + +# The 8 partitioned metadata tables. Excludes +# ``tx_log_persist_segment_metadata`` (CHA-507) and ``segment_delete_set`` +# (CHA-531), which are catalog-wide and unpartitioned by design. +METADATA_PARENT_TAGS = ( + TABLE_PERSIST_METADATA, + TABLE_PERSIST_SEGMENT_METADATA, + TABLE_PURGE_METADATA, + TABLE_SNAPSHOT_METADATA, + TABLE_SNAPSHOT_SEGMENT_METADATA, + COMPACT_SEGMENT_METADATA, + TABLE_SNAPSHOT_INDEX_METADATA, + TABLE_SNAPSHOT_SEGMENT_INDEX_METADATA, +) + +# Generous relative to any of these ops on an idle stack, so a failure means +# "blocked on a lock", not "slow". +OP_DEADLINE_S = 20.0 + +# The holder must win its own locks first. A branch-scoped statement that +# names a parent can make even this contended, so failing here is the same +# defect surfacing one step earlier — the message says so. `lock_timeout` is +# per *statement*, so this bounds the whole acquisition only because all 8 +# parents are locked by a single `LOCK TABLE`. +HOLDER_ACQUIRE_TIMEOUT_S = 30.0 + +_SEED = pa.table({"name": ["alice", "bob"], "value": [1, 2]}, schema=USER_SCHEMA) +_MORE = pa.table({"name": ["carol", "dave"], "value": [3, 4]}, schema=USER_SCHEMA) + +# Composed rather than interpolated: `SQL` accepts only literals, and a lookup +# keeps the mode out of the statement text entirely. +_LOCK_MODES = { + "EXCLUSIVE": SQL("EXCLUSIVE"), + "ACCESS EXCLUSIVE": SQL("ACCESS EXCLUSIVE"), +} + + +class _ParentLockHolder: + """Holds ``mode`` on every metadata parent in one transaction. + + ``mode`` is ``EXCLUSIVE`` for the write half and ``ACCESS EXCLUSIVE`` for + the read half — see the module docstring for why the two halves cannot + share one mode. + + ``ONLY`` is what makes this a valid model of teardown and a test that can + actually go green: without it Postgres locks the named table *and every + descendant*, so the fixture would hold the branch's own leaves and block a + partition-targeted statement too. Teardown locks the deleted branch's + leaves plus the parent descriptor — never a sibling branch's leaves — so + parents-only is the state under test. + + All 8 are locked by a single statement because ``lock_timeout`` is + per-statement: eight statements would let acquisition run to 8× the bound. + + Owns its own single connection rather than borrowing ``get_pg_driver()``'s + shared pool: the locks must live exactly as long as this transaction, and a + pooled connection handed back mid-hold would carry them to another caller. + """ + + def __init__(self, catalog_uuid: str, mode: str) -> None: + self._mode = mode + self._parents = [f"{catalog_uuid}_{tag}" for tag in METADATA_PARENT_TAGS] + self._driver = make_lock_driver() + self._held = threading.Event() + self._release = threading.Event() + # `_held` is also set on the failure path so `__enter__` never hangs, so + # it alone does not mean the locks are held — this does. + self._acquired = False + self._error: BaseException | None = None + self._thread = threading.Thread(target=self._run, daemon=True) + + def _run(self) -> None: + try: + with self._driver.transaction() as tx: + tx.execute_no_result( + f"SET LOCAL lock_timeout = '{int(HOLDER_ACQUIRE_TIMEOUT_S)}s'" + ) + tx.execute_no_result( + SQL("LOCK TABLE {tbls} IN {mode} MODE").format( + tbls=SQL(", ").join( + SQL("ONLY {tbl}").format(tbl=Identifier(parent)) + for parent in self._parents + ), + mode=_LOCK_MODES[self._mode], + ) + ) + + self._acquired = True + self._held.set() + self._release.wait() + raise _Rollback() + except _Rollback: + pass + except BaseException as exc: # noqa: BLE001 - surfaced to the test thread + self._error = exc + finally: + self._held.set() + self._driver.close() + + def __enter__(self) -> _ParentLockHolder: + self._thread.start() + signalled = self._held.wait(timeout=HOLDER_ACQUIRE_TIMEOUT_S + 5.0) + if not signalled or not self._acquired: + # Silently proceeding here would run both tests with no locks held + # and report green — the worst outcome for a red test. + self._release.set() + raise AssertionError( + f"could not take {self._mode} on the metadata parents " + f"({self._parents}). Something else is holding a lock on them — " + "which is itself the CHA-546 defect, one step earlier." + ) from self._error + + return self + + def __exit__(self, *_exc) -> None: + self._release.set() + self._thread.join(timeout=30.0) + + +class _Rollback(Exception): + """Unwinds the holder's transaction without committing.""" + + +def _within_deadline(label: str, mode: str, fn, *args, **kwargs): + """Run ``fn`` on a worker thread and fail if it does not return in time. + + The client exposes no per-call deadline, so the timeout lives here. The + worker stays blocked on the RPC after a timeout; the holder's ``__exit__`` + releases the locks and lets it drain, which is why the executor is not + shut down with ``wait=True``. + """ + executor = ThreadPoolExecutor(max_workers=1) + try: + future = executor.submit(fn, *args, **kwargs) + try: + return future.result(timeout=OP_DEADLINE_S) + except FutureTimeout: + pytest.fail( + f"{label} did not complete within {OP_DEADLINE_S}s while another " + f"session held {mode} on the 8 metadata parents. A branch-scoped " + "statement is still naming a parent instead of the branch's " + "partition (CHA-546)." + ) + finally: + executor.shutdown(wait=False) + + +@pytest.fixture(scope="module") +def seeded_branch(): + """Catalog + partitioned table + a branch with persist and snapshot state. + + Module-scoped: the setup is DDL-heavy and identical for both tests. The + write test does mutate what the read test reads — it persists, snapshots + and compacts this table — but the read test only asserts ``num_rows > 0``, + which those additions cannot falsify. Keep any new assertion here monotone + in the same way, or make the fixture function-scoped. + """ + client, catalog_uuid, schema_uuid, table_uuid, main_branch_uuid = ( + setup_partitioned_table("cha546_lockfoot") + ) + write_cycle( + client, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=main_branch_uuid, + upserts=_SEED, + ) + # Leaves an unsealed persist tail so compact has segments to merge. + write_and_persist( + client, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=main_branch_uuid, + upserts=_MORE, + ) + return client, catalog_uuid, schema_uuid, table_uuid, main_branch_uuid + + +class TestParentLockFootprint: + def test_branch_writes_proceed_under_parent_exclusive(self, seeded_branch): + """Cost 1: the write path must not contend with teardown. + + ``EXCLUSIVE`` is exactly what ``lock_branch_teardown_partitions`` takes, + and is the strongest mode a writer can be asked to clear: its parent + ``AccessShare`` is compatible, its pre-fix ``RowExclusive`` was not. + """ + client, catalog_uuid, schema_uuid, table_uuid, branch_uuid = seeded_branch + mode = "EXCLUSIVE" + + with _ParentLockHolder(catalog_uuid, mode): + _within_deadline( + "write -> commit -> persist", + mode, + write_and_persist, + client, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=branch_uuid, + upserts=pa.table({"name": ["erin"], "value": [5]}, schema=USER_SCHEMA), + ) + _within_deadline( + "snapshot", + mode, + client.snapshot, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=branch_uuid, + ) + # enumerate_unsealed_persist_segments_for_scope's + # `SELECT ... FOR UPDATE OF seg` — the site that makes DeleteBranch + # fail Aborted in steady state. It reads under this mode, but its + # merged write is what needs the parent to be at most AccessShare. + _within_deadline( + "compact_persist_segments", + mode, + client.compact_persist_segments, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=branch_uuid, + ) + _within_deadline( + "purge", + mode, + client.purge, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=branch_uuid, + ) + + def test_branch_reads_proceed_under_parent_access_exclusive(self, seeded_branch): + """Cost 2: the read path — the larger of the two, and the ticket's point. + + ``ACCESS EXCLUSIVE`` conflicts with every mode there is, so a read + completing under it is the strongest available statement: the read path + takes NO lock on a metadata parent, not merely a compatible one. + """ + client, catalog_uuid, schema_uuid, table_uuid, branch_uuid = seeded_branch + mode = "ACCESS EXCLUSIVE" + + with _ParentLockHolder(catalog_uuid, mode): + # meta_plan.rs: phase_one_fence_and_existence, + # read_and_classify_persist_segments, hot_min_and_snapshot_pick. + result = _within_deadline( + "read_data", + mode, + client.read_data, + catalog_uuid=catalog_uuid, + schema_uuid=schema_uuid, + table_uuid=table_uuid, + branch_uuid=branch_uuid, + ) + + assert result.num_rows > 0, "read returned no rows — setup did not seed" diff --git a/tests/integration/integration_snapshot_list_cache_test.py b/tests/integration/integration_snapshot_list_cache_test.py index 96329d0c..de674804 100644 --- a/tests/integration/integration_snapshot_list_cache_test.py +++ b/tests/integration/integration_snapshot_list_cache_test.py @@ -11,8 +11,9 @@ Counting is via ``pg_stat_statements`` (the CHA-367 resolution-count seam): ``count_stmts_referencing`` sums ``calls`` over normalized statements whose text -contains the per-catalog ``…_table_snapshot_segment_metadata`` identifier, so -background activity on other catalogs can't pollute the count. +contains this branch's ``…_table_snapshot_segment_metadata_partition`` leaf name +(CHA-546 — the read path names the leaf, never the catalog-wide parent), so +background activity on other branches or catalogs can't pollute the count. Run: ``just integration-test --test-arg integration_snapshot_list_cache_test``. """ @@ -23,7 +24,7 @@ import pytest from penca_client import Mutation from penca_client._time import micros_to_datetime -from penca_client.naming import TABLE_SNAPSHOT_SEGMENT_METADATA +from penca_client.naming import table_snapshot_segment_metadata_partition from .integration_helpers import ( USER_SCHEMA, @@ -77,9 +78,11 @@ def test_snapshot_list_cache_hit_cuts_pg_read(self): pg = get_pg_driver() ensure_pg_stat_statements(pg) - # Per-catalog needle: pg_stat_statements preserves identifiers, so this - # matches only this catalog's snapshot-segment-metadata reads. - seg_table = f"{cat}_{TABLE_SNAPSHOT_SEGMENT_METADATA}" + # Per-BRANCH needle: pg_stat_statements preserves identifiers, and since + # CHA-546 the read path names the branch's partition, not the catalog-wide + # parent. A parent-name needle here would match nothing, silently turning + # every `== 0` assertion below into a tautology. + seg_table = table_snapshot_segment_metadata_partition(cat, br) # First current-time read — cache miss, populates the entry. reset_pg_stat(pg) @@ -147,7 +150,7 @@ def test_time_travel_read_shares_wsnap_cache_entry(self): pg = get_pg_driver() ensure_pg_stat_statements(pg) - seg_table = f"{cat}_{TABLE_SNAPSHOT_SEGMENT_METADATA}" + seg_table = table_snapshot_segment_metadata_partition(cat, br) # Warm the cache, then confirm a current-time read is a hit AND sees carol. client.read_data( @@ -220,7 +223,7 @@ def test_distinct_snapshots_get_distinct_wsnap_entries(self): pg = get_pg_driver() ensure_pg_stat_statements(pg) - seg_table = f"{cat}_{TABLE_SNAPSHOT_SEGMENT_METADATA}" + seg_table = table_snapshot_segment_metadata_partition(cat, br) # Warm the cache on the LATEST snapshot (S2 → its own W_snap key). current = client.read_data( @@ -460,7 +463,7 @@ class TestSystemTableResolveCache: itself COLD via ``_persist_purge_system_tables_past_grace`` so the resolve must consult a snapshot segment list — the read the W_snap-keyed snapshot-list cache (CHA-472/492) serves from cache on both the query and - write paths. Same per-catalog ``…_table_snapshot_segment_metadata`` needle. + write paths. Same per-branch partition-leaf needle. """ def test_system_table_resolve_cache_hit_cuts_pg_read(self): @@ -486,7 +489,7 @@ def test_system_table_resolve_cache_hit_cuts_pg_read(self): pg = get_pg_driver() ensure_pg_stat_statements(pg) - seg_table = f"{cat}_{TABLE_SNAPSHOT_SEGMENT_METADATA}" + seg_table = table_snapshot_segment_metadata_partition(cat, br) # 1st read — warms the system-table snapshot-list cache entry. reset_pg_stat(pg) @@ -536,7 +539,7 @@ def test_write_path_resolve_cache_hit_cuts_pg_read(self): pg = get_pg_driver() ensure_pg_stat_statements(pg) - seg_table = f"{cat}_{TABLE_SNAPSHOT_SEGMENT_METADATA}" + seg_table = table_snapshot_segment_metadata_partition(cat, br) def _autocommit_write(name, value): client.write_data( @@ -596,7 +599,7 @@ def test_system_table_resolve_time_travel_shares_wsnap_cache(self): pg = get_pg_driver() ensure_pg_stat_statements(pg) - seg_table = f"{cat}_{TABLE_SNAPSHOT_SEGMENT_METADATA}" + seg_table = table_snapshot_segment_metadata_partition(cat, br) # Warm the system-table cache via a current-time read. client.read_data( diff --git a/tests/static/static_cha546_partition_naming_test.py b/tests/static/static_cha546_partition_naming_test.py new file mode 100644 index 00000000..bd6d053a --- /dev/null +++ b/tests/static/static_cha546_partition_naming_test.py @@ -0,0 +1,209 @@ +"""CHA-546: every branch-scoped statement targets the branch's partition. + +The 6 tx-log tables already resolve by partition name. The 8 metadata +tables did not — call sites named the catalog-wide parent and let +Postgres route on ``branch_uuid``. Naming the parent takes a lock on the +parent, so a writer deadlocks branch teardown (teardown walks +leaf-to-parent, the writer parent-to-leaf) and compact's ``SELECT ... FOR +UPDATE OF seg`` holds ``ROW SHARE`` on the parent across its whole cold +read and merged write — long enough that ``DeleteBranch`` trips +``lock_branch_teardown_partitions``' 5s ``lock_timeout`` in ordinary +operation. Naming a partition takes no parent lock at all for ``SELECT``, +``SELECT ... FOR UPDATE`` and ``DELETE``, and downgrades an ``INSERT`` or +``UPDATE`` from ``RowExclusive`` to a commit-held ``AccessShare`` — which +still clears teardown's ``EXCLUSIVE`` step. That is what makes this a fix +rather than a tidy-up; the measured table is in +``tests/integration/integration_cha546_partition_lock_footprint_test.py``. + +Two exceptions are enumerated, not incidental: + +* DDL in ``crates/penca-db/src/dialect/pg.rs`` must name parents to + create, attach, and drop partitions. +* ``compact.rs``'s ``segment_delete_set`` refcount gate probes three + parents catalog-wide on purpose (CHA-531) — carry-forward crosses fork + edges, so a branch-scoped probe would delete a segment another branch's + snapshot still references. + +The ``compact.rs`` allowance is scoped to the two gate functions rather +than the whole file, so the three branch-scoped sites this ticket +converted cannot come back. + +These are pure source-input checks — no Docker. They live under +``tests/static/`` (run via ``just static-test``, also wired into +``just check``). +""" + +from __future__ import annotations + +import re +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +CRATES = REPO_ROOT / "crates" + +# The 8 partitioned metadata tables' PARENT-name helpers. Their +# ``*_partition`` counterparts all exist in +# ``crates/penca-core/src/naming/tables.rs`` and are what call sites use. +# Deliberately excludes ``tx_log_persist_segment_metadata`` (CHA-507) and +# ``segment_delete_set`` (CHA-531) — both are catalog-wide and +# unpartitioned, so they have no partition to target. +PARENT_HELPERS = ( + "table_persist_metadata_table", + "table_persist_segment_metadata_table", + "table_purge_metadata_table", + "table_snapshot_metadata_table", + "table_snapshot_segment_metadata_table", + "compact_segment_metadata_table", + "table_snapshot_index_metadata_table", + "table_snapshot_segment_index_metadata_table", +) + +_PARENT_RE = re.compile(r"\b(" + "|".join(PARENT_HELPERS) + r")\b") +_FN_RE = re.compile( + r"\b(?:pub(?:\([^)]*\))?\s+)?(?:async\s+)?fn\s+([A-Za-z_][A-Za-z0-9_]*)" +) + +# Files where naming a parent is the point. +UNRESTRICTED = frozenset( + { + "crates/penca-core/src/naming/tables.rs", # the definitions + "crates/penca-core/src/naming/mod.rs", # the re-exports + "crates/penca-db/src/dialect/pg.rs", # partition DDL + } +) + +# CHA-531's catalog-wide refcount gate. Both build their ``NOT EXISTS`` +# probes through ``segment_delete_set_referenced_predicate``, which takes +# the table names as arguments rather than deriving them. +GATE_FILE = "crates/penca-storage-meta/src/compact.rs" + +# The three parents each gate function must probe. Pinning the exact set +# matters: asserting merely that *a* parent survives would let two of the +# three be narrowed to partitions while the third keeps the assertion green, +# and the forward guard exempts the whole function body — so CHA-531's bug +# would come back silently for the narrowed probes. +GATE_PARENTS = frozenset( + { + "table_snapshot_segment_metadata_table", + "table_snapshot_segment_index_metadata_table", + "table_persist_segment_metadata_table", + } +) +GATE_FUNCTIONS = frozenset( + { + "eligible_segment_delete_set_rows", + "reap_referenced_segment_delete_set_rows", + } +) + +_PARTITION_RE = re.compile(r"\b\w+_metadata_partition\b") + + +def _rust_sources() -> list[Path]: + return sorted(CRATES.rglob("*.rs")) + + +def _violations() -> list[str]: + offenders: list[str] = [] + for path in _rust_sources(): + rel = path.relative_to(REPO_ROOT).as_posix() + if rel in UNRESTRICTED: + continue + + enclosing = "" + for lineno, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): + fn_match = _FN_RE.search(line) + if fn_match: + enclosing = fn_match.group(1) + + hit = _PARENT_RE.search(line) + if not hit: + continue + + if rel == GATE_FILE and enclosing in GATE_FUNCTIONS: + continue + + offenders.append( + f"{rel}:{lineno}: {hit.group(1)} (in {enclosing or ''})" + ) + + return offenders + + +class TestPartitionDirectNaming: + def test_no_branch_scoped_statement_names_a_metadata_parent(self): + offenders = _violations() + assert not offenders, ( + f"{len(offenders)} site(s) build a metadata table name from a " + "parent-name helper. Branch-scoped statements must call the " + "matching `*_partition(&catalog, &branch)` helper instead " + "(CHA-546). Allowed: DDL in penca-db/src/dialect/pg.rs and the " + "catalog-wide segment_delete_set refcount gate in " + f"{GATE_FILE}::{{{', '.join(sorted(GATE_FUNCTIONS))}}}.\n" + + "\n".join(offenders) + ) + + def test_refcount_gate_still_probes_parents_catalog_wide(self): + # The converse guard: narrowing CHA-531's gate to a partition would + # let the sweep delete a segment a forked branch's carried-forward + # snapshot still references, and the check above would happily pass. + text = (REPO_ROOT / GATE_FILE).read_text(encoding="utf-8") + for fn in sorted(GATE_FUNCTIONS): + body = _function_body(text, fn) + assert body is not None, f"{GATE_FILE} no longer defines `{fn}`" + assert set(_PARENT_RE.findall(body)) == GATE_PARENTS, ( + f"{GATE_FILE}::{fn} must probe exactly {sorted(GATE_PARENTS)} " + "catalog-wide — the segment_delete_set refcount gate spans fork " + "edges (CHA-531). Narrowing any one probe to a partition " + "reintroduces the bug where a parent's segment is deleted out " + "from under a child's carried-forward snapshot. Found: " + f"{sorted(set(_PARENT_RE.findall(body)))}" + ) + assert not _PARTITION_RE.search(body), ( + f"{GATE_FILE}::{fn} must not name any branch partition — the " + "refcount gate is deliberately catalog-wide (CHA-531). Found: " + f"{sorted(set(_PARTITION_RE.findall(body)))}" + ) + + def test_no_open_cha546_todos(self): + # The comments in pg.rs::lock_branch_teardown_partitions and + # write/mod.rs describe the parent-naming contention as the root + # cause and defer it to this ticket. Once the conversion lands they + # describe something that no longer happens. + offenders: list[str] = [] + for path in _rust_sources(): + for lineno, line in enumerate( + path.read_text(encoding="utf-8").splitlines(), 1 + ): + if "TODO(CHA-546)" in line: + offenders.append(f"{path.relative_to(REPO_ROOT)}:{lineno}") + + assert not offenders, ( + "TODO(CHA-546) markers must be retired along with the fix:\n" + + "\n".join(offenders) + ) + + +def _function_body(text: str, name: str) -> str | None: + """Return the brace-balanced body of ``fn ``, or None if absent.""" + match = re.search( + r"\b(?:pub(?:\([^)]*\))?\s+)?(?:async\s+)?fn\s+" + re.escape(name) + r"\b", + text, + ) + if match is None: + return None + + start = text.find("{", match.end()) + if start == -1: + return None + + depth = 0 + for idx in range(start, len(text)): + if text[idx] == "{": + depth += 1 + elif text[idx] == "}": + depth -= 1 + if depth == 0: + return text[start : idx + 1] + + return None diff --git a/tests/static/static_naming_parity_test.py b/tests/static/static_naming_parity_test.py index 7c3dfa83..0665fcc0 100644 --- a/tests/static/static_naming_parity_test.py +++ b/tests/static/static_naming_parity_test.py @@ -34,6 +34,7 @@ table_persist_segment_uuid, table_persist_uuid, table_purge_uuid, + table_snapshot_segment_metadata_partition, table_snapshot_segment_uuid, table_snapshot_uuid, tx_table_log_partition, @@ -123,6 +124,16 @@ def test_commit_tx_log_seq_num_partition(self): == "4bb9308a-f277-9b8b-0631-bd6d5aa5c2f9_commit_tx_log_seq_num_partition" ) + def test_table_snapshot_segment_metadata_partition(self): + # CHA-546: the metadata family resolves by leaf name too, and this + # is the first such leaf Python has to compute. Golden mirrors the + # Rust unit test in crates/penca-core/src/naming/tables.rs. + assert ( + table_snapshot_segment_metadata_partition(CAT, BR) + == "1a93a047-8229-c30f-2de8-08a4482c2051" + "_table_snapshot_segment_metadata_partition" + ) + def test_write_sequence(self): # CHA-431: per-(table, branch) sequence; prefix = # row_uuid_for_pk(table_uuid, [branch_uuid]) — the same data-object