mirror of
https://github.com/ModernRelay/omnigraph.git
synced 2026-07-12 03:12:11 +02:00
refactor(engine): retire the open_dataset_head_for_write alias
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.
This commit is contained in:
parent
9292ff42a9
commit
61dc84caf2
8 changed files with 23 additions and 69 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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<bool> {
|
||||
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<bool> {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -233,13 +233,6 @@ pub trait TableStorage: sealed::Sealed + Send + Sync + Debug {
|
|||
branch: Option<&str>,
|
||||
) -> Result<SnapshotHandle>;
|
||||
|
||||
async fn open_dataset_head_for_write(
|
||||
&self,
|
||||
table_key: &str,
|
||||
dataset_uri: &str,
|
||||
branch: Option<&str>,
|
||||
) -> Result<SnapshotHandle>;
|
||||
|
||||
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<SnapshotHandle> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<Dataset> {
|
||||
// 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<Dataset> {
|
||||
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)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue