From 61dc84caf2e0017264bdb869b3bb7a3f25d183d7 Mon Sep 17 00:00:00 2001 From: aaltshuler Date: Sat, 4 Jul 2026 17:54:16 +0300 Subject: [PATCH] refactor(engine): retire the open_dataset_head_for_write alias MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The method was a delegating twin of open_dataset_head that discarded its table_key argument ("retained for signature stability") — a dead parameter every caller still had to supply, and a second name for the same open policy on the sealed TableStorage surface. Remove it from the trait, the TableStore impl, and all call sites, which also lets the two needs_index_work_* helpers drop their now-unused table_key parameters. Purely mechanical; no behavior change. --- crates/omnigraph/src/db/omnigraph.rs | 2 +- crates/omnigraph/src/db/omnigraph/optimize.rs | 10 +++--- crates/omnigraph/src/db/omnigraph/repair.rs | 2 +- .../src/db/omnigraph/schema_apply.rs | 5 ++- .../omnigraph/src/db/omnigraph/table_ops.rs | 33 +++++++------------ crates/omnigraph/src/exec/staging.rs | 2 +- crates/omnigraph/src/storage_layer.rs | 18 ---------- crates/omnigraph/src/table_store.rs | 20 +---------- 8 files changed, 23 insertions(+), 69 deletions(-) diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index bbff6837..48440356 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -98,7 +98,7 @@ pub struct SchemaApplyPreview { /// (each of which re-runs `ensure_schema_state_valid`). When absent (`None` — every /// non-mutate/load caller), every threaded function behaves byte-identically to /// before. The carrier never removes a version guard or changes which dataset version -/// the per-table open targets: strict ops keep `open_dataset_head_for_write` + +/// the per-table open targets: strict ops keep `open_dataset_head` + /// `ensure_expected_version`, and the commit-time OCC re-read still opens a fresh /// manifest snapshot (via `fresh_snapshot_for_branch_unchecked`) — only the redundant /// schema re-validation is dropped. diff --git a/crates/omnigraph/src/db/omnigraph/optimize.rs b/crates/omnigraph/src/db/omnigraph/optimize.rs index de652c2c..70ee8a7e 100644 --- a/crates/omnigraph/src/db/omnigraph/optimize.rs +++ b/crates/omnigraph/src/db/omnigraph/optimize.rs @@ -385,7 +385,7 @@ async fn optimize_one_table( // opaque `SnapshotHandle` via `into_dataset()` (gated to the maintenance path). let mut ds = db .storage() - .open_dataset_head_for_write(&table_key, &full_path, None) + .open_dataset_head(&full_path, None) .await? .into_dataset(); @@ -444,10 +444,10 @@ async fn optimize_one_table( // recovery and rolls back siblings). let needs_reindex = TableStore::has_unindexed_fragments(&ds).await?; let needs_index_create = if let Some(type_name) = table_key.strip_prefix("node:") { - super::table_ops::needs_index_work_node(db, type_name, &table_key, &full_path, None) + super::table_ops::needs_index_work_node(db, type_name, &full_path, None) .await? } else { - super::table_ops::needs_index_work_edge(db, &table_key, &full_path, None).await? + super::table_ops::needs_index_work_edge(db, &full_path, None).await? }; if !will_compact && !needs_reindex && !needs_index_create { if head_advanced { @@ -778,7 +778,7 @@ async fn compact_internal_table( for attempt in 0..COMPACTION_RETRY_BUDGET { let handle = db .storage() - .open_dataset_head_for_write(table_key, &uri, None) + .open_dataset_head(&uri, None) .await?; let mut ds = handle.into_dataset(); @@ -911,7 +911,7 @@ pub async fn cleanup_all_tables( // surfaced through `TableStorage` — see the optimize path // above for the same rationale. Unwrap via `into_dataset()`. let handle = storage - .open_dataset_head_for_write(&table_key, &full_path, None) + .open_dataset_head(&full_path, None) .await?; let ds = handle.into_dataset(); let before_version = keep_versions diff --git a/crates/omnigraph/src/db/omnigraph/repair.rs b/crates/omnigraph/src/db/omnigraph/repair.rs index 8e7146ac..721fed2f 100644 --- a/crates/omnigraph/src/db/omnigraph/repair.rs +++ b/crates/omnigraph/src/db/omnigraph/repair.rs @@ -171,7 +171,7 @@ pub async fn repair_all_tables(db: &Omnigraph, options: RepairOptions) -> Result // unwrap the opaque handle (mirrors optimize / cleanup). let ds = db .storage() - .open_dataset_head_for_write(&table_key, &full_path, None) + .open_dataset_head(&full_path, None) .await? .into_dataset(); let manifest_version = snapshot diff --git a/crates/omnigraph/src/db/omnigraph/schema_apply.rs b/crates/omnigraph/src/db/omnigraph/schema_apply.rs index a9638653..d7e6ce5e 100644 --- a/crates/omnigraph/src/db/omnigraph/schema_apply.rs +++ b/crates/omnigraph/src/db/omnigraph/schema_apply.rs @@ -587,7 +587,7 @@ where // of the lock-check can't quietly open the wrong HEAD here. let existing = db .storage() - .open_dataset_head_for_write(table_key, &dataset_uri, entry.table_branch.as_deref()) + .open_dataset_head(&dataset_uri, entry.table_branch.as_deref()) .await?; let staged = db.storage().stage_overwrite(&existing, batch).await?; let target_ds = db.storage().commit_staged(existing, staged).await?; @@ -857,8 +857,7 @@ pub(super) async fn ensure_snapshot_entry_head_matches( let dataset_uri = db.storage().dataset_uri(&entry.table_path); let ds = db .storage() - .open_dataset_head_for_write( - &entry.table_key, + .open_dataset_head( &dataset_uri, entry.table_branch.as_deref(), ) diff --git a/crates/omnigraph/src/db/omnigraph/table_ops.rs b/crates/omnigraph/src/db/omnigraph/table_ops.rs index 72cdb086..b3c3a69a 100644 --- a/crates/omnigraph/src/db/omnigraph/table_ops.rs +++ b/crates/omnigraph/src/db/omnigraph/table_ops.rs @@ -58,7 +58,7 @@ pub(super) async fn failpoint_publish_table_head_without_index_rebuild_for_test( let full_path = format!("{}/{}", db.root_uri, entry.table_path); let ds = db .storage() - .open_dataset_head_for_write(table_key, &full_path, table_branch) + .open_dataset_head(&full_path, table_branch) .await?; let state = db.storage().table_state(&full_path, &ds).await?; let update = crate::db::SubTableUpdate { @@ -118,14 +118,7 @@ pub(super) async fn ensure_indices_for_branch( continue; } let full_path = format!("{}/{}", db.root_uri, entry.table_path); - if needs_index_work_node( - db, - type_name, - &table_key, - &full_path, - entry.table_branch.as_deref(), - ) - .await? + if needs_index_work_node(db, type_name, &full_path, entry.table_branch.as_deref()).await? { recovery_pins.push(crate::db::manifest::SidecarTablePin { table_key, @@ -154,7 +147,7 @@ pub(super) async fn ensure_indices_for_branch( continue; } let full_path = format!("{}/{}", db.root_uri, entry.table_path); - if needs_index_work_edge(db, &table_key, &full_path, entry.table_branch.as_deref()).await? { + if needs_index_work_edge(db, &full_path, entry.table_branch.as_deref()).await? { recovery_pins.push(crate::db::manifest::SidecarTablePin { table_key, table_path: full_path, @@ -226,7 +219,7 @@ pub(super) async fn ensure_indices_for_branch( }, None => ( db.storage() - .open_dataset_head_for_write(&table_key, &full_path, None) + .open_dataset_head(&full_path, None) .await?, None, ), @@ -274,7 +267,7 @@ pub(super) async fn ensure_indices_for_branch( }, None => ( db.storage() - .open_dataset_head_for_write(&table_key, &full_path, None) + .open_dataset_head(&full_path, None) .await?, None, ), @@ -405,13 +398,12 @@ async fn vector_column_trainable( pub(super) async fn needs_index_work_node( db: &Omnigraph, type_name: &str, - table_key: &str, full_path: &str, table_branch: Option<&str>, ) -> Result { let ds = db .storage() - .open_dataset_head_for_write(table_key, full_path, table_branch) + .open_dataset_head(full_path, table_branch) .await?; // Empty tables are skipped by the ensure_indices loop, so they must // not be pinned in the sidecar — pinning a table that produces zero @@ -479,13 +471,12 @@ pub(super) async fn needs_index_work_node( /// way node tables are; see `needs_index_work_node`. pub(super) async fn needs_index_work_edge( db: &Omnigraph, - table_key: &str, full_path: &str, table_branch: Option<&str>, ) -> Result { let ds = db .storage() - .open_dataset_head_for_write(table_key, full_path, table_branch) + .open_dataset_head(full_path, table_branch) .await?; if db.storage().count_rows(&ds, None).await? == 0 { return Ok(false); @@ -581,7 +572,7 @@ pub(super) async fn open_for_mutation_on_branch( // publisher's CAS fence), which is exactly the pinned base version. The base // already validated the schema contract once, and the staging reopen // (`reopen_for_mutation`) plus the publisher CAS in `commit_all` are the real - // drift guards. So skip `open_dataset_head_for_write` entirely and source the + // drift guards. So skip `open_dataset_head` entirely and source the // expected version from the pinned entry. // // Gated on `txn.is_some()`: without a txn (branch merge's `open_for_mutation`) @@ -620,7 +611,7 @@ pub(super) async fn open_for_mutation_on_branch( None => { let ds = db .storage() - .open_dataset_head_for_write(table_key, &full_path, None) + .open_dataset_head(&full_path, None) .await?; if op_kind.strict_pre_stage_version_check() { db.storage() @@ -669,7 +660,7 @@ pub(super) async fn open_owned_dataset_for_branch_write( Some(branch) if branch == active_branch => { let ds = db .storage() - .open_dataset_head_for_write(table_key, full_path, Some(active_branch)) + .open_dataset_head(full_path, Some(active_branch)) .await?; if op_kind.strict_pre_stage_version_check() { db.storage() @@ -711,7 +702,7 @@ pub(super) async fn open_owned_dataset_for_branch_write( .await?; let ds = db .storage() - .open_dataset_head_for_write(table_key, full_path, Some(active_branch)) + .open_dataset_head(full_path, Some(active_branch)) .await?; if op_kind.strict_pre_stage_version_check() { db.storage() @@ -931,7 +922,7 @@ pub(super) async fn reopen_for_mutation( // [`crate::db::MutationOpKind`] for the policy rationale. let _ = expected_version; db.storage() - .open_dataset_head_for_write(table_key, full_path, table_branch) + .open_dataset_head(full_path, table_branch) .await } } diff --git a/crates/omnigraph/src/exec/staging.rs b/crates/omnigraph/src/exec/staging.rs index 1aa21bcf..2f865f37 100644 --- a/crates/omnigraph/src/exec/staging.rs +++ b/crates/omnigraph/src/exec/staging.rs @@ -773,7 +773,7 @@ impl StagedMutation { // already-open staged handle (the #2 staging open) WITHOUT a fresh // `Dataset::open` — the same cheap live-HEAD probe // `ManifestCoordinator::probe_latest_version` uses. This replaces a - // redundant `open_dataset_head_for_write` (RFC-013 step 3b, collapse + // redundant `open_dataset_head` (RFC-013 step 3b, collapse // #3): the drift comparison below is byte-identical; only how `head` // is obtained changes (probe vs cold open). let head = entry diff --git a/crates/omnigraph/src/storage_layer.rs b/crates/omnigraph/src/storage_layer.rs index fb4d34b7..250937ac 100644 --- a/crates/omnigraph/src/storage_layer.rs +++ b/crates/omnigraph/src/storage_layer.rs @@ -233,13 +233,6 @@ pub trait TableStorage: sealed::Sealed + Send + Sync + Debug { branch: Option<&str>, ) -> Result; - async fn open_dataset_head_for_write( - &self, - table_key: &str, - dataset_uri: &str, - branch: Option<&str>, - ) -> Result; - async fn fork_branch_from_state( &self, dataset_uri: &str, @@ -497,17 +490,6 @@ impl TableStorage for TableStore { .map(SnapshotHandle::new) } - async fn open_dataset_head_for_write( - &self, - table_key: &str, - dataset_uri: &str, - branch: Option<&str>, - ) -> Result { - TableStore::open_dataset_head_for_write(self, table_key, dataset_uri, branch) - .await - .map(SnapshotHandle::new) - } - async fn fork_branch_from_state( &self, dataset_uri: &str, diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs index 166b3733..b149bf2e 100644 --- a/crates/omnigraph/src/table_store.rs +++ b/crates/omnigraph/src/table_store.rs @@ -222,22 +222,6 @@ impl TableStore { } } - pub async fn open_dataset_head_for_write( - &self, - table_key: &str, - dataset_uri: &str, - branch: Option<&str>, - ) -> Result { - // RFC-013 step 3a: open writes via the direct opener (O(1)) instead of the - // lance-namespace builder, which re-resolved the table's version chain - // O(depth) per write. The namespace is a catalog/discovery layer, not a - // per-open hot-path component (RFC §2.4); the manifest already holds the - // location, and `ensure_expected_version` still validates head == pinned - // for strict ops. `table_key` retained for signature stability. - let _ = table_key; - self.open_dataset_head(dataset_uri, branch).await - } - pub async fn delete_branch(&self, dataset_uri: &str, branch: &str) -> Result<()> { let mut ds = crate::instrumentation::open_dataset( dataset_uri, @@ -327,9 +311,7 @@ impl TableStore { table_key: &str, expected_version: u64, ) -> Result { - let ds = self - .open_dataset_head_for_write(table_key, dataset_uri, branch) - .await?; + let ds = self.open_dataset_head(dataset_uri, branch).await?; self.ensure_expected_version(&ds, table_key, expected_version)?; Ok(ds) }