diff --git a/crates/omnigraph/src/blob.rs b/crates/omnigraph/src/blob.rs index fea5288ff..9f1897458 100644 --- a/crates/omnigraph/src/blob.rs +++ b/crates/omnigraph/src/blob.rs @@ -1077,7 +1077,9 @@ impl Omnigraph { .await? }); let actual_table_version = dataset.version().version; - if actual_table_version != expected_table_version { + // RFC 0067: a pending pin resolves to its staged version. + let pending_staged = entry.version_metadata.staged_version() == Some(actual_table_version); + if actual_table_version != expected_table_version && !pending_staged { return Err(OmniError::blob_integrity(format!( "selected dataset for {} type '{}' opened at Lance version {}, expected published dataset version {}", entity_label(cell.entity), @@ -1124,9 +1126,7 @@ impl Omnigraph { ))); } } - if let Some(expected_e_tag) = entry.version_metadata.e_tag() - && dataset.manifest_location().e_tag.as_deref() != Some(expected_e_tag) - { + if !entry.version_metadata.witnesses(&dataset) { return Err(OmniError::blob_integrity(format!( "selected dataset for {} type '{}' opened a different Lance manifest incarnation at published dataset version {}", entity_label(cell.entity), diff --git a/crates/omnigraph/src/changes/mod.rs b/crates/omnigraph/src/changes/mod.rs index ebaa25dde..24e40f39b 100644 --- a/crates/omnigraph/src/changes/mod.rs +++ b/crates/omnigraph/src/changes/mod.rs @@ -236,7 +236,24 @@ pub(crate) async fn diff_snapshots( (Some(from), None) => diff_table_removed(table_store, from, is_edge, filter).await?, // Fast path: version-column diff (Some(from), Some(to)) if same_lineage(from_entry, to_entry) => { - diff_table_same_lineage(table_store, from, to, is_edge, filter).await? + match diff_table_same_lineage(table_store, from, to, is_edge, filter).await? { + Some(changes) => changes, + // RFC 0067: the `to` endpoint is a pending pin's detached + // version. Rows staged on a detached predecessor carry + // stamps outside the linear interval, so compare exactly. + None => { + diff_table_cross_branch( + table_store, + from, + to, + is_edge, + filter, + type_name, + to_commit_id.as_deref().unwrap_or_default(), + ) + .await? + } + } } // Cross-branch path: streaming ID-based diff (Some(from), Some(to)) => { @@ -321,11 +338,14 @@ async fn diff_table_same_lineage( to_entry: &DatasetEntry, is_edge: bool, filter: &ChangeFilter, -) -> Result> { +) -> Result>> { let vf = from_entry.published_dataset_version; let vt = to_entry.published_dataset_version; let storage: &dyn TableStorage = table_store; let to_ds = storage.open_snapshot_at_entry(to_entry).await?; + if TableStore::is_detached_version(to_ds.dataset().version().version) { + return Ok(None); + } let to_columns = system_columns_at_image(to_ds.dataset().schema(), &to_entry.type_key)?; let cols: Vec<&str> = if is_edge { @@ -364,7 +384,7 @@ async fn diff_table_same_lineage( Vec::new() }; if changed_rows.is_empty() && !wants_deletes { - return Ok(changes); + return Ok(Some(changes)); } let from_ds = storage.open_snapshot_at_entry(from_entry).await?; let from_columns = system_columns_at_image(from_ds.dataset().schema(), &from_entry.type_key)?; @@ -392,7 +412,7 @@ async fn diff_table_same_lineage( ); } - Ok(changes) + Ok(Some(changes)) } // ─── Cross-branch path: streaming ID-based diff ──────────────────────────── diff --git a/crates/omnigraph/src/db/manifest.rs b/crates/omnigraph/src/db/manifest.rs index f5887ad42..15414d5c8 100644 --- a/crates/omnigraph/src/db/manifest.rs +++ b/crates/omnigraph/src/db/manifest.rs @@ -578,6 +578,8 @@ impl Snapshot { entry.native_dataset_branch.as_deref(), entry.published_dataset_version, entry.version_metadata.e_tag(), + entry.version_metadata.staged_version(), + entry.version_metadata.transaction_uuid(), &location, Some(&caches.session), ) @@ -887,9 +889,11 @@ impl DatasetEntry { // cached path (`Snapshot::open_lance_dataset` → handle cache) calls the same opener on // a miss with the shared session, so both paths count on the per-query // `table_wrapper`. - crate::instrumentation::open_dataset( + crate::instrumentation::open_pinned_dataset( &location, - crate::instrumentation::VersionResolution::At(self.published_dataset_version), + self.published_dataset_version, + self.version_metadata.staged_version(), + self.version_metadata.transaction_uuid(), session, crate::instrumentation::table_wrapper(), ) diff --git a/crates/omnigraph/src/db/manifest/metadata.rs b/crates/omnigraph/src/db/manifest/metadata.rs index 8f19d90a7..921277628 100644 --- a/crates/omnigraph/src/db/manifest/metadata.rs +++ b/crates/omnigraph/src/db/manifest/metadata.rs @@ -15,6 +15,9 @@ use super::layout::table_id_to_key; pub(super) const OMNIGRAPH_ROW_COUNT_KEY: &str = "omnigraph.row_count"; const OMNIGRAPH_TABLE_BRANCH_KEY: &str = "omnigraph.table_branch"; const OMNIGRAPH_TABLE_FORK_OWNER_KEY: &str = "omnigraph.table_fork_owner"; +/// RFC 0067 prototype pin fields, carried like the fork owner. +const OMNIGRAPH_STAGED_VERSION_KEY: &str = "omnigraph.staged_version"; +const OMNIGRAPH_TRANSACTION_UUID_KEY: &str = "omnigraph.transaction_uuid"; pub(super) fn namespace_version_metadata( row_count: u64, @@ -56,6 +59,10 @@ pub(super) fn parse_namespace_version_request( e_tag: request.e_tag.clone(), naming_scheme: request.naming_scheme.clone(), table_fork_owner: metadata.get(OMNIGRAPH_TABLE_FORK_OWNER_KEY).cloned(), + staged_version: metadata + .get(OMNIGRAPH_STAGED_VERSION_KEY) + .and_then(|value| value.parse::().ok()), + transaction_uuid: metadata.get(OMNIGRAPH_TRANSACTION_UUID_KEY).cloned(), }; Ok(( @@ -75,6 +82,14 @@ pub(crate) struct TableVersionMetadata { naming_scheme: Option, #[serde(default, skip_serializing_if = "Option::is_none")] table_fork_owner: Option, + /// RFC 0067: the detached Lance version this pin was staged + /// as before publication. `None` on pins published by the sidecar path. + #[serde(default, skip_serializing_if = "Option::is_none")] + staged_version: Option, + /// RFC 0067: the uuid of the staged transaction; promotion + /// replays it at the linear target and readers verify it. + #[serde(default, skip_serializing_if = "Option::is_none")] + transaction_uuid: Option, } impl TableVersionMetadata { @@ -93,9 +108,38 @@ impl TableVersionMetadata { e_tag: dataset.manifest_location().e_tag.clone(), naming_scheme: Some(format!("{:?}", dataset.manifest_location().naming_scheme)), table_fork_owner: None, + staged_version: None, + transaction_uuid: None, }) } + pub(crate) fn staged_version(&self) -> Option { + self.staged_version + } + + pub(crate) fn transaction_uuid(&self) -> Option<&str> { + self.transaction_uuid.as_deref() + } + + /// RFC 0067: whether `dataset` is the manifest this pin + /// witnesses. The e_tag identifies the exact manifest object; a pin that + /// carries a transaction uuid also accepts the promoted twin, whose + /// manifest differs but whose transaction is the same. + pub(crate) fn witnesses(&self, dataset: &Dataset) -> bool { + match self.e_tag.as_deref() { + None => return true, + Some(expected) if dataset.manifest_location().e_tag.as_deref() == Some(expected) => { + return true; + } + Some(_) => {} + } + match self.transaction_uuid.as_deref() { + Some(uuid) => crate::table_store::StagedTransactionIdentity::recorded_by(dataset) + .is_some_and(|identity| identity.uuid == uuid), + None => false, + } + } + pub(crate) fn table_fork_owner(&self) -> Option<&str> { self.table_fork_owner.as_deref() } @@ -159,6 +203,12 @@ impl TableVersionMetadata { if let Some(owner) = &self.table_fork_owner { metadata.insert(OMNIGRAPH_TABLE_FORK_OWNER_KEY.to_string(), owner.clone()); } + if let Some(staged) = self.staged_version { + metadata.insert(OMNIGRAPH_STAGED_VERSION_KEY.to_string(), staged.to_string()); + } + if let Some(uuid) = &self.transaction_uuid { + metadata.insert(OMNIGRAPH_TRANSACTION_UUID_KEY.to_string(), uuid.clone()); + } request.metadata = Some(metadata); request } @@ -377,10 +427,27 @@ mod tests { /// GQT does not expose the namespace registration request and response metadata. #[test] fn namespace_pointer_roundtrip_keeps_source_ownership() { - let source = TableVersionMetadata::from_json_str(LEGACY_JSON) - .unwrap() - .with_table_fork_owner(Some(OWNER)); + let source = TableVersionMetadata { + staged_version: Some(9_223_372_036_854_775_815), + transaction_uuid: Some("8b8d7a1e-6f2c-4f0e-9d3a-1c2b3a4d5e6f".to_string()), + ..TableVersionMetadata::from_json_str(LEGACY_JSON) + .unwrap() + .with_table_fork_owner(Some(OWNER)) + }; let request = source.to_create_table_version_request("node:Person", 7, 3, Some(FORK)); + let request_metadata = request.metadata.as_ref().unwrap(); + assert_eq!( + request_metadata + .get(OMNIGRAPH_STAGED_VERSION_KEY) + .map(String::as_str), + Some("9223372036854775815") + ); + assert_eq!( + request_metadata + .get(OMNIGRAPH_TRANSACTION_UUID_KEY) + .map(String::as_str), + Some("8b8d7a1e-6f2c-4f0e-9d3a-1c2b3a4d5e6f") + ); assert_eq!( request .metadata @@ -397,6 +464,14 @@ mod tests { assert_eq!(rows, 3); assert_eq!(native_ref.as_deref(), Some(FORK)); assert_eq!(target_registration, source); + assert_eq!( + target_registration.staged_version(), + Some(9_223_372_036_854_775_815) + ); + assert_eq!( + target_registration.transaction_uuid(), + Some("8b8d7a1e-6f2c-4f0e-9d3a-1c2b3a4d5e6f") + ); assert!(target_registration.is_table_fork_of(FORK, OWNER)); assert!(!target_registration.is_table_fork_of(FORK, TARGET)); diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index 42c9f4540..4898b3410 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -32,6 +32,7 @@ use crate::table_store::TableStore; mod export; pub(crate) mod optimize; +pub(crate) mod promotion; mod repair; pub(crate) mod schema_apply; pub(crate) mod system_column_upgrade; @@ -2281,6 +2282,22 @@ impl Omnigraph { ), )); } + // RFC 0067: a promotion that landed after this writer captured + // its snapshot moves HEAD one past the published version the + // writer holds. When the current manifest explains that HEAD, + // the writer's read set is merely stale and it reprepares + // instead of being sent to repair. + let graph_branch = normalized_branch.map(crate::branch_names::logical_branch_name); + if let Ok(current) = self.fresh_snapshot_for_branch_unchecked(graph_branch).await + && let Some(entry) = current.dataset(table_key) + && entry.published_dataset_version != expected_version + { + return Err(OmniError::manifest_read_set_changed( + format!("published_dataset_version:{table_key}"), + Some(expected_version.to_string()), + Some(entry.published_dataset_version.to_string()), + )); + } Err(OmniError::manifest_conflict(format!( "{} is at Lance HEAD version {}, ahead of published dataset version {}; \ run `omnigraph repair` before writing", diff --git a/crates/omnigraph/src/db/omnigraph/optimize.rs b/crates/omnigraph/src/db/omnigraph/optimize.rs index 3e011c47c..9ed1eb5c5 100644 --- a/crates/omnigraph/src/db/omnigraph/optimize.rs +++ b/crates/omnigraph/src/db/omnigraph/optimize.rs @@ -1284,6 +1284,14 @@ pub async fn cleanup_all_datasets( // Any branch snapshot read failure aborts before the first table GC: an // unknown live reference is never evidence that a version is disposable. let mut oldest_live_main_version_by_path = std::collections::HashMap::::new(); + // RFC 0067: a table whose pin a foreign commit blocks keeps its + // acknowledged rows only in a detached version, which stock version GC + // does not protect. Skip GC on it, and protect every link of its chain + // from the detached-manifest reaper below. + let mut blocked_gc_paths = std::collections::HashSet::::new(); + let mut protected_detached = + std::collections::HashMap::>::new(); + let mut table_locations = std::collections::BTreeSet::<(String, Option)>::new(); for branch_target in &graph_branches { if branch_target .as_deref() @@ -1301,6 +1309,26 @@ pub async fn cleanup_all_datasets( )) })?; for entry in branch_snapshot.datasets() { + // RFC 0067: a pending pin is promoted before any version is + // reclaimed, so stock Lance cleanup only ever sees linear history + // and the pin's detached manifest becomes surplus. + let full_path = format!("{}/{}", db.root_uri, entry.dataset_path); + table_locations.insert((full_path.clone(), entry.native_dataset_branch.clone())); + if let (Some(staged), Some(uuid)) = ( + entry.version_metadata.staged_version(), + entry.version_metadata.transaction_uuid(), + ) && let Some(chain) = settle_pin_before_cleanup(db, entry, staged, uuid).await? + { + blocked_gc_paths.insert(full_path.clone()); + let location = super::promotion::table_location( + &full_path, + entry.native_dataset_branch.as_deref(), + ); + protected_detached + .entry(location) + .or_default() + .extend(chain); + } // Validate that the exact protected version is still openable // before GC starts. This catches pre-existing damage from an older // cleanup implementation and keeps the sweep fail-closed instead @@ -1314,7 +1342,6 @@ pub async fn cleanup_all_datasets( if entry.native_dataset_branch.is_some() { continue; } - let full_path = format!("{}/{}", db.root_uri, entry.dataset_path); if branch_target.is_none() { let head = db.storage().open_dataset_head(&full_path, None).await?; if head.version() != entry.published_dataset_version { @@ -1335,6 +1362,12 @@ pub async fn cleanup_all_datasets( } let before_timestamp = options.older_than.map(|d| crate::dst_clock::now_utc() - d); + let reaped = + reap_detached_manifests(db, &table_locations, &protected_detached, before_timestamp) + .await?; + if reaped > 0 { + tracing::info!(reaped, "cleanup reaped aged surplus detached manifests"); + } let reconciled = reconcile_orphaned_branches_under_control_gates(db, before_timestamp).await?; if !reconciled.reclaimed.is_empty() { tracing::info!( @@ -1366,6 +1399,7 @@ pub async fn cleanup_all_datasets( let concurrency = maint_concurrency().min(table_tasks.len()).max(1); let storage = db.storage(); + let blocked_gc_paths = &blocked_gc_paths; // Fault-isolated per table: a single table's GC failure is recorded on its // stats row (`error: Some`) and logged, never aborting the healthy tables. @@ -1375,6 +1409,13 @@ pub async fn cleanup_all_datasets( .map(|(table_key, full_path, live_main_floor)| async move { let outcome: Result = async { fail(&CLEANUP_TABLE_GC)?; + if blocked_gc_paths.contains(&full_path) { + return Err(OmniError::manifest_conflict( + "a published write's promotion is blocked by a foreign commit at its \ + target version; version GC is skipped for this table until `omnigraph \ + repair` resolves the block", + )); + } // `cleanup_old_versions` is a Lance-only maintenance API not // surfaced through `TableStorage` — see the optimize path // above for the same rationale. It only needs a raw read borrow. @@ -2006,3 +2047,111 @@ mod tests { ); } } + +/// Promote one pending pin, then delete its detached manifest once the +/// pin is linear (RFC 0067). Returns the detached versions of a blocked +/// pin's chain, which cleanup must neither GC around nor reap. +async fn settle_pin_before_cleanup( + db: &Omnigraph, + entry: &crate::db::manifest::DatasetEntry, + staged: u64, + uuid: &str, +) -> Result>> { + let full_path = format!("{}/{}", db.root_uri, entry.dataset_path); + let table_branch = entry.native_dataset_branch.as_deref(); + let location = super::promotion::table_location(&full_path, table_branch); + let outcome = super::promotion::promote_pin( + db, + &entry.type_key, + &full_path, + table_branch, + entry.published_dataset_version, + staged, + uuid, + ) + .await?; + match outcome { + super::promotion::Promotion::Promoted(_) | super::promotion::Promotion::AlreadyPromoted => { + let detached = super::promotion::detached_manifest_path(&location, staged); + if let Err(error) = db.storage_adapter().delete(&detached).await { + tracing::warn!( + error = %error, + detached, + "could not delete a promoted pin's detached manifest; cleanup retries it" + ); + } + Ok(None) + } + super::promotion::Promotion::Blocked(reason) => { + tracing::warn!( + table = entry.type_key.as_str(), + target = entry.published_dataset_version, + reason, + "cleanup found a blocked pin; version GC is skipped for its table" + ); + let (_, chain) = super::promotion::walk_chain(db, &location, staged).await?; + Ok(Some( + chain.into_iter().map(|(version, _)| version).collect(), + )) + } + } +} + +/// Reap detached manifests that no pending pin protects and that are older +/// than the cleanup age policy (RFC 0067): the manifests of promoted pins a +/// later write superseded, the intermediate links of promoted chains, and +/// attempts that never published. A pin promoted in this pass had its +/// manifest deleted already; a recent manifest may belong to a write still +/// in flight, so the age gate is required and no policy reaps nothing. +async fn reap_detached_manifests( + db: &Omnigraph, + locations: &std::collections::BTreeSet<(String, Option)>, + protected: &std::collections::HashMap>, + before_timestamp: Option>, +) -> Result { + let Some(cutoff) = before_timestamp else { + return Ok(0); + }; + let mut reaped = 0u64; + for (full_path, table_branch) in locations { + let handle = db + .storage() + .open_dataset_head(full_path, table_branch.as_deref()) + .await?; + let dataset = handle.dataset(); + let store = dataset + .object_store(None) + .await + .map_err(OmniError::storage)?; + let location = super::promotion::table_location(full_path, table_branch.as_deref()); + let protected = protected.get(&location); + let mut files = store.read_dir_all(&dataset.versions_dir(), None); + while let Some(file) = files.next().await { + let file = file.map_err(OmniError::storage)?; + let Some(version) = file + .location + .filename() + .and_then(|name| name.strip_prefix('d')) + .and_then(|name| name.strip_suffix(".manifest")) + .and_then(|version| version.parse::().ok()) + else { + continue; + }; + if protected.is_some_and(|versions| versions.contains(&version)) + || file.last_modified >= cutoff + { + continue; + } + let detached = super::promotion::detached_manifest_path(&location, version); + match db.storage_adapter().delete(&detached).await { + Ok(()) => reaped += 1, + Err(error) => tracing::warn!( + error = %error, + detached, + "could not reap a surplus detached manifest; cleanup retries it" + ), + } + } + } + Ok(reaped) +} diff --git a/crates/omnigraph/src/db/omnigraph/promotion.rs b/crates/omnigraph/src/db/omnigraph/promotion.rs new file mode 100644 index 000000000..f97319201 --- /dev/null +++ b/crates/omnigraph/src/db/omnigraph/promotion.rs @@ -0,0 +1,546 @@ +//! Promotion of table pins onto their linear history (RFC 0067). +//! +//! A pin published as `(target, staged, uuid)` names a detached Lance version +//! whose twin has not yet been written at `target`. Promotion replays the +//! recorded transaction at `target - 1` through the sealed storage boundary. +//! It is derived, idempotent reconciler work: any process may run it, two +//! promoters racing on one pin leave exactly one twin, and a pin that cannot +//! be promoted because a foreign commit occupies its target stays readable +//! through its staged version and degrades only reclamation. +//! +//! A chain of detached commits links itself: every detached commit records +//! the version it was staged on, so a pin whose linear base is absent is the +//! tip of a chain found by following read-version links, never by reading +//! manifest history. + +use crate::db::omnigraph::Omnigraph; +use crate::error::{OmniError, Result}; +use crate::instrumentation::{VersionResolution, open_dataset, table_wrapper}; +use crate::storage_layer::{PromotionOutcome, SnapshotHandle}; + +/// Outcome of promoting one pin. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum Promotion { + /// The twin landed at the target. + Promoted(u64), + /// The target already carried the pin's transaction. + AlreadyPromoted, + /// The pin cannot be promoted; the reason names why. Reads stay correct + /// through the staged version. + Blocked(String), +} + +/// Where a table's versions live: the dataset root for main, the native +/// branch tree otherwise. +pub(crate) fn table_location(full_path: &str, table_branch: Option<&str>) -> String { + match table_branch.filter(|branch| *branch != "main") { + Some(branch) => { + let mut location = crate::storage::join_uri(full_path, "tree"); + for segment in branch.split('/') { + location = crate::storage::join_uri(&location, segment); + } + location + } + None => full_path.to_string(), + } +} + +/// Where Lance keeps a detached version's manifest: `d{version}.manifest` +/// under `_versions`, the masked version spelled in full. +pub(crate) fn detached_manifest_path(location: &str, staged: u64) -> String { + format!("{location}/_versions/d{staged}.manifest") +} + +/// Open one version of a table, or `None` when that version is reclaimed. +async fn open_at(db: &Omnigraph, location: &str, version: u64) -> Result> { + let session = db.read_caches().session.clone(); + match open_dataset( + location, + VersionResolution::At(version), + Some(&session), + table_wrapper(), + ) + .await + { + Ok(dataset) => Ok(Some(SnapshotHandle::new(dataset))), + Err(OmniError::HistoricalVersionReclaimed { .. }) => Ok(None), + Err(error) => Err(error), + } +} + +/// Whether the target carries the pin's transaction, a foreign one, or is +/// absent. +async fn target_state( + db: &Omnigraph, + location: &str, + target: u64, + uuid: &str, +) -> Result> { + let Some(existing) = open_at(db, location, target).await? else { + return Ok(None); + }; + Ok(Some(match db.storage().transaction_identity(&existing) { + Ok(link) if link.uuid == uuid => Promotion::AlreadyPromoted, + _ => Promotion::Blocked(format!( + "linear version {target} exists with a foreign or unreadable transaction" + )), + })) +} + +/// Follow a detached tip's read-version links back to the linear base it was +/// staged from. Returns the base, or `None` when the walk met a predecessor +/// whose detached manifest is already reaped, and the chain tip first as +/// `(version, uuid)`. +pub(crate) async fn walk_chain( + db: &Omnigraph, + location: &str, + tip: u64, +) -> Result<(Option, Vec<(u64, String)>)> { + let mut chain = Vec::new(); + let mut version = tip; + loop { + let Some(handle) = open_at(db, location, version).await? else { + if chain.is_empty() { + return Err(OmniError::HistoricalVersionReclaimed { + published_dataset_version: version, + }); + } + return Ok((None, chain)); + }; + let link = db.storage().transaction_identity(&handle)?; + let base_is_detached = link.base_is_detached(); + let read_version = link.read_version; + chain.push((version, link.uuid)); + if !base_is_detached { + return Ok((Some(read_version), chain)); + } + version = read_version; + } +} + +/// Promote the pin `(target, staged, uuid)` of one table, and every detached +/// commit behind it, oldest first. A promoter that did not stage the pin +/// checks the target first: the same pin is registered on every branch that +/// inherits it, and a promoted pin's detached manifest may already be +/// reaped. Bounded by the chain length. +pub(crate) async fn promote_pin( + db: &Omnigraph, + table_key: &str, + full_path: &str, + table_branch: Option<&str>, + target: u64, + staged: u64, + uuid: &str, +) -> Result { + let location = table_location(full_path, table_branch); + if let Some(known) = target_state(db, &location, target, uuid).await? { + return Ok(known); + } + let (base, chain) = walk_chain(db, &location, staged).await?; + if chain[0].1 != uuid { + return Ok(Promotion::Blocked( + "staged version carries no matching transaction".to_string(), + )); + } + let base = match base { + Some(base) if base + chain.len() as u64 == target => base, + Some(base) => { + return Ok(Promotion::Blocked(format!( + "{table_key}: a chain of {} detached commits on linear base {base} does not reach target {target}", + chain.len() + ))); + } + None => { + let base = target.saturating_sub(chain.len() as u64); + if open_at(db, &location, base).await?.is_none() { + return Ok(Promotion::Blocked(format!( + "{table_key}: chain predecessors were reclaimed and linear base {base} is absent" + ))); + } + base + } + }; + for (index, (staged, uuid)) in chain.iter().rev().enumerate() { + let step_target = base + 1 + index as u64; + match promote_step(db, &location, step_target, *staged, uuid).await? { + Promotion::Promoted(_) | Promotion::AlreadyPromoted => {} + blocked => return Ok(blocked), + } + } + Ok(Promotion::Promoted(target)) +} + +/// One link of a chain: replay `staged` at `target`, rechecking the target +/// when Lance refuses the replay because a racing promoter landed first. +async fn promote_step( + db: &Omnigraph, + location: &str, + target: u64, + staged: u64, + uuid: &str, +) -> Result { + if let Some(known) = target_state(db, location, target, uuid).await? { + return Ok(known); + } + let Some(base) = open_at(db, location, target - 1).await? else { + return Ok(Promotion::Blocked(format!( + "linear base {} is absent", + target - 1 + ))); + }; + let Some(staged_handle) = open_at(db, location, staged).await? else { + return Ok(Promotion::Blocked(format!( + "staged version {staged} is reclaimed" + ))); + }; + match db + .storage() + .promote_detached(base, &staged_handle, target, uuid) + .await? + { + PromotionOutcome::Landed(twin) => { + debug_assert_eq!(twin.version(), target); + Ok(Promotion::Promoted(target)) + } + PromotionOutcome::Refused => Ok(target_state(db, location, target, uuid) + .await? + .unwrap_or_else(|| { + Promotion::Blocked("replay refused and the target is absent".to_string()) + })), + PromotionOutcome::Unsafe(reason) => Ok(Promotion::Blocked(reason)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::manifest::DatasetEntry; + use crate::instrumentation::open_pinned_dataset; + use crate::loader::LoadMode; + use crate::table_store::{StagedTransactionIdentity, TableStore}; + + const SCHEMA: &str = "node Person { name: String @key }\n"; + const PERSON: &str = "node:Person"; + + /// A graph whose Person table holds `a`..`h` after one linear load. + async fn graph_with_people() -> (tempfile::TempDir, Omnigraph) { + let dir = tempfile::tempdir().unwrap(); + let db = Omnigraph::init(dir.path().to_str().unwrap(), SCHEMA) + .await + .unwrap(); + let rows: Vec = ('a'..='h') + .map(|name| format!(r#"{{"type":"Person","data":{{"name":"{name}"}}}}"#)) + .collect(); + db.load_as("main", None, &rows.join("\n"), LoadMode::Merge, None) + .await + .unwrap(); + (dir, db) + } + + async fn person_entry(db: &Omnigraph) -> DatasetEntry { + db.snapshot_for_branch(None) + .await + .unwrap() + .dataset(PERSON) + .unwrap() + .clone() + } + + /// The table's full path and the location its versions live under. + fn location_of(db: &Omnigraph, entry: &DatasetEntry) -> (String, String) { + let full_path = format!("{}/{}", db.root_uri, entry.dataset_path); + let location = table_location(&full_path, entry.native_dataset_branch.as_deref()); + (full_path, location) + } + + /// Stage one delete on `base` and commit it as a detached version. + async fn detached_delete( + db: &Omnigraph, + base: SnapshotHandle, + name: &str, + ) -> (SnapshotHandle, StagedTransactionIdentity) { + let staged = db + .storage() + .stage_delete(&base, &format!("name = '{name}'")) + .await + .unwrap() + .expect("the row exists"); + db.storage() + .commit_staged_detached(base, staged) + .await + .unwrap() + } + + async fn uuid_at(db: &Omnigraph, location: &str, version: u64) -> Option { + open_at(db, location, version) + .await + .unwrap() + .map(|handle| db.storage().transaction_identity(&handle).unwrap().uuid) + } + + async fn rows_at(db: &Omnigraph, location: &str, version: u64) -> usize { + let handle = open_at(db, location, version).await.unwrap().unwrap(); + db.storage().count_rows(&handle, None).await.unwrap() + } + + async fn load_one(db: &Omnigraph, name: &str) { + let row = format!(r#"{{"type":"Person","data":{{"name":"{name}"}}}}"#); + db.load_as("main", None, &row, LoadMode::Merge, None) + .await + .unwrap(); + } + + /// GQT cannot observe a detached Lance version: it never appears in a + /// manifest. A detached commit is private, links to its base, resolves + /// through its pin, and promotion lands its twin exactly once. + #[tokio::test] + async fn detached_commit_stays_private_until_its_pin_is_promoted() { + let (_dir, db) = graph_with_people().await; + let entry = person_entry(&db).await; + let (full_path, location) = location_of(&db, &entry); + let branch = entry.native_dataset_branch.as_deref(); + let base = entry.published_dataset_version; + let head = db.storage().open_snapshot_at_entry(&entry).await.unwrap(); + + let (detached, identity) = detached_delete(&db, head, "b").await; + assert!(TableStore::is_detached_version(detached.version())); + assert_eq!(identity.read_version, base); + assert!( + open_at(&db, &location, base + 1).await.unwrap().is_none(), + "a detached commit must not move the linear history" + ); + let link = db.storage().transaction_identity(&detached).unwrap(); + assert_eq!(link.uuid, identity.uuid); + assert_eq!(link.read_version, base); + assert!(!link.base_is_detached()); + assert_eq!(db.storage().count_rows(&detached, None).await.unwrap(), 7); + + // While the twin is absent the pin resolves to the staged version. + let pending = open_pinned_dataset( + &location, + base + 1, + Some(detached.version()), + Some(&identity.uuid), + None, + None, + ) + .await + .unwrap(); + assert_eq!(pending.version().version, detached.version()); + + let promoted = promote_pin( + &db, + PERSON, + &full_path, + branch, + base + 1, + detached.version(), + &identity.uuid, + ) + .await + .unwrap(); + assert_eq!(promoted, Promotion::Promoted(base + 1)); + assert_eq!( + uuid_at(&db, &location, base + 1).await.as_deref(), + Some(identity.uuid.as_str()) + ); + assert_eq!(rows_at(&db, &location, base + 1).await, 7); + + // Promotion is idempotent and the pin now resolves to the twin. + let again = promote_pin( + &db, + PERSON, + &full_path, + branch, + base + 1, + detached.version(), + &identity.uuid, + ) + .await + .unwrap(); + assert_eq!(again, Promotion::AlreadyPromoted); + let resolved = open_pinned_dataset( + &location, + base + 1, + Some(detached.version()), + Some(&identity.uuid), + None, + None, + ) + .await + .unwrap(); + assert_eq!(resolved.version().version, base + 1); + + // Replaying over the landed twin is refused, never duplicated. + let stale_base = open_at(&db, &location, base).await.unwrap().unwrap(); + let refused = db + .storage() + .promote_detached(stale_base, &detached, base + 1, &identity.uuid) + .await + .unwrap(); + assert!(matches!(refused, PromotionOutcome::Refused), "{refused:?}"); + assert!(open_at(&db, &location, base + 2).await.unwrap().is_none()); + } + + /// A pin staged on a detached version is the tip of a chain. Promotion + /// lands the chain oldest first, and a reaped predecessor does not block + /// a tip whose linear base already exists. + #[tokio::test] + async fn chain_promotes_oldest_first_and_tolerates_reaped_predecessors() { + let (_dir, db) = graph_with_people().await; + let entry = person_entry(&db).await; + let (full_path, location) = location_of(&db, &entry); + let branch = entry.native_dataset_branch.as_deref(); + let base = entry.published_dataset_version; + let head = db.storage().open_snapshot_at_entry(&entry).await.unwrap(); + + let (first, first_id) = detached_delete(&db, head, "a").await; + let first_again = open_at(&db, &location, first.version()) + .await + .unwrap() + .unwrap(); + let (second, second_id) = detached_delete(&db, first_again, "b").await; + let link = db.storage().transaction_identity(&second).unwrap(); + assert!(link.base_is_detached()); + assert_eq!(link.read_version, first.version()); + + let outcome = promote_pin( + &db, + PERSON, + &full_path, + branch, + base + 2, + second.version(), + &second_id.uuid, + ) + .await + .unwrap(); + assert_eq!(outcome, Promotion::Promoted(base + 2)); + assert_eq!(uuid_at(&db, &location, base + 1).await, Some(first_id.uuid)); + assert_eq!( + uuid_at(&db, &location, base + 2).await, + Some(second_id.uuid) + ); + assert_eq!(rows_at(&db, &location, base + 2).await, 6); + + // Promote a chain's first link, reap its detached manifest, then + // promote the tip through the reaped predecessor. + let head = open_at(&db, &location, base + 2).await.unwrap().unwrap(); + let (third, third_id) = detached_delete(&db, head, "c").await; + let third_again = open_at(&db, &location, third.version()) + .await + .unwrap() + .unwrap(); + let (fourth, fourth_id) = detached_delete(&db, third_again, "d").await; + let outcome = promote_pin( + &db, + PERSON, + &full_path, + branch, + base + 3, + third.version(), + &third_id.uuid, + ) + .await + .unwrap(); + assert_eq!(outcome, Promotion::Promoted(base + 3)); + db.storage_adapter() + .delete(&detached_manifest_path(&location, third.version())) + .await + .unwrap(); + assert!( + open_at(&db, &location, third.version()) + .await + .unwrap() + .is_none() + ); + let outcome = promote_pin( + &db, + PERSON, + &full_path, + branch, + base + 4, + fourth.version(), + &fourth_id.uuid, + ) + .await + .unwrap(); + assert_eq!(outcome, Promotion::Promoted(base + 4)); + assert_eq!( + uuid_at(&db, &location, base + 4).await, + Some(fourth_id.uuid) + ); + assert_eq!(rows_at(&db, &location, base + 4).await, 4); + } + + /// A foreign linear commit at the target blocks promotion while the pin + /// keeps reading its own staged effect; a target pruned behind the + /// linear head is reclaimed, never silently served from the staged + /// version. + #[tokio::test] + async fn foreign_target_blocks_and_resolution_follows_the_linear_head() { + let (_dir, db) = graph_with_people().await; + let entry = person_entry(&db).await; + let (full_path, location) = location_of(&db, &entry); + let branch = entry.native_dataset_branch.as_deref(); + let base = entry.published_dataset_version; + let head = db.storage().open_snapshot_at_entry(&entry).await.unwrap(); + let (detached, identity) = detached_delete(&db, head, "a").await; + + load_one(&db, "z").await; + assert_eq!(person_entry(&db).await.published_dataset_version, base + 1); + let foreign = uuid_at(&db, &location, base + 1).await.unwrap(); + assert_ne!(foreign, identity.uuid); + + let outcome = promote_pin( + &db, + PERSON, + &full_path, + branch, + base + 1, + detached.version(), + &identity.uuid, + ) + .await + .unwrap(); + assert!( + matches!(&outcome, Promotion::Blocked(reason) if reason.contains("foreign")), + "{outcome:?}" + ); + assert_eq!(uuid_at(&db, &location, base + 1).await, Some(foreign)); + assert!(open_at(&db, &location, base + 2).await.unwrap().is_none()); + let resolved = open_pinned_dataset( + &location, + base + 1, + Some(detached.version()), + Some(&identity.uuid), + None, + None, + ) + .await + .unwrap(); + assert_eq!(resolved.version().version, detached.version()); + + // Once the head has passed a pruned target, the pin is reclaimed. + load_one(&db, "y").await; + let linear = format!( + "{location}/_versions/{:020}.manifest", + u64::MAX - (base + 1) + ); + db.storage_adapter().delete(&linear).await.unwrap(); + let error = open_pinned_dataset( + &location, + base + 1, + Some(detached.version()), + Some(&identity.uuid), + None, + None, + ) + .await + .unwrap_err(); + assert!( + matches!(error, OmniError::HistoricalVersionReclaimed { .. }), + "{error}" + ); + } +} diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs index 29ae25df1..d789cc295 100644 --- a/crates/omnigraph/src/exec/merge.rs +++ b/crates/omnigraph/src/exec/merge.rs @@ -4118,6 +4118,9 @@ async fn classify_general_adopt( /// An empty delta does not imply an empty publish: source and target can hold /// the same content at different Lance versions (#473). #[must_use = "the adopt plan decides whether this table is a merge candidate"] +// A registration carries the pin metadata RFC 0067 added; the pointer +// variant is the common one and boxing it buys nothing. +#[allow(clippy::large_enum_variant)] enum AdoptPublish { /// The planned registration is field-for-field the stored entry /// (`reregisters_current_entry`). diff --git a/crates/omnigraph/src/instrumentation.rs b/crates/omnigraph/src/instrumentation.rs index 101663bfd..c0fa96ca4 100644 --- a/crates/omnigraph/src/instrumentation.rs +++ b/crates/omnigraph/src/instrumentation.rs @@ -1464,6 +1464,88 @@ pub(crate) enum VersionResolution { At(u64), } +/// Open a table pin (RFC 0067). A pin without a staged version is an exact +/// open of its target. A pin with one opens the linear target when it exists +/// and carries the pin's transaction uuid, and otherwise the staged detached +/// version: either the pin is pending promotion, or its promotion is blocked +/// by a foreign commit at the target. An absent target counts as pending only +/// while the table's linear head has not reached it; once the head is at or +/// past the target, the twin was promoted and later pruned, and the staged +/// manifest must not resurrect pruned history. +pub(crate) async fn open_pinned_dataset( + uri: &str, + target_version: u64, + staged_version: Option, + transaction_uuid: Option<&str>, + session: Option<&Arc>, + wrapper: Option>, +) -> Result { + let Some(staged) = staged_version else { + return open_dataset(uri, VersionResolution::At(target_version), session, wrapper).await; + }; + match open_dataset( + uri, + VersionResolution::At(target_version), + session, + wrapper.clone(), + ) + .await + { + Ok(dataset) => { + let ours = crate::table_store::StagedTransactionIdentity::recorded_by(&dataset) + .is_some_and(|identity| Some(identity.uuid.as_str()) == transaction_uuid); + if ours { + return Ok(dataset); + } + tracing::warn!( + uri, + target_version, + staged, + "pin target carries a foreign or unreadable transaction; resolving the staged version" + ); + open_dataset(uri, VersionResolution::At(staged), session, wrapper).await + } + Err(error @ OmniError::HistoricalVersionReclaimed { .. }) => { + let latest = open_dataset(uri, VersionResolution::Latest, session, wrapper.clone()) + .await? + .version() + .version; + if latest < target_version { + return open_dataset(uri, VersionResolution::At(staged), session, wrapper).await; + } + // HEAD reached the target between the two reads, so a promotion + // may have landed the twin after the first probe missed it. Read + // the exact target once more before reporting reclaimed history. + match open_dataset( + uri, + VersionResolution::At(target_version), + session, + wrapper.clone(), + ) + .await + { + Ok(dataset) => { + let ours = crate::table_store::StagedTransactionIdentity::recorded_by(&dataset) + .is_some_and(|identity| Some(identity.uuid.as_str()) == transaction_uuid); + if ours { + return Ok(dataset); + } + tracing::warn!( + uri, + target_version, + staged, + "pin target carries a foreign or unreadable transaction; resolving the staged version" + ); + open_dataset(uri, VersionResolution::At(staged), session, wrapper).await + } + Err(OmniError::HistoricalVersionReclaimed { .. }) => Err(error), + Err(other) => Err(other), + } + } + Err(error) => Err(error), + } +} + /// THE dataset-open chokepoint. Every engine `Dataset` open routes through /// here so three things hold uniformly, on every path: /// diff --git a/crates/omnigraph/src/runtime_cache.rs b/crates/omnigraph/src/runtime_cache.rs index ab7f284ca..a609ca153 100644 --- a/crates/omnigraph/src/runtime_cache.rs +++ b/crates/omnigraph/src/runtime_cache.rs @@ -376,6 +376,8 @@ impl TableHandleCache { table_branch: Option<&str>, version: u64, e_tag: Option<&str>, + staged_version: Option, + transaction_uuid: Option<&str>, location: &str, session: Option<&Arc>, ) -> Result { @@ -394,9 +396,11 @@ impl TableHandleCache { // Miss: open without holding the lock (the open is async IO). A concurrent // double-miss opens twice and one wins the insert — correct (the dataset // at a version is immutable) and rare. - let ds = crate::instrumentation::open_dataset( + let ds = crate::instrumentation::open_pinned_dataset( location, - crate::instrumentation::VersionResolution::At(version), + version, + staged_version, + transaction_uuid, session, crate::instrumentation::table_wrapper(), ) diff --git a/crates/omnigraph/src/seams/catalog.rs b/crates/omnigraph/src/seams/catalog.rs index e99e5230c..c9387ee57 100644 --- a/crates/omnigraph/src/seams/catalog.rs +++ b/crates/omnigraph/src/seams/catalog.rs @@ -95,6 +95,7 @@ omnigraph_seams::catalog! { crate::loader::LOAD_POST_BRANCH_CREATE_PRE_STAGE, crate::table_store::CHANGE_FEED_ETAG_WITNESS, crate::table_store::FORK_POST_CREATE_PRE_OPEN, + crate::table_store::PROMOTION_PRE_REPLAY, } /// The one string-keyed lookup: a case file or the harness names a seam, the catalog answers. diff --git a/crates/omnigraph/src/storage_layer.rs b/crates/omnigraph/src/storage_layer.rs index 1e6e05835..f0182cf7d 100644 --- a/crates/omnigraph/src/storage_layer.rs +++ b/crates/omnigraph/src/storage_layer.rs @@ -220,6 +220,15 @@ impl ProvenInsertChunk { // ─── opaque handles ──────────────────────────────────────────────────────── +/// Outcome of replaying a detached transaction at its linear target, in the +/// storage boundary's own terms (RFC 0067). +#[derive(Debug)] +pub enum PromotionOutcome { + Landed(SnapshotHandle), + Refused, + Unsafe(String), +} + /// Opaque handle to a snapshot of a single sub-table dataset at a /// specific version. /// @@ -726,6 +735,32 @@ pub trait TableStorage: sealed::Sealed + Send + Sync + Debug { staged: StagedHandle, ) -> Result; + /// RFC 0067: commit one staged effect as a detached version. + async fn commit_staged_detached( + &self, + snapshot: SnapshotHandle, + staged: StagedHandle, + ) -> Result<( + SnapshotHandle, + crate::table_store::StagedTransactionIdentity, + )>; + + /// RFC 0067: the identity of the transaction a version records. + fn transaction_identity( + &self, + snapshot: &SnapshotHandle, + ) -> Result; + + /// RFC 0067: replay the transaction recorded in `staged` on `base` so the + /// linear history gains its twin at `target`. + async fn promote_detached( + &self, + base: SnapshotHandle, + staged: &SnapshotHandle, + target: u64, + expected_uuid: &str, + ) -> Result; + /// Stage an overwrite (Operation::Overwrite). MR-793 Phase 2. async fn stage_overwrite( &self, @@ -1237,6 +1272,50 @@ impl TableStorage for TableStore { .map(SnapshotHandle::new) } + async fn commit_staged_detached( + &self, + snapshot: SnapshotHandle, + staged: StagedHandle, + ) -> Result<( + SnapshotHandle, + crate::table_store::StagedTransactionIdentity, + )> { + let ds_arc = snapshot.into_arc(); + let (dataset, identity) = + TableStore::commit_staged_detached(self, ds_arc, staged.into_staged()).await?; + Ok((SnapshotHandle::new(dataset), identity)) + } + + fn transaction_identity( + &self, + snapshot: &SnapshotHandle, + ) -> Result { + TableStore::transaction_identity(self, snapshot.dataset()) + } + + async fn promote_detached( + &self, + base: SnapshotHandle, + staged: &SnapshotHandle, + target: u64, + expected_uuid: &str, + ) -> Result { + let base = base.into_arc(); + Ok( + match TableStore::promote_detached(self, base, staged.dataset(), target, expected_uuid) + .await? + { + crate::table_store::PromotionCommit::Landed(dataset) => { + PromotionOutcome::Landed(SnapshotHandle::new(*dataset)) + } + crate::table_store::PromotionCommit::Refused => PromotionOutcome::Refused, + crate::table_store::PromotionCommit::Unsafe(reason) => { + PromotionOutcome::Unsafe(reason) + } + }, + ) + } + async fn commit_staged_exact( &self, snapshot: SnapshotHandle, diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs index 4f1b10d2d..91873a829 100644 --- a/crates/omnigraph/src/table_store.rs +++ b/crates/omnigraph/src/table_store.rs @@ -470,6 +470,11 @@ pub enum IndexCoverage { /// started from the same version. Lance may preserve both fields while /// rebasing, so enrolled callers must also require the achieved table version /// to be exactly `read_version + 1`. +/// +/// RFC 0067 reads the same pair from the transaction file name the manifest +/// records (`{read_version}-{uuid}.txn`, pinned in `lance_surface_guards`), +/// so identifying a pin's promoted twin or following a chain of detached +/// commits costs no request beyond the manifest itself. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct StagedTransactionIdentity { pub read_version: u64, @@ -485,6 +490,37 @@ impl From<&Transaction> for StagedTransactionIdentity { } } +impl StagedTransactionIdentity { + /// The identity a manifest records, or `None` when it names no + /// transaction file or the name has an unrecognized shape. + pub(crate) fn recorded_by(dataset: &Dataset) -> Option { + let name = dataset.manifest().transaction_file.as_deref()?; + let (read_version, uuid) = name.strip_suffix(".txn")?.split_once('-')?; + Some(Self { + read_version: read_version.parse().ok()?, + uuid: uuid.to_string(), + }) + } + + /// Whether the transaction was staged on a detached version, which is + /// how a chain of detached commits links itself without manifest history. + pub(crate) fn base_is_detached(&self) -> bool { + TableStore::is_detached_version(self.read_version) + } +} + +/// Outcome of replaying a detached transaction at its linear target. +#[derive(Debug)] +pub enum PromotionCommit { + /// The twin landed at the target with the expected uuid. + Landed(Box), + /// Lance's conflict pass refused the replay: a racing promoter landed + /// first, or a foreign commit occupies the target. The caller rechecks. + Refused, + /// The replay must not run or did not land where it should. + Unsafe(String), +} + /// A Lance write that has produced fragment files on object storage but is /// not yet committed to the dataset's manifest. The staged-write primitives /// are consumed by `MutationStaging` (`exec/staging.rs`, @@ -962,6 +998,13 @@ decide_seam! { pub static FORK_POST_CREATE_PRE_OPEN = ("fork.post_create_pre_open", AnyWrite, [Fail]); } +decide_seam! { + /// After a promotion's existence check and before its linear replay + /// (RFC 0067). A decision here aligns two promoters on one pin, + /// or fails a promotion so the next writer inherits it. + pub static PROMOTION_PRE_REPLAY = ("promotion.pre_replay", Unreachable, [Fail]); +} + decide_seam! { /// The e_tag comparison in `open_at_entry_verified`. Skipping it simulates /// a store whose persisted table version metadata carries no e_tag. Tests @@ -1183,10 +1226,7 @@ impl TableStore { // failpoint seam simulates the e_tag-less-store configuration so tests // can prove the logical witness alone refuses a branch delete/recreate. let etag_witness_unavailable = skip(&CHANGE_FEED_ETAG_WITNESS); - if !etag_witness_unavailable - && let Some(expected) = entry.version_metadata.e_tag() - && dataset.manifest_location().e_tag.as_deref() != Some(expected) - { + if !etag_witness_unavailable && !entry.version_metadata.witnesses(&dataset) { return Err(OmniError::manifest(format!( "change feed table '{}' has no persisted native-branch incarnation \ witness at the reopened dataset; the branch was deleted and \ @@ -3762,6 +3802,108 @@ impl TableStore { Ok((dataset, committed_identity)) } + /// RFC 0067: commit a staged effect as a Lance detached + /// version of its base. No conflict pass runs, nothing at HEAD moves, and + /// the result is invisible until a manifest pin references it. + pub async fn commit_staged_detached( + &self, + ds: Arc, + staged: StagedWrite, + ) -> Result<(Dataset, StagedTransactionIdentity)> { + let mut builder = CommitBuilder::new(ds) + .with_skip_auto_cleanup(true) + .with_detached(true); + if let Some(affected_rows) = staged.commit_metadata.affected_rows { + builder = builder.with_affected_rows(affected_rows); + } + let dataset = builder + .execute(staged.transaction) + .await + .map_err(OmniError::storage)?; + let identity = self.transaction_identity(&dataset)?; + Ok((dataset, identity)) + } + + /// Whether a Lance version id names a detached version (RFC 0067). + pub(crate) fn is_detached_version(version: u64) -> bool { + version & lance_table::format::DETACHED_VERSION_MASK != 0 + } + + /// The identity of the transaction a version records (RFC 0067), read + /// from the manifest alone. + pub fn transaction_identity(&self, ds: &Dataset) -> Result { + StagedTransactionIdentity::recorded_by(ds).ok_or_else(|| { + OmniError::manifest_internal(format!( + "version {} of {} records no transaction file", + ds.version().version, + ds.uri() + )) + }) + } + + /// Replay the transaction recorded in `staged` linearly on `base`, so the + /// linear history gains an identical twin at `target` (RFC 0067). The + /// replay runs with zero retries; a twin that a racing promoter already + /// landed is refused by Lance's conflict pass for every kind the engine + /// stages detached (the self-conflict rule pinned in + /// `lance_surface_guards`) and reported as `Refused` for the caller to + /// recheck. A bare `Append` never replays: it would rebase over its twin + /// and duplicate rows. + pub async fn promote_detached( + &self, + base: Arc, + staged: &Dataset, + target: u64, + expected_uuid: &str, + ) -> Result { + let mut transaction = match staged + .read_transaction() + .await + .map_err(OmniError::storage)? + { + Some(transaction) if transaction.uuid == expected_uuid => transaction, + _ => { + return Ok(PromotionCommit::Unsafe( + "staged version carries no matching transaction".to_string(), + )); + } + }; + if matches!(transaction.operation, Operation::Append { .. }) { + return Ok(PromotionCommit::Unsafe( + "a bare Append is not replay-safe; the engine never stages one".to_string(), + )); + } + if base.version().version + 1 != target { + return Ok(PromotionCommit::Unsafe(format!( + "base {} is not the predecessor of target {target}", + base.version().version + ))); + } + transaction.read_version = target - 1; + fail(&PROMOTION_PRE_REPLAY)?; + match CommitBuilder::new(base) + .with_max_retries(0) + .with_skip_auto_cleanup(true) + .execute(transaction) + .await + { + Ok(dataset) => { + let landed = dataset.version().version; + let uuid_matches = StagedTransactionIdentity::recorded_by(&dataset) + .is_some_and(|identity| identity.uuid == expected_uuid); + if landed == target && uuid_matches { + Ok(PromotionCommit::Landed(Box::new(dataset))) + } else { + Ok(PromotionCommit::Unsafe(format!( + "replay landed at {landed} for target {target} (uuid match {uuid_matches})" + ))) + } + } + Err(lance::Error::RetryableCommitConflict { .. }) => Ok(PromotionCommit::Refused), + Err(error) => Err(OmniError::storage(error)), + } + } + /// Commit a staged first-touch dataset creation with no conflict retry. /// /// A create transaction is Lance's `Operation::Overwrite` at diff --git a/crates/omnigraph/tests/forbidden_apis.rs b/crates/omnigraph/tests/forbidden_apis.rs index eae7d7ad1..12649884a 100644 --- a/crates/omnigraph/tests/forbidden_apis.rs +++ b/crates/omnigraph/tests/forbidden_apis.rs @@ -585,6 +585,7 @@ gateway_surfaces! { "write_text_if_match", "delete_prefix", ], "storage_layer.rs" => "TableStorage" => GatewayDisposition::ReadOrPure => [ + "transaction_identity", "open_snapshot_at_entry", "open_snapshot_at_table", "open_dataset_head", "branch_identifier", "list_native_branches", "reopen_for_mutation", "ensure_expected_version", "scan", "scan_with_row_id", "scan_batches", @@ -617,7 +618,14 @@ gateway_surfaces! { "storage_layer.rs" => "TableStorage" => GatewayDisposition::Durable(WriteProtocol::Exact("staged exact commit gateway")) => [ "commit_staged_exact", ], + "storage_layer.rs" => "TableStorage" => GatewayDisposition::Durable(WriteProtocol::Exact("RFC 0067 detached staged commit gateway")) => [ + "commit_staged_detached", + ], + "storage_layer.rs" => "TableStorage" => GatewayDisposition::Durable(WriteProtocol::Exact("RFC 0067 promotion replay gateway")) => [ + "promote_detached", + ], "table_store.rs" => "TableStore" => GatewayDisposition::ReadOrPure => [ + "is_detached_version", "transaction_identity", "new", "root_uri", "dataset_uri", "open_snapshot_table", "open_at_entry", "open_at_entry_verified", "open_dataset_head", "list_native_branches", "named_fork_is_absent", "ensure_expected_version", @@ -662,6 +670,12 @@ gateway_surfaces! { "table_store.rs" => "TableStore" => GatewayDisposition::Durable(WriteProtocol::Exact("staged exact commit gateway")) => [ "commit_staged_exact", ], + "table_store.rs" => "TableStore" => GatewayDisposition::Durable(WriteProtocol::Exact("RFC 0067 detached staged commit gateway")) => [ + "commit_staged_detached", + ], + "table_store.rs" => "TableStore" => GatewayDisposition::Durable(WriteProtocol::Exact("RFC 0067 promotion replay gateway")) => [ + "promote_detached", + ], "table_store.rs" => "TableStore" => GatewayDisposition::Durable(WriteProtocol::EphemeralScratch) => [ "append_or_create_batch", "create_empty_dataset", "write_dataset", ], @@ -745,15 +759,20 @@ durable_calls! { ("storage_layer.rs", ".commit_staged_create_exact(", 1, WriteProtocol::Exact("sealed TableStorage create forwarding")), ("storage_layer.rs", ".commit_staged(", 1, WriteProtocol::Composed("sealed TableStorage forwarding")), ("storage_layer.rs", ".commit_staged_exact(", 1, WriteProtocol::Exact("sealed TableStorage forwarding")), - ("storage_layer.rs", ".dataset()", 26, WriteProtocol::Composed("sealed TableStorage forwarding")), - ("storage_layer.rs", ".into_arc()", 4, WriteProtocol::Composed("sealed TableStorage forwarding")), - ("storage_layer.rs", "SnapshotHandle::new(", 3, WriteProtocol::Composed("sealed TableStorage forwarding")), + ("storage_layer.rs", ".commit_staged_detached(", 1, WriteProtocol::Exact("sealed TableStorage forwarding")), + ("storage_layer.rs", ".promote_detached(", 1, WriteProtocol::Exact("sealed TableStorage forwarding")), + ("db/omnigraph/promotion.rs", ".promote_detached(", 1, WriteProtocol::Exact("RFC 0067 promotion replay")), + ("db/omnigraph/promotion.rs", "SnapshotHandle::new(", 1, WriteProtocol::ReadOnlyAccess), + ("db/omnigraph/optimize.rs", ".delete(", 2, WriteProtocol::Composed("RFC 0067 reap of a promoted pin's detached manifest after promotion, and of aged surplus detached manifests under the cleanup age policy")), + ("storage_layer.rs", ".dataset()", 28, WriteProtocol::Composed("sealed TableStorage forwarding")), + ("storage_layer.rs", ".into_arc()", 6, WriteProtocol::Composed("sealed TableStorage forwarding")), + ("storage_layer.rs", "SnapshotHandle::new(", 5, WriteProtocol::Composed("sealed TableStorage forwarding")), ("table_store.rs", ".raw_dataset_append(", 1, WriteProtocol::EphemeralScratch), ("table_store.rs", "Dataset::write(", 2, WriteProtocol::EphemeralScratch), ("table_store.rs", "DeleteBuilder::new(", 1, WriteProtocol::Composed("staged delete primitive")), ("table_store.rs", "InsertBuilder::new(", 3, WriteProtocol::Composed("staged insert primitive")), ("table_store.rs", "MergeInsertBuilder::try_new(", 1, WriteProtocol::Composed("staged merge primitive")), - ("table_store.rs", "CommitBuilder::new(", 2, WriteProtocol::Composed("staged commit primitive")), + ("table_store.rs", "CommitBuilder::new(", 4, WriteProtocol::Composed("staged commit primitive")), ("table_store.rs", ".create_index_builder(", 3, WriteProtocol::Composed("staged index primitive")), ("table_store.rs", ".execute_uncommitted(", 8, WriteProtocol::Composed("staged physical primitive")), ("exec/staging.rs", "write_sidecar(", 1, WriteProtocol::Exact("Mutation/Load v9")), @@ -877,12 +896,12 @@ durable_calls! { // pinned snapshot handles, plus the same-lineage path reading each // side's schema to resolve its system column spellings (RFC 0040 // Historical reads). Read-only — the diff stages and publishes nothing. - ("changes/mod.rs", ".dataset()", 4, WriteProtocol::ReadOnlyAccess), + ("changes/mod.rs", ".dataset()", 5, WriteProtocol::ReadOnlyAccess), ("db/omnigraph/schema_apply.rs", ".dataset()", 2, SCHEMA_V9), ("db/omnigraph/repair.rs", ".dataset()", 1, WriteProtocol::ManifestAdoption), // The sixth accessor reports deferred FTS coverage from an immutable // snapshot; it only reads index metadata and never stages or publishes. - ("db/omnigraph/optimize.rs", ".dataset()", 9, WriteProtocol::Composed("Optimize v9 planning + read-only coverage and native-fork inventory + physical cleanup")), + ("db/omnigraph/optimize.rs", ".dataset()", 10, WriteProtocol::Composed("Optimize v9 planning + read-only coverage and native-fork inventory + physical cleanup")), ("db/omnigraph/optimize.rs", ".into_dataset()", 2, OPTIMIZE_V9), ("db/omnigraph/optimize.rs", "SnapshotHandle::new(", 1, OPTIMIZE_V9), ("exec/merge.rs", "SnapshotHandle::new(", 5, MERGE_V9), @@ -899,6 +918,8 @@ const DURABLE_PRIMITIVES: &[&str] = &[ ".commit(", ".commit_staged_create_exact(", ".commit_staged_exact(", + ".commit_staged_detached(", + ".promote_detached(", ".commit_staged(", ".fork_branch_from_state(", "commit_updates_on_branch_with_expected(", diff --git a/crates/omnigraph/tests/lance_surface_guards.rs b/crates/omnigraph/tests/lance_surface_guards.rs index 5181ca177..3f889495a 100644 --- a/crates/omnigraph/tests/lance_surface_guards.rs +++ b/crates/omnigraph/tests/lance_surface_guards.rs @@ -4505,11 +4505,13 @@ async fn branch_ref_read_survives_concurrent_metadata_rewrite() { // linear commit rebases over a moved HEAD even at zero retries (why the // design stages detached); a detached commit from a pinned base ignores the // HEAD, never moves it, chains, and needs a retry budget; replaying a -// recorded transaction linearly produces an identical twin; and a replay -// over its own twin is refused only for the transaction kinds whose -// conflict rules see the twin, which decides what may be staged as its own -// detached commit. A Lance bump that changes any of them is a design -// review, not a test to weaken. +// recorded transaction linearly produces an identical twin, and every +// manifest names its transaction file `{read_version}-{uuid}.txn`, which is +// how the engine reads a version's transaction identity without a second +// request; and a replay over its own twin is refused only for the +// transaction kinds whose conflict rules see the twin, which decides what +// may be staged as its own detached commit. A Lance bump that changes any +// of them is a design review, not a test to weaken. fn rfc0067_schema() -> Arc { Arc::new(Schema::new(vec![ @@ -4758,6 +4760,11 @@ async fn rfc_0067_replaying_a_detached_transaction_promotes_an_identical_twin() recorded.uuid, transaction.uuid, "the detached manifest records the transaction" ); + assert_eq!( + d1.manifest().transaction_file.as_deref(), + Some(format!("1-{}.txn", transaction.uuid).as_str()), + "the manifest names its transaction file `{{read_version}}-{{uuid}}.txn`" + ); let promoted = rfc0067_linear(base.clone()) .execute(recorded) @@ -4768,6 +4775,11 @@ async fn rfc_0067_replaying_a_detached_transaction_promotes_an_identical_twin() promoted.read_transaction().await.unwrap().unwrap().uuid, transaction.uuid ); + assert_eq!( + promoted.manifest().transaction_file.as_deref(), + Some(format!("1-{}.txn", transaction.uuid).as_str()), + "the twin's manifest names the same transaction under the same base" + ); assert_eq!(rfc0067_ids(&promoted).await, rfc0067_ids(&d1).await); let d1_fragments: Vec = d1.get_fragments().iter().map(|f| f.id()).collect(); let twin_fragments: Vec = promoted.get_fragments().iter().map(|f| f.id()).collect(); @@ -4797,10 +4809,26 @@ async fn rfc_0067_replaying_a_detached_transaction_promotes_an_identical_twin() .unwrap(); let mut replay = d2.read_transaction().await.unwrap().unwrap(); assert_eq!(replay.read_version, d1.version().version); + assert_eq!( + d2.manifest().transaction_file.as_deref(), + Some(format!("{}-{}.txn", d1.version().version, replay.uuid).as_str()), + "a chained detached commit names its detached base in the file name" + ); replay.read_version = 2; let twin_base = rfc0067_pinned(uri, 2).await; let promoted2 = rfc0067_linear(twin_base).execute(replay).await.unwrap(); assert_eq!(promoted2.version().version, 3); + assert_eq!( + promoted2.manifest().transaction_file.as_deref(), + Some( + format!( + "2-{}.txn", + d2.read_transaction().await.unwrap().unwrap().uuid + ) + .as_str() + ), + "the chained twin names its linear base" + ); assert_eq!(rfc0067_ids(&promoted2).await, rfc0067_ids(&d2).await); let d2_fragments: Vec = d2.get_fragments().iter().map(|f| f.id()).collect(); let twin2_fragments: Vec = promoted2.get_fragments().iter().map(|f| f.id()).collect(); diff --git a/docs/dev/lance.md b/docs/dev/lance.md index 0be102ad6..0184584a3 100644 --- a/docs/dev/lance.md +++ b/docs/dev/lance.md @@ -175,6 +175,7 @@ history of dependency bumps. | Branch ref reads | Stock Lance 11 reads a ref as `head` then `get_range(0..size)`, so a `replace_metadata` landing between the two calls serves a prefix of the rewritten object and the read fails to parse instead of returning either version. `branch_control::read_branch_refs` rereads a torn read (`Arrow`, `CorruptFile`) or a raced delete (`NotFound`), at most three reads in all, around every engine ref read outside the writer gates (`list_branches`, `branches().get`, `get_identifier`); the two reads inside `retire_branch_recoverably` and `reclaim_ref_absent_tree` stay raw because the gate leaves them no concurrent rewriter. Resolving a logical branch name (`branch_control::resolve_live_native_branch`) lists `_refs/branches/` once and reads only that branch's own incarnations, so another branch's retirement rewrite, or its malformed retirement record, no longer fails a branch open or a held-ref re-resolve; only the operations that need every branch (list, create, delete, cleanup, upgrade, recovery) read every ref. Delete the retry arm and un-ignore the guard at the bump whose `from_path` reads a ref in one `get`. | `branch_control.rs` stale-head tests (they go red at that bump), `lance_surface_guards.rs::branch_ref_read_survives_concurrent_metadata_rewrite` (run with `--ignored`) | | Cleanup | Explicit cleanup discovers table identities through canonical manifest registrations, including tombstoned registrations, before inventorying native refs and trees. It protects exact live table refs, native ancestry, path dependencies, tags, and recovery pins. Only unprotected forks and retired leaves are reclaimed before version floors apply. Unreadable live roots abort cleanup; an invalid table inventory preserves that dataset and reports the failure. Writes and branch deletion defer physical reclamation. | `maintenance.rs`, `branching.rs` | | Table forks | First touch uses `fork.{incarnation}.m{base_version}.{commit_ULID}`, at most 80 ASCII bytes, independently of logical-name length. Legacy owners use `legacy` in the incarnation position. The existing commit ULID distinguishes attempts; registration metadata proves ownership, and source-pointer adoption preserves it. Missing legacy ownership requires exact ref equality. Naming and the existing live-only write identifier lookup add no storage requests. | `long_branch_names_first_touch.gqt`, `branching.rs`, `failpoints.rs`, metadata tests | +| Detached commits | RFC 0067 stages a table pin as a Lance detached commit (`_versions/d{version}.manifest`), which never moves HEAD and needs no conflict pass; promotion replays the recorded transaction linearly at zero retries, and a replay over an already-landed twin is refused for every kind the engine stages detached. Every manifest names its transaction file `{read_version}-{uuid}.txn`, which is how the engine identifies a pin's twin and follows a chain of detached commits without a second request. | `lance_surface_guards.rs` (`rfc_0067_*`), `db/omnigraph/promotion.rs` tests | | MemWAL | Upstream support exists, but OmniGraph's RFC 0018 and RFC 0026 experiments were removed. No stream profile, token ledger, hidden stream column, or `_mem_wal` path is current. | `lifecycle.rs`, cluster removed-field diagnostics | The clone adapter is installed during normal dataset opens and initial dataset diff --git a/docs/rfcs/0067-detached-table-commits.md b/docs/rfcs/0067-detached-table-commits.md index e2efce6b5..a81eabf87 100644 --- a/docs/rfcs/0067-detached-table-commits.md +++ b/docs/rfcs/0067-detached-table-commits.md @@ -208,7 +208,13 @@ cannot be read (Lance 12 returns an error for undecodable transaction bytes and `None` only when none was recorded), the promotion is blocked; open `staged_version` and report the block. A promoter applies the same rule and never treats an unreadable transaction as its own. Detached versions are -immutable and cache like any version. +immutable and cache like any version. The uuid check costs no request +beyond the manifest: Lance records the transaction file name, +`{read_version}-{uuid}.txn`, in every manifest, so the pin resolver, both +witnesses and the chain walk read a version's transaction identity from the +opened manifest, and only promotion reads the transaction it replays. A +manifest that names no transaction file is treated as foreign; +`lance_surface_guards` pins the name. Every check that today compares a pin's manifest e-tag with the opened dataset's manifest (the Blob facade's integrity check and the change feed's