diff --git a/crates/omnigraph/src/changes/candidate_scan.rs b/crates/omnigraph/src/changes/candidate_scan.rs new file mode 100644 index 00000000..552c7a70 --- /dev/null +++ b/crates/omnigraph/src/changes/candidate_scan.rs @@ -0,0 +1,773 @@ +//! Candidate-pruning optimization for the per-commit change enumerator +//! (RFC-030 §4.2/§4.3). +//! +//! The authority path in [`super::enumerate`] derives one commit's changes by a +//! full ordered-by-id merge of both pinned dataset versions — O(dataset +//! extent). When one adjacent Lance transaction has a mature, +//! row-set-preserving insert/update shape, the same logical changes can be +//! derived from that transaction's physical footprint: candidate rows come +//! from the newly assigned child-fragment suffix and before-images come from +//! only the parent fragments the transaction updated or removed. Both sides +//! are merged as ordered streams. This module decides, per interval, whether +//! that optimization is available; the caller falls back to the exact full +//! merge on any doubt. +//! +//! **Why the pruned path needs no delete handling.** Eligibility first requires +//! one graph-visible dataset interval to advance by exactly one Lance version. +//! The parse-time D2 rule keeps inserts/updates and deletes out of the same +//! mutation, so an engine-authored adjacent transaction is either a +//! row-set-preserving insert/update or a delete — never both. If that one +//! transaction is an `Append` or a **provably** row-set-preserving merge +//! `Update`, no live logical id can disappear and the candidate scan is +//! complete. Any wider interval or operation that can remove, reuse, or +//! re-stamp rows (`Delete`, `Overwrite`, `Restore`, compaction `Rewrite`, …) +//! falls back to the exact merge, which classifies deletes correctly. +//! +//! A `RewriteRows` `Update` is trusted as row-set-preserving only with a +//! **durable per-transaction provenance proof** — the `omnigraph.no_by_source_delete` +//! marker every general keyed MergeInsert update stamps, or the RFC-023 +//! `insert_absence` certificate that proven strict inserts carry instead. The +//! op shape alone is not enough: `repair --force --confirm` can +//! adopt an external Lance merge whose delete-capable by-source arm persists as +//! `Update { RewriteRows }`, and its child-only candidate scan has no delete +//! pass. The source-walk guard (`no_delete_capable_merge_arm_in_engine_source`) +//! proves only that *current engine code* builds no such arm; the marker +//! authenticates the *persisted* transaction. An unproven `Update` falls back. +//! See [`transaction_is_row_set_preserving`]. + +use datafusion::prelude::{col, lit}; +use lance::Dataset; +use lance::dataset::transaction::{Operation, Transaction, UpdateMode}; +use lance_table::format::Fragment; + +use super::enumerate::{Emit, next_emit}; +use super::model::ChangeFeedScope; +use super::row_compare::{OrderedRows, ScanTargets, rows_equal}; +use crate::db::DatasetEntry; +use crate::error::Result; +use crate::table_store::{has_insert_absence_certificate, has_no_by_source_delete_marker}; + +/// Whether one Lance transaction's operation preserves the live logical row set +/// — i.e. can only add or modify rows in place, never remove, reuse, or re-stamp +/// a logical id. Only such operations are safe to derive from candidate and +/// transaction-touched parent fragments; everything else forces the exact +/// ordered merge. +/// +/// The match is exhaustive with **no wildcard arm**: a new Lance `Operation` +/// variant is a compile error that forces this classification to be reviewed +/// (RFC-030 §9 — new variants must fall back until reviewed). +pub(crate) fn operation_is_row_set_preserving(operation: &Operation) -> bool { + match operation { + // Append only adds fragments; it never removes or reuses a logical id. + Operation::Append { .. } => true, + // OmniGraph's keyed writes (strict-insert / upsert / known-present + // update) are all `RewriteRows` merge_insert and never delete an + // unmatched-by-source row — no delete-capable by-source merge arm exists + // in the engine (locked by the write-path guard in forbidden_apis.rs). A + // different update mode is a foreign or unknown shape, so fall back. + Operation::Update { update_mode, .. } => update_mode == &Some(UpdateMode::RewriteRows), + // Everything that can remove, reuse, or re-stamp rows falls back to the + // exact ordered merge. Listed explicitly so a new variant fails to + // compile here. + Operation::Delete { .. } + | Operation::Overwrite { .. } + | Operation::CreateIndex { .. } + | Operation::Rewrite { .. } + | Operation::DataReplacement { .. } + | Operation::DataOverlay { .. } + | Operation::Merge { .. } + | Operation::Restore { .. } + | Operation::ReserveFragments { .. } + | Operation::Project { .. } + | Operation::UpdateConfig { .. } + | Operation::UpdateMemWalState { .. } + | Operation::Clone { .. } + | Operation::UpdateBases { .. } => false, + } +} + +/// Whether one transaction is safe to derive by the touched-fragment candidate +/// path. +/// +/// It must have a row-set-preserving operation SHAPE +/// ([`operation_is_row_set_preserving`]) AND, for a `RewriteRows` `Update`, a +/// durable OmniGraph provenance proof that it removed no rows: either the +/// no-by-source-delete marker every keyed write stamps, or the RFC-023 +/// `insert_absence` certificate (a pure insert deletes nothing). `Append` is +/// unconditionally additive and needs no marker. +/// +/// The shape check alone is NOT sufficient: `repair --force --confirm` can adopt +/// an external Lance merge whose delete-capable by-source arm persists as +/// `Operation::Update { RewriteRows }`. Such a transaction carries neither proof, +/// so it falls back to the exact ordered merge — whose delete pass reports the +/// removed rows the child-only candidate scan would miss. +pub(crate) fn transaction_is_row_set_preserving(transaction: &Transaction) -> bool { + if !operation_is_row_set_preserving(&transaction.operation) { + return false; + } + match &transaction.operation { + Operation::Append { .. } => true, + Operation::Update { .. } => { + has_no_by_source_delete_marker(transaction) + || has_insert_absence_certificate(transaction) + } + // Unreachable: the shape guard above already rejected every other + // variant. Keeping the match total means a future variant that + // `operation_is_row_set_preserving` starts accepting must be classified + // here too, rather than silently pruning. + _ => false, + } +} + +/// Immutable physical plan for a proven adjacent candidate interval. Both +/// vectors are bounded by the one transaction's touched-fragment footprint; +/// neither is a copy of the full manifest. +#[derive(Debug, Clone)] +pub(crate) struct CandidatePlan { + child_fragments: Vec, + parent_fragments: Vec, +} + +/// Return the candidate plan for one adjacent, row-set-preserving Lance +/// transaction, or `None` to use the exact ordered merge. +/// +/// The adjacency requirement is deliberate. A stateless page may be resumed +/// many times; walking up to 1,024 historical transactions on every page would +/// multiply history work by page count. An interval wider than one version now +/// falls back *before any transaction read*. For the accepted shape we read +/// only the already-open child's one transaction and require its `read_version` +/// to name the pinned parent exactly. +/// +/// Fragment discovery uses Lance's manifest invariants: fragments are sorted +/// by id, ids never recycle, and one Append/Update assigns every new fragment +/// consecutively above the parent's high-water mark. The child suffix is found +/// by binary search; affected parent fragments are found by one binary search +/// per transaction-reported id. Work is therefore +/// O(log(manifest) + touched_fragments log(manifest)), with allocations bounded +/// by touched fragments rather than total dataset extent. +/// +/// `read_transaction` follows the transaction reference already captured in +/// the pinned child manifest; Lance transaction objects are UUID-named. The +/// sole caller nevertheless stores the complete plan before the final +/// named-branch head witness, and emission performs no later history lookup. +pub(crate) async fn interval_candidate_plan( + from_entry: &DatasetEntry, + to_entry: &DatasetEntry, + from_dataset: &Dataset, + to_dataset: &Dataset, +) -> Result> { + if from_entry.native_dataset_branch != to_entry.native_dataset_branch + || from_entry.identity != to_entry.identity + { + return Ok(None); + } + if from_entry.published_dataset_version.checked_add(1) + != Some(to_entry.published_dataset_version) + { + return Ok(None); + } + if to_dataset.version().version != to_entry.published_dataset_version + || from_dataset.version().version != from_entry.published_dataset_version + || !to_dataset.manifest.uses_stable_row_ids() + || !from_dataset.manifest.uses_stable_row_ids() + { + return Ok(None); + } + + crate::instrumentation::record_candidate_transaction_read(); + let Ok(Some(transaction)) = to_dataset.read_transaction().await else { + return Ok(None); + }; + if transaction.read_version != from_entry.published_dataset_version + || !transaction_is_row_set_preserving(&transaction) + { + return Ok(None); + } + + Ok(candidate_plan_from_transaction( + &transaction.operation, + from_dataset, + to_dataset, + )) +} + +fn candidate_plan_from_transaction( + operation: &Operation, + from_dataset: &Dataset, + to_dataset: &Dataset, +) -> Option { + let (new_fragment_count, mut parent_fragment_ids) = match operation { + Operation::Append { fragments } => (fragments.len(), Vec::new()), + Operation::Update { + removed_fragment_ids, + updated_fragments, + new_fragments, + .. + } => { + let mut ids = Vec::with_capacity( + removed_fragment_ids + .len() + .saturating_add(updated_fragments.len()), + ); + ids.extend(removed_fragment_ids.iter().copied()); + ids.extend(updated_fragments.iter().map(|fragment| fragment.id)); + (new_fragments.len(), ids) + } + _ => return None, + }; + parent_fragment_ids.sort_unstable(); + parent_fragment_ids.dedup(); + + let parent_high_water = from_dataset.manifest.max_fragment_id(); + let (suffix_start, mut metadata_steps) = + first_fragment_after(to_dataset.fragments(), parent_high_water); + let child_fragments = &to_dataset.fragments()[suffix_start..]; + metadata_steps = metadata_steps.saturating_add(child_fragments.len() as u64); + if child_fragments.len() != new_fragment_count { + crate::instrumentation::record_candidate_fragment_metadata_steps(metadata_steps); + return None; + } + + let expected_first = match parent_high_water { + Some(id) => id.checked_add(1)?, + None => 0, + }; + for (offset, fragment) in child_fragments.iter().enumerate() { + let expected_id = expected_first.checked_add(u64::try_from(offset).ok()?)?; + if fragment.id != expected_id { + crate::instrumentation::record_candidate_fragment_metadata_steps(metadata_steps); + return None; + } + } + let expected_high_water = if new_fragment_count == 0 { + parent_high_water + } else { + Some(expected_first.checked_add(u64::try_from(new_fragment_count - 1).ok()?)?) + }; + if to_dataset.manifest.max_fragment_id() != expected_high_water { + crate::instrumentation::record_candidate_fragment_metadata_steps(metadata_steps); + return None; + } + + let mut parent_fragments = Vec::with_capacity(parent_fragment_ids.len()); + for fragment_id in parent_fragment_ids { + let (fragment, steps) = find_fragment(from_dataset.fragments(), fragment_id); + metadata_steps = metadata_steps.saturating_add(steps); + let Some(fragment) = fragment else { + crate::instrumentation::record_candidate_fragment_metadata_steps(metadata_steps); + return None; + }; + parent_fragments.push(fragment.clone()); + } + + // Row-version metadata is correctness-bearing on the pruned path: the + // candidate scan filters on `_row_last_updated_at_version`, and pinned + // Lance 10 silently fills that column with 1 for a fragment whose sequence + // is missing OR fails to load ("Default to version 1 if sequence not + // provided" in lance-table's stream reader — a failed `load_sequence()` is + // swallowed the same way). For any interval with `begin > 1` such rows + // fall outside the candidate window and real updates vanish without an + // error. Require every changed fragment to carry loadable, structurally + // valid last-updated metadata; any gap is a normal miss that falls back to + // the exact ordered merge (which does not consume the version column). + if !child_fragments + .iter() + .all(fragment_version_metadata_is_loadable) + { + crate::instrumentation::record_candidate_fragment_metadata_steps(metadata_steps); + return None; + } + crate::instrumentation::record_candidate_fragment_metadata_steps(metadata_steps); + Some(CandidatePlan { + child_fragments: child_fragments.to_vec(), + parent_fragments, + }) +} + +/// Binary-search the first manifest fragment above `high_water`, counting the +/// metadata comparisons for the checked-in fragment-scaling cost gate. +fn first_fragment_after(fragments: &[Fragment], high_water: Option) -> (usize, u64) { + let Some(high_water) = high_water else { + return (0, 0); + }; + let mut left = 0usize; + let mut right = fragments.len(); + let mut steps = 0u64; + while left < right { + steps = steps.saturating_add(1); + let middle = left + (right - left) / 2; + if fragments[middle].id <= high_water { + left = middle + 1; + } else { + right = middle; + } + } + (left, steps) +} + +fn find_fragment(fragments: &[Fragment], id: u64) -> (Option<&Fragment>, u64) { + let mut left = 0usize; + let mut right = fragments.len(); + let mut steps = 0u64; + while left < right { + steps = steps.saturating_add(1); + let middle = left + (right - left) / 2; + match fragments[middle].id.cmp(&id) { + std::cmp::Ordering::Less => left = middle + 1, + std::cmp::Ordering::Greater => right = middle, + std::cmp::Ordering::Equal => return (Some(&fragments[middle]), steps), + } + } + (None, steps) +} + +/// Whether one changed fragment's `_row_last_updated_at_version` sequence is +/// present, decodable, and complete — the §4.2 "genuinely active" requirement +/// at the fragment level. `uses_stable_row_ids()` alone is a dataset-level +/// flag and cannot prove a specific fragment's sequence survives loading. +/// +/// The Inline bytes are decoded directly rather than through +/// `load_sequence()`: pinned Lance 10's External arm of that method is +/// `todo!()` (it panics, it does not `Err`), so the variant must be +/// structurally unreachable here. And a clean decode alone does not prove the +/// sequence covers the fragment — Lance's single-run fast path stamps its one +/// version across every requested row without consulting the encoded length — +/// so the sequence must be exactly `physical_rows` long. Absent, external, +/// undecodable, short, or unmeasurable all mean "not provably complete" and +/// route the interval to the exact ordered merge. +fn fragment_version_metadata_is_loadable(fragment: &Fragment) -> bool { + let Some(lance_table::rowids::version::RowDatasetVersionMeta::Inline(data)) = + fragment.last_updated_at_version_meta.as_ref() + else { + return false; + }; + let Ok(sequence) = lance_table::rowids::version::read_dataset_versions(data) else { + return false; + }; + fragment + .physical_rows + .is_some_and(|rows| sequence.len() == rows as u64) +} + +/// Emitter for a proven adjacent row-set-preserving interval. Candidate child +/// rows and the transaction-touched parent fragments are both scanned in id +/// order and merged one row at a time. No BTREE is required, so a missing or +/// partially covered index cannot turn the parent lookup into a hidden +/// full-dataset scan. At most one prepared row from either stream is retained; +/// scanner batch targets come from the current page budget. +pub(crate) struct CandidateUpserts { + parent_dataset: Dataset, + parents: Option, + candidates: OrderedRows, + scope: ChangeFeedScope, +} + +impl CandidateUpserts { + async fn open( + from_entry: &DatasetEntry, + to_entry: &DatasetEntry, + from_dataset: Dataset, + to_dataset: Dataset, + plan: CandidatePlan, + after_id: Option<&str>, + scope: ChangeFeedScope, + scan_targets: ScanTargets, + ) -> Result { + crate::instrumentation::record_candidate_scan_targets( + scan_targets.rows(), + scan_targets.bytes(), + ); + // Scan only the new fragments this commit wrote, and within them + // keep rows whose last update lands in (begin, end] — this drops the + // carried-over rows a fragment rewrite pulled along, leaving exactly the + // inserted and updated rows (the touched-parent merge classifies which). + let window = col("_row_last_updated_at_version") + .gt(lit(from_entry.published_dataset_version)) + .and( + col("_row_last_updated_at_version").lt_eq(lit(to_entry.published_dataset_version)), + ); + let candidates = OrderedRows::open_scan( + to_dataset, + after_id, + Some(window), + Some(plan.child_fragments), + scan_targets, + ) + .await?; + let parents = if plan.parent_fragments.is_empty() { + None + } else { + Some( + OrderedRows::open_scan( + from_dataset.clone(), + after_id, + None, + Some(plan.parent_fragments), + scan_targets, + ) + .await?, + ) + }; + Ok(Self { + parent_dataset: from_dataset, + parents, + candidates, + scope, + }) + } + + fn parent_dataset(&self) -> &Dataset { + &self.parent_dataset + } + + fn child_dataset(&self) -> &Dataset { + self.candidates.dataset() + } + + async fn next(&mut self) -> Result> { + loop { + let Some(candidate) = self.candidates.pop().await? else { + return Ok(None); + }; + crate::instrumentation::record_candidate_row_examined(); + + let mut before = None; + if let Some(parents) = self.parents.as_mut() { + loop { + let parent_id = parents.peek().await?.map(|row| row.id.clone()); + match parent_id { + Some(parent_id) if parent_id < candidate.id => { + // An unrelated row carried by a touched source + // fragment; it cannot be this candidate's before image. + parents.pop().await?; + } + Some(parent_id) if parent_id == candidate.id => { + before = parents.pop().await?; + break; + } + _ => break, + } + } + } + + let emit = match before { + None => Emit::Insert(candidate), + Some(before) => { + if rows_equal( + &self.parent_dataset, + &before, + self.candidates.dataset(), + &candidate, + ) + .await? + { + continue; + } + Emit::Update { + before, + after: candidate, + } + } + }; + if self.scope.wants_op(emit.op()) { + return Ok(Some(emit)); + } + } + } +} + +/// Per-interval change emitter: the touched-fragment candidate path when the +/// interval is provably row-set-preserving, else the exact full ordered merge. +/// Both yield the same id-ordered `Emit` stream; before-images come from the +/// parent handle and after-images from the child handle. +pub(crate) struct FullMergeRows { + from: OrderedRows, + to: OrderedRows, + scope: ChangeFeedScope, +} + +pub(crate) enum EmitSource { + FullMerge(Box), + Pruned(Box), +} + +impl EmitSource { + /// Open the emitter for one interval. `candidate_plan` is the pruning + /// decision [`interval_candidate_plan`] computed in `plan_intervals` before + /// (and therefore covered by) the final head witness. This constructor + /// performs no live history read, so a branch delete/recreate after the + /// witness cannot reroute the interval. + pub(crate) async fn plan( + from_entry: &DatasetEntry, + to_entry: &DatasetEntry, + from_dataset: Dataset, + to_dataset: Dataset, + candidate_plan: Option, + after_id: Option<&str>, + scope: &ChangeFeedScope, + scan_targets: ScanTargets, + ) -> Result { + if let Some(candidate_plan) = candidate_plan { + Ok(Self::Pruned(Box::new( + CandidateUpserts::open( + from_entry, + to_entry, + from_dataset, + to_dataset, + candidate_plan, + after_id, + scope.clone(), + scan_targets, + ) + .await?, + ))) + } else { + let from = OrderedRows::open(from_dataset, after_id).await?; + let to = OrderedRows::open(to_dataset, after_id).await?; + Ok(Self::FullMerge(Box::new(FullMergeRows { + from, + to, + scope: scope.clone(), + }))) + } + } + + pub(crate) async fn next(&mut self) -> Result> { + match self { + Self::FullMerge(full) => next_emit(&mut full.from, &mut full.to, &full.scope).await, + Self::Pruned(candidates) => candidates.next().await, + } + } + + pub(crate) fn parent_dataset(&self) -> &Dataset { + match self { + Self::FullMerge(full) => full.from.dataset(), + Self::Pruned(candidates) => candidates.parent_dataset(), + } + } + + pub(crate) fn child_dataset(&self) -> &Dataset { + match self { + Self::FullMerge(full) => full.to.dataset(), + Self::Pruned(candidates) => candidates.child_dataset(), + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + use lance::dataset::transaction::Operation; + use lance_table::format::Fragment; + + fn update(mode: Option) -> Operation { + Operation::Update { + removed_fragment_ids: Vec::new(), + updated_fragments: Vec::new(), + new_fragments: vec![Fragment::new(0)], + fields_modified: Vec::new(), + compacted_sstables: Vec::new(), + fields_for_preserving_frag_bitmap: Vec::new(), + update_mode: mode, + inserted_rows_filter: None, + updated_fragment_offsets: None, + } + } + + #[test] + fn append_and_rewrite_rows_update_are_row_set_preserving() { + assert!(operation_is_row_set_preserving(&Operation::Append { + fragments: vec![Fragment::new(7)], + })); + // A merge Update that also modifies existing rows (non-empty + // updated_fragments / removed_fragment_ids) is still row-set-preserving — + // unlike the pure-insert certificate, this classifier accepts updates. + let mut upsert = update(Some(UpdateMode::RewriteRows)); + if let Operation::Update { + updated_fragments, + removed_fragment_ids, + .. + } = &mut upsert + { + updated_fragments.push(Fragment::new(1)); + removed_fragment_ids.push(1); + } + assert!(operation_is_row_set_preserving(&upsert)); + } + + #[test] + fn foreign_update_mode_and_removing_ops_fall_back() { + // A non-RewriteRows update mode is a foreign/unknown shape. + assert!(!operation_is_row_set_preserving(&update(Some( + UpdateMode::RewriteColumns + )))); + assert!(!operation_is_row_set_preserving(&update(None))); + // Operations that can remove, reuse, or re-stamp rows. + assert!(!operation_is_row_set_preserving(&Operation::Delete { + updated_fragments: Vec::new(), + deleted_fragment_ids: Vec::new(), + predicate: String::new(), + })); + assert!(!operation_is_row_set_preserving(&Operation::Restore { + version: 0 + })); + assert!(!operation_is_row_set_preserving( + &Operation::ReserveFragments { num_fragments: 1 } + )); + } + + fn txn(operation: Operation, properties: &[(&str, &str)]) -> Transaction { + let transaction_properties = (!properties.is_empty()).then(|| { + std::sync::Arc::new( + properties + .iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect::>(), + ) + }); + Transaction { + read_version: 0, + uuid: "test".to_string(), + operation, + tag: None, + transaction_properties, + } + } + + #[test] + fn transaction_update_prunes_only_with_a_durable_no_delete_proof() { + use crate::table_store::{ + INSERT_ABSENCE_PROPERTY, INSERT_ABSENCE_V1, NO_BY_SOURCE_DELETE_PROPERTY, + NO_BY_SOURCE_DELETE_V1, + }; + // Append is unconditionally additive — no marker required. + assert!(transaction_is_row_set_preserving(&txn( + Operation::Append { + fragments: vec![Fragment::new(1)], + }, + &[], + ))); + // A RewriteRows Update prunes with the keyed-write no-delete marker ... + assert!(transaction_is_row_set_preserving(&txn( + update(Some(UpdateMode::RewriteRows)), + &[(NO_BY_SOURCE_DELETE_PROPERTY, NO_BY_SOURCE_DELETE_V1)], + ))); + // ... or the RFC-023 insert_absence certificate (a pure insert deletes + // nothing) ... + assert!(transaction_is_row_set_preserving(&txn( + update(Some(UpdateMode::RewriteRows)), + &[(INSERT_ABSENCE_PROPERTY, INSERT_ABSENCE_V1)], + ))); + // ... but NOT without a durable proof. An external Lance merge with a + // delete-capable by-source arm, adopted via `repair --force`, persists + // this exact `Update{RewriteRows}` shape and carries no marker — this is + // the data-loss regression the fix closes (the op-shape classifier alone + // returns true here). + assert!(operation_is_row_set_preserving(&update(Some( + UpdateMode::RewriteRows + )))); + assert!(!transaction_is_row_set_preserving(&txn( + update(Some(UpdateMode::RewriteRows)), + &[], + ))); + // An unrelated property is not a no-delete proof. + assert!(!transaction_is_row_set_preserving(&txn( + update(Some(UpdateMode::RewriteRows)), + &[("lance.something", "x")], + ))); + // The shape guard still fences a delete even if a marker were spoofed + // onto it, so a stray marker can never make a removing op prune. + assert!(!transaction_is_row_set_preserving(&txn( + Operation::Delete { + updated_fragments: Vec::new(), + deleted_fragment_ids: Vec::new(), + predicate: String::new(), + }, + &[(NO_BY_SOURCE_DELETE_PROPERTY, NO_BY_SOURCE_DELETE_V1)], + ))); + } + + #[test] + fn fragment_discovery_is_binary_search_plus_delta() { + let fragments = (0..65_536).map(Fragment::new).collect::>(); + + let (suffix, suffix_steps) = first_fragment_after(&fragments, Some(65_530)); + assert_eq!(suffix, 65_531); + assert!( + suffix_steps <= 17, + "65k-fragment suffix lookup must stay logarithmic, got {suffix_steps} steps" + ); + + let (fragment, lookup_steps) = find_fragment(&fragments, 42_424); + assert_eq!(fragment.map(|fragment| fragment.id), Some(42_424)); + assert!( + lookup_steps <= 17, + "65k-fragment parent lookup must stay logarithmic, got {lookup_steps} steps" + ); + } + + #[test] + fn missing_row_version_metadata_is_not_loadable() { + // Pinned Lance 10 fills `_row_last_updated_at_version` with 1 when a + // fragment's sequence is missing or fails to load, which would silently + // empty the candidate window for begin > 1. A fragment without the + // metadata (Lance's `Fragment::new` default) must therefore fail the + // loadability gate so the interval falls back to the exact merge. The + // positive case — every real OmniGraph-written changed fragment carries + // a loadable sequence — is proven end-to-end by the pruned cost/image + // tests staying flat/green (they would fall back and fail otherwise). + assert!(!fragment_version_metadata_is_loadable(&Fragment::new(7))); + } + + #[test] + fn external_row_version_metadata_is_not_loadable() { + // Pinned Lance 10's `RowDatasetVersionMeta::External` arm of + // `load_sequence()` is `todo!()` — it panics rather than returning + // `Err`. The gate must classify the variant structurally instead of + // probing it, so an externally-stored sequence routes the interval to + // the exact ordered merge instead of aborting the poll. + let mut fragment = Fragment::new(7); + fragment.last_updated_at_version_meta = Some( + lance_table::rowids::version::RowDatasetVersionMeta::External( + lance_table::format::ExternalFile { + path: "external.versions".to_string(), + offset: 0, + size: 64, + }, + ), + ); + assert!(!fragment_version_metadata_is_loadable(&fragment)); + } + + #[test] + fn short_row_version_sequence_is_not_loadable() { + // A sequence can decode cleanly yet cover fewer rows than the fragment + // holds; pinned Lance 10's single-run fast path then stamps that run's + // version across every requested row without consulting the encoded + // length. Loadable therefore means decodable AND exactly + // `physical_rows` long — anything shorter (or a fragment that does not + // even record its physical row count) falls back to the exact merge. + use lance_table::rowids::version::{ + RowDatasetVersionMeta, RowDatasetVersionSequence, write_dataset_versions, + }; + + let mut fragment = Fragment::new(7); + fragment.physical_rows = Some(5); + let short = RowDatasetVersionSequence::from_uniform_row_count(3, 42); + fragment.last_updated_at_version_meta = Some(RowDatasetVersionMeta::Inline( + write_dataset_versions(&short).into(), + )); + assert!(!fragment_version_metadata_is_loadable(&fragment)); + + let exact = RowDatasetVersionSequence::from_uniform_row_count(5, 42); + fragment.last_updated_at_version_meta = Some(RowDatasetVersionMeta::Inline( + write_dataset_versions(&exact).into(), + )); + assert!(fragment_version_metadata_is_loadable(&fragment)); + + fragment.physical_rows = None; + assert!(!fragment_version_metadata_is_loadable(&fragment)); + } +} diff --git a/crates/omnigraph/src/changes/enumerate.rs b/crates/omnigraph/src/changes/enumerate.rs index ebcd3f86..bff8e635 100644 --- a/crates/omnigraph/src/changes/enumerate.rs +++ b/crates/omnigraph/src/changes/enumerate.rs @@ -17,13 +17,15 @@ use std::collections::BTreeSet; use lance::Dataset; +use super::candidate_scan::{CandidatePlan, EmitSource}; use super::model::{ COMMIT_CHANGES_MAX_BYTES, ChangeEntityKind, ChangeFeedScope, ChangeOpKind, EntityEndpoints, EntityImage, GraphEntityChange, GraphTypeRef, }; -use super::row_compare::{OrderedRows, RawRow, rows_equal, user_schema_fingerprint}; +use super::row_compare::{OrderedRows, RawRow, ScanTargets, rows_equal, user_schema_fingerprint}; use super::token::{cursor_rejected, opaque_type_id}; use super::{changed_table_intervals, parse_table_key}; +use crate::db::DatasetEntry; use crate::db::logical_row_image; use crate::db::manifest::Snapshot; use crate::error::{OmniError, Result}; @@ -126,6 +128,7 @@ async fn emitted_image( raw: &RawRow, kind: ChangeEntityKind, ) -> Result { + crate::instrumentation::record_change_image_materialized(); let mut properties = logical_row_image(dataset, &raw.slice, 0).await?; properties.remove("id"); let endpoints = if kind == ChangeEntityKind::Edge { @@ -147,14 +150,14 @@ async fn emitted_image( }) } -enum Emit { +pub(crate) enum Emit { Delete(RawRow), Insert(RawRow), Update { before: RawRow, after: RawRow }, } impl Emit { - fn op(&self) -> ChangeOpKind { + pub(crate) fn op(&self) -> ChangeOpKind { match self { Self::Insert(_) => ChangeOpKind::Insert, Self::Update { .. } => ChangeOpKind::Update, @@ -166,7 +169,7 @@ impl Emit { /// The next in-scope logical change in the ordered id merge, or `None` when /// both sides are exhausted. Equal rows and out-of-scope operations are /// consumed without image or payload work. -async fn next_emit( +pub(crate) async fn next_emit( from: &mut OrderedRows, to: &mut OrderedRows, scope: &ChangeFeedScope, @@ -206,14 +209,23 @@ async fn next_emit( /// One paired table lifetime that survived the schema gate, with both pinned /// datasets already open (the same handles the scans consume, so a changed /// interval costs at most two opens per page). -struct IntervalPlan { +pub(crate) struct IntervalPlan { /// The published opaque type identity: the block ordering key, the /// continuation key, and `GraphTypeRef.id`, all one value. opaque_id: String, kind: ChangeEntityKind, type_name: String, - from_dataset: Dataset, - to_dataset: Dataset, + /// The paired manifest entries (begin/end version, branch, identity). + pub(crate) from_entry: DatasetEntry, + pub(crate) to_entry: DatasetEntry, + pub(crate) from_dataset: Dataset, + pub(crate) to_dataset: Dataset, + /// The candidate-pruning decision, computed by `plan_intervals` BEFORE the + /// final post-open head witness. The adjacent transaction is referenced by + /// the already-pinned child manifest, and `Some` stores the complete + /// transaction-touched parent/child fragment plan so emission performs no + /// later history lookup. `None` means the exact ordered merge. + pub(crate) candidate_plan: Option, } /// Resolve the exceptional bounded digest position to its exact logical ID. @@ -288,6 +300,7 @@ async fn plan_intervals( child: &Snapshot, schema_identity_domain: &str, graph_commit_id: &str, + scope: &ChangeFeedScope, ) -> Result> { let intervals = changed_table_intervals(parent, child); @@ -329,12 +342,33 @@ async fn plan_intervals( return Err(schema_boundary(graph_commit_id, table_key)); } let (kind, type_name) = parse_table_key(table_key); + let kind: ChangeEntityKind = kind.into(); + // Classify the interval HERE — before the head witness below — + // and retain the complete physical plan. The one transaction is + // referenced by the already-pinned child manifest; no history + // lookup is permitted later during emission. Scope-filtered + // intervals are never emitted, so their stored decision is + // irrelevant. + let candidate_plan = if scope.wants_kind(kind) && scope.wants_type_name(type_name) { + super::candidate_scan::interval_candidate_plan( + from, + to, + &from_dataset, + &to_dataset, + ) + .await? + } else { + None + }; plans.push(IntervalPlan { opaque_id: opaque_type_id(schema_identity_domain, interval.identity), - kind: kind.into(), + kind, type_name: type_name.to_string(), + from_entry: from.clone(), + to_entry: to.clone(), from_dataset, to_dataset, + candidate_plan, }); } (None, None) => unreachable!("changed intervals have at least one endpoint"), @@ -356,6 +390,12 @@ async fn plan_intervals( // cannot undergo branch-name ABA and pays no extra manifest resolution. reprove_named_branch_heads(store, parent, &parent_branches).await?; reprove_named_branch_heads(store, child, &child_branches).await?; + // Nothing after this witness may read the branch's numeric-path history + // live: version manifests sit at replaceable numeric paths (unlike + // UUID-named data and transaction files), so a later live read would see a + // recreated branch's history under this commit's label. Tests park here + // and recreate the branch to pin that contract. + crate::failpoints::maybe_fail(crate::failpoints::names::CHANGE_FEED_POST_HEAD_WITNESS)?; // The opaque ids are domain-scoped SHA-256 projections of distinct // immutable identities, so this order is total and deterministic. plans.sort_by(|left, right| { @@ -416,6 +456,7 @@ pub(crate) async fn enumerate_commit_changes( child, schema_identity_domain, graph_commit_id, + scope, ) .await?; @@ -458,23 +499,55 @@ pub(crate) async fn enumerate_commit_changes( id: plan.opaque_id.clone(), name: plan.type_name.clone(), }; - let mut left = OrderedRows::open(plan.from_dataset, after_id.as_deref()).await?; - let mut right = OrderedRows::open(plan.to_dataset, after_id.as_deref()).await?; + // Per-interval emitter: the touched-fragment candidate path when the + // commit's effect is a proven adjacent row-set-preserving shape, else + // the exact full ordered merge. Both yield the same id-ordered `Emit` + // stream, so the budgeting/continuation loop below is identical. + // Before-images come from the parent handle, after-images from the + // child handle. The pruning decision itself was made by + // `plan_intervals` before, and covered by, the final head witness — no + // live history read happens here. + let mut source = EmitSource::plan( + &plan.from_entry, + &plan.to_entry, + plan.from_dataset, + plan.to_dataset, + plan.candidate_plan, + after_id.as_deref(), + scope, + ScanTargets::for_page(budget.remaining_rows, budget.remaining_bytes), + ) + .await?; - while let Some(emit) = next_emit(&mut left, &mut right, scope).await? { + while let Some(emit) = source.next().await? { + // The source yields one look-ahead change so we can distinguish a + // complete block from a truncated one. If the page's row budget (or + // an already-used byte budget) is exhausted, that sentinel must not + // materialize JSON or Blob payloads merely to prove continuation. + if budget.remaining_rows == 0 || (budget.remaining_bytes == 0 && budget.has_emitted()) { + return Ok(match last_emitted { + Some(key) if emitted_this_call => CommitEnumeration::Truncated(key), + // A feed can carry an exhausted page-wide budget into the + // next commit. Its caller ends at the previous block + // boundary; no image-size value is needed in that case. + _ => CommitEnumeration::Exhausted { required_bytes: 0 }, + }); + } let op = emit.op(); let (id, before, after) = match emit { Emit::Insert(raw) => { - let image = emitted_image(right.dataset(), &raw, plan.kind).await?; + let image = emitted_image(source.child_dataset(), &raw, plan.kind).await?; (raw.id, None, Some(image)) } Emit::Delete(raw) => { - let image = emitted_image(left.dataset(), &raw, plan.kind).await?; + let image = emitted_image(source.parent_dataset(), &raw, plan.kind).await?; (raw.id, Some(image), None) } Emit::Update { before, after } => { - let before_image = emitted_image(left.dataset(), &before, plan.kind).await?; - let after_image = emitted_image(right.dataset(), &after, plan.kind).await?; + let before_image = + emitted_image(source.parent_dataset(), &before, plan.kind).await?; + let after_image = + emitted_image(source.child_dataset(), &after, plan.kind).await?; (after.id, Some(before_image), Some(after_image)) } }; @@ -500,11 +573,11 @@ pub(crate) async fn enumerate_commit_changes( // over budget), so a legal committed change — whose two images can // exceed the write-path-derived ceiling once managed Blobs inline as // base64 — is always deliverable, one per page if needed. The row - // budget still bounds packing. `Exhausted` now only signals a - // zero-capacity request (`max_changes == 0`), which validation - // already rejects; it is retained defensively. + // budget still bounds packing. `Exhausted` is retained for a feed + // carrying a page-wide budget already consumed by a prior block; + // a standalone request's validated budget starts nonzero. let over_bytes = encoded_bytes > budget.remaining_bytes; - if budget.remaining_rows == 0 || (over_bytes && budget.has_emitted()) { + if over_bytes && budget.has_emitted() { return Ok(match last_emitted { Some(key) if emitted_this_call => CommitEnumeration::Truncated(key), _ => CommitEnumeration::Exhausted { diff --git a/crates/omnigraph/src/changes/mod.rs b/crates/omnigraph/src/changes/mod.rs index adebd6a1..82978cac 100644 --- a/crates/omnigraph/src/changes/mod.rs +++ b/crates/omnigraph/src/changes/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod candidate_scan; pub(crate) mod enumerate; pub(crate) mod feed; pub mod model; diff --git a/crates/omnigraph/src/changes/row_compare.rs b/crates/omnigraph/src/changes/row_compare.rs index 53800507..3e3451ae 100644 --- a/crates/omnigraph/src/changes/row_compare.rs +++ b/crates/omnigraph/src/changes/row_compare.rs @@ -12,15 +12,16 @@ //! Lance virtual columns are skipped — a legal `_row_`-prefixed user property //! participates in change detection. -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet}; use std::pin::Pin; use arrow_array::{Array, RecordBatch, StringArray, StructArray, UInt64Array}; -use datafusion::prelude::{col, lit}; +use datafusion::prelude::{Expr, col, lit}; use futures::TryStreamExt; use lance::Dataset; use lance::dataset::scanner::{ColumnOrdering, DatasetRecordBatchStream}; use lance_core::datatypes::BlobHandling; +use lance_table::format::Fragment; use super::model::{COMMIT_CHANGES_MAX_BYTES, is_reserved_storage_system_column}; use crate::blob::{BlobDescriptor, BlobDescriptorDecoder}; @@ -76,6 +77,45 @@ struct BlobColumnSig { /// serialized size of each emitted change, never by trusting the scanner. const CHANGE_SCAN_TARGET_ROWS: usize = 8_192; +/// Approximate scanner batch targets. Candidate scans derive these from the +/// caller's remaining page budget (plus one continuation sentinel), so a +/// one-change page never asks Lance to prepare an 8,192-row candidate batch. +/// Lance treats both values as targets rather than hard limits; the hard page +/// bound remains the serialized-change accounting in `enumerate`. +#[derive(Debug, Clone, Copy)] +pub(crate) struct ScanTargets { + rows: usize, + bytes: u64, +} + +impl ScanTargets { + pub(crate) fn for_page(remaining_rows: usize, remaining_bytes: u64) -> Self { + Self { + rows: remaining_rows + .saturating_add(1) + .clamp(1, CHANGE_SCAN_TARGET_ROWS), + bytes: remaining_bytes.clamp(1, COMMIT_CHANGES_MAX_BYTES), + } + } + + pub(crate) fn rows(self) -> usize { + self.rows + } + + pub(crate) fn bytes(self) -> u64 { + self.bytes + } +} + +impl Default for ScanTargets { + fn default() -> Self { + Self { + rows: CHANGE_SCAN_TARGET_ROWS, + bytes: COMMIT_CHANGES_MAX_BYTES, + } + } +} + /// One scanned row prepared for comparison. Holds a zero-copy one-row slice /// (retaining `_rowid` and descriptor columns for lazy image and payload /// access) plus per-Blob-column physical descriptor identities. No JSON image @@ -93,35 +133,210 @@ pub(crate) struct RawRow { } impl RawRow { - /// Build the typed comparison unit for one row of a scanned batch — the - /// per-row form of [`prepare_batch`]. The batch must carry `id` (and, for a - /// Blob table, `_rowaddr`); Blob identities resolve from the in-memory - /// `dataset` manifest, no object-store I/O. Used by the branch-merge cursor - /// so merge and the change surfaces classify rows through ONE comparator. + /// Build the typed comparison unit for one row of a scanned batch. The + /// batch must carry `id` (and, for a Blob dataset, `_rowaddr`); Blob + /// identities resolve from the in-memory dataset manifest, with no + /// object-store I/O. Used by the branch-merge cursor so merge and the + /// change surfaces classify rows through one comparator. pub(crate) fn single( dataset: &Dataset, batch: &RecordBatch, row_index: usize, ) -> Result { - prepare_batch(dataset, &batch.slice(row_index, 1))? - .pop_front() - .ok_or_else(|| { - OmniError::manifest_internal("single-row batch produced no comparison row") - }) + let mut cursor = BatchCursor::try_new(dataset, batch.clone())?; + cursor.next_row = row_index; + cursor.next(dataset)?.ok_or_else(|| { + OmniError::manifest_internal("single-row batch produced no comparison row") + }) + } +} + +#[derive(Debug)] +struct BlobBatchColumn { + name: String, + column_index: usize, + field_id: i32, +} + +/// One scanner batch plus the small amount of schema metadata needed to +/// prepare rows lazily. The batch itself is retained once; only the current +/// row is sliced and decorated. This avoids the former `VecDeque` of +/// up to 8,192 slices, ids, and Blob-signature vectors. +#[derive(Debug)] +struct BatchCursor { + batch: RecordBatch, + next_row: usize, + id_index: usize, + row_address_index: Option, + blob_columns: Vec, +} + +impl BatchCursor { + fn try_new(dataset: &Dataset, batch: RecordBatch) -> Result { + let id_index = batch + .schema_ref() + .index_of("id") + .map_err(|_| OmniError::manifest_internal("change row is missing string id"))?; + batch + .column(id_index) + .as_any() + .downcast_ref::() + .ok_or_else(|| OmniError::manifest_internal("change row is missing string id"))?; + + let mut blob_columns = Vec::new(); + for (column_index, (field, column)) in batch + .schema_ref() + .fields() + .iter() + .zip(batch.columns()) + .enumerate() + { + if is_reserved_storage_system_column(field.name()) { + continue; + } + let lance_field = lance::datatypes::Field::try_from(field.as_ref()) + .map_err(OmniError::lance_internal)?; + if lance_field.is_blob() { + let descriptions = + column + .as_any() + .downcast_ref::() + .ok_or_else(|| { + OmniError::blob_integrity(format!( + "expected blob descriptions for change column '{}'", + field.name() + )) + })?; + // Validate the descriptor shape once when the batch arrives. + BlobDescriptorDecoder::try_new(descriptions)?; + let field_id = dataset.schema().field_id(field.name()).map_err(|error| { + OmniError::blob_integrity(format!( + "blob column '{}' has no field id: {error}", + field.name() + )) + })?; + blob_columns.push(BlobBatchColumn { + name: field.name().to_string(), + column_index, + field_id, + }); + } + } + blob_columns.sort_by(|left, right| left.name.cmp(&right.name)); + + let row_address_index = if blob_columns.is_empty() { + None + } else { + let index = batch.schema_ref().index_of("_rowaddr").map_err(|_| { + OmniError::manifest_internal( + "change scan is missing _rowaddr; managed Blob comparison needs the owning data file", + ) + })?; + batch + .column(index) + .as_any() + .downcast_ref::() + .ok_or_else(|| { + OmniError::manifest_internal( + "change scan is missing _rowaddr; managed Blob comparison needs the owning data file", + ) + })?; + Some(index) + }; + + Ok(Self { + batch, + next_row: 0, + id_index, + row_address_index, + blob_columns, + }) + } + + fn next(&mut self, dataset: &Dataset) -> Result> { + if self.next_row >= self.batch.num_rows() { + return Ok(None); + } + let row = self.next_row; + let ids = self + .batch + .column(self.id_index) + .as_any() + .downcast_ref::() + .expect("id shape was validated when the batch arrived"); + let row_addresses = self.row_address_index.map(|index| { + self.batch + .column(index) + .as_any() + .downcast_ref::() + .expect("row-address shape was validated when the batch arrived") + }); + + let mut blob_signatures = Vec::with_capacity(self.blob_columns.len()); + for blob_column in &self.blob_columns { + let descriptions = self + .batch + .column(blob_column.column_index) + .as_any() + .downcast_ref::() + .expect("Blob descriptor shape was validated when the batch arrived"); + let decoder = BlobDescriptorDecoder::try_new(descriptions)?; + let fragment_id = row_addresses + .expect("Blob batches carry row addresses") + .value(row) + >> 32; + let data_file_path = + data_file_path_for_field(dataset.fragments(), fragment_id, blob_column.field_id)?; + blob_signatures.push(BlobColumnSig { + name: blob_column.name.clone(), + identity: decoder.physical_identity(row, data_file_path)?, + managed: matches!(decoder.classify(row)?, BlobDescriptor::Managed { .. }), + }); + } + + self.next_row += 1; + Ok(Some(RawRow { + id: ids.value(row).to_string(), + slice: self.batch.slice(row, 1), + blob_signatures, + })) } } -/// An `id`-ordered stream of one table snapshot's rows, filled lazily one Lance +/// An `id`-ordered stream of one dataset snapshot's rows, filled lazily one Lance /// batch at a time. Both endpoints of a diff are walked in lockstep so the /// merge stays streaming and never buffers a delta-wide row set. pub(crate) struct OrderedRows { dataset: Dataset, stream: Option>>, - pending: VecDeque, + batch: Option, + pending: Option, } impl OrderedRows { pub(crate) async fn open(dataset: Dataset, after_id: Option<&str>) -> Result { + Self::open_scan(dataset, after_id, None, None, ScanTargets::default()).await + } + + /// The full scan surface. `fragments` scopes the scan to exactly those + /// physical fragments — the candidate path passes only the transaction's + /// new or touched fragments, while the version-window `extra_filter` drops + /// carried-over rows a fragment rewrite pulled along. + pub(crate) async fn open_scan( + dataset: Dataset, + after_id: Option<&str>, + extra_filter: Option, + fragments: Option>, + targets: ScanTargets, + ) -> Result { + if fragments.as_ref().is_some_and(Vec::is_empty) { + return Ok(Self { + dataset, + stream: None, + batch: None, + pending: None, + }); + } let after_id = after_id.map(str::to_string); let stream = Box::pin( TableStore::scan_stream_with( @@ -131,18 +346,27 @@ impl OrderedRows { Some(vec![ColumnOrdering::asc_nulls_last("id".to_string())]), true, move |scanner| { - if let Some(after_id) = after_id { - scanner.filter_expr(col("id").gt(lit(after_id))); + if let Some(fragments) = fragments { + scanner.with_fragments(fragments); + } + let resume = after_id.map(|after_id| col("id").gt(lit(after_id))); + if let Some(filter) = match (resume, extra_filter) { + (Some(resume), Some(extra)) => Some(resume.and(extra)), + (Some(resume), None) => Some(resume), + (None, Some(extra)) => Some(extra), + (None, None) => None, + } { + scanner.filter_expr(filter); } // Descriptor-sized batches bounded by rows AND bytes, the // same shape as every other production ordered-by-id scan. // strict_batch_size is deliberately absent: Lance's strict // stream coalesces to a row count the environment can // override and ignores the byte target while accumulating. - scanner.batch_size(CHANGE_SCAN_TARGET_ROWS); - scanner.batch_size_bytes(COMMIT_CHANGES_MAX_BYTES); + scanner.batch_size(targets.rows); + scanner.batch_size_bytes(targets.bytes); scanner.blob_handling(BlobHandling::BlobsDescriptions); - // Managed Blob descriptors are file-relative, so `prepare_batch` + // Managed Blob descriptors are file-relative, so `BatchCursor` // maps the row's fragment (high 32 bits of `_rowaddr`) to the // owning data file's immutable UUID path — the qualifier that // tells an unchanged row from a same-length Blob-only update @@ -158,27 +382,35 @@ impl OrderedRows { Ok(Self { dataset, stream: Some(stream), - pending: VecDeque::new(), + batch: None, + pending: None, }) } pub(crate) async fn peek(&mut self) -> Result> { self.fill().await?; - Ok(self.pending.front()) + Ok(self.pending.as_ref()) } pub(crate) async fn pop(&mut self) -> Result> { self.fill().await?; - Ok(self.pending.pop_front()) + Ok(self.pending.take()) } async fn fill(&mut self) -> Result<()> { - while self.pending.is_empty() { + while self.pending.is_none() { + if let Some(batch) = self.batch.as_mut() { + if let Some(row) = batch.next(&self.dataset)? { + self.pending = Some(row); + return Ok(()); + } + self.batch = None; + } let Some(stream) = self.stream.as_mut() else { return Ok(()); }; match stream.try_next().await { - Ok(Some(batch)) => self.pending = prepare_batch(&self.dataset, &batch)?, + Ok(Some(batch)) => self.batch = Some(BatchCursor::try_new(&self.dataset, batch)?), Ok(None) => { self.stream = None; return Ok(()); @@ -194,119 +426,23 @@ impl OrderedRows { } } -/// Turn one scanned batch into comparison-ready rows: one-row slices plus -/// physical descriptor identities for Blob columns. Pure in-memory work — Blob -/// payloads are never touched here, and the data-file resolution below reads -/// only the already-loaded `dataset` manifest (no object-store call). -fn prepare_batch(dataset: &Dataset, batch: &RecordBatch) -> Result> { - let ids = batch - .column_by_name("id") - .and_then(|column| column.as_any().downcast_ref::()) - .ok_or_else(|| OmniError::manifest_internal("change row is missing string id"))?; - - let mut blob_columns = Vec::new(); - for (field, column) in batch.schema_ref().fields().iter().zip(batch.columns()) { - if is_reserved_storage_system_column(field.name()) { - continue; - } - let lance_field = - lance::datatypes::Field::try_from(field.as_ref()).map_err(OmniError::lance_internal)?; - if lance_field.is_blob() { - let descriptions = column - .as_any() - .downcast_ref::() - .ok_or_else(|| { - OmniError::blob_integrity(format!( - "expected blob descriptions for change column '{}'", - field.name() - )) - })?; - // The Lance field id keys the row's owning data file (a Blob column - // has one field id spanning several physical columns). - let field_id = dataset.schema().field_id(field.name()).map_err(|error| { - OmniError::blob_integrity(format!( - "blob column '{}' has no field id: {error}", - field.name() - )) - })?; - blob_columns.push(( - field.name().to_string(), - BlobDescriptorDecoder::try_new(descriptions)?, - field_id, - )); - } - } - // Name-order the signatures so `rows_equal`'s positional zip aligns by - // column NAME whenever the two sides' column sets match — the same - // order-insensitivity the schema gate's name-keyed fingerprint provides. - blob_columns.sort_by(|left, right| left.0.cmp(&right.0)); - - // Managed Blob descriptors resolve relative to the owning DATA FILE, so each - // managed identity is qualified by that file's immutable path (a per-file - // UUID). Map the row's fragment id (high 32 bits of `_rowaddr`) to the - // fragment, then the fragment's data file that holds the blob column's field - // id. All in-memory manifest reads. Only needed when the table has a Blob. - let fragments_by_id: HashMap = if blob_columns.is_empty() { - HashMap::new() - } else { - dataset - .fragments() - .iter() - .map(|fragment| (fragment.id, fragment)) - .collect() - }; - let row_addresses = if blob_columns.is_empty() { - None - } else { - Some( - batch - .column_by_name("_rowaddr") - .and_then(|column| column.as_any().downcast_ref::()) - .ok_or_else(|| { - OmniError::manifest_internal( - "change scan is missing _rowaddr; managed Blob comparison needs the owning data file" - ) - })?, - ) - }; - - let mut rows = VecDeque::with_capacity(batch.num_rows()); - for row in 0..batch.num_rows() { - let mut blob_signatures = Vec::with_capacity(blob_columns.len()); - if let Some(row_addresses) = row_addresses { - let fragment_id = row_addresses.value(row) >> 32; - for (name, decoder, field_id) in &blob_columns { - let data_file_path = - data_file_path_for_field(&fragments_by_id, fragment_id, *field_id)?; - blob_signatures.push(BlobColumnSig { - name: name.clone(), - identity: decoder.physical_identity(row, data_file_path)?, - managed: matches!(decoder.classify(row)?, BlobDescriptor::Managed { .. }), - }); - } - } - rows.push_back(RawRow { - id: ids.value(row).to_string(), - slice: batch.slice(row, 1), - blob_signatures, - }); - } - Ok(rows) -} - /// Resolve the immutable data-file path that holds `field_id` in the fragment /// owning a row — the stable UUID qualifier for a managed-Blob identity. Reads /// only the in-memory manifest. -fn data_file_path_for_field<'a>( - fragments_by_id: &'a HashMap, +fn data_file_path_for_field( + fragments: &[lance_table::format::Fragment], fragment_id: u64, field_id: i32, -) -> Result<&'a str> { - let fragment = fragments_by_id.get(&fragment_id).ok_or_else(|| { - OmniError::blob_integrity(format!( - "change scan referenced fragment {fragment_id} absent from the manifest" - )) - })?; +) -> Result<&str> { + let fragment = fragments + .binary_search_by_key(&fragment_id, |fragment| fragment.id) + .ok() + .map(|index| &fragments[index]) + .ok_or_else(|| { + OmniError::blob_integrity(format!( + "change scan referenced fragment {fragment_id} absent from the manifest" + )) + })?; fragment .files .iter() diff --git a/crates/omnigraph/src/failpoints.rs b/crates/omnigraph/src/failpoints.rs index e7078f6f..8cdcc552 100644 --- a/crates/omnigraph/src/failpoints.rs +++ b/crates/omnigraph/src/failpoints.rs @@ -121,6 +121,14 @@ pub mod names { /// the LOGICAL post-open head re-prove still refuses the replacement — /// the e_tag is defense-in-depth, not the load-bearing witness. pub const CHANGE_FEED_SKIP_ETAG_WITNESS: &str = "change_feed.skip_etag_witness"; + /// A change-feed poll has passed the final post-open logical head witness + /// for one commit, and is about to plan each interval's emitter. Tests + /// delete and recreate a named branch here: any live read of the branch's + /// numeric-path history after this point (the replaceable read — version + /// manifests sit at numeric paths, unlike UUID-named data and transaction + /// files) would classify the interval from the REPLACEMENT branch's + /// transactions and can silently omit the original commit's deletes. + pub const CHANGE_FEED_POST_HEAD_WITNESS: &str = "change_feed.post_head_witness"; pub const CLEANUP_RECONCILE_FORK: &str = "cleanup.reconcile_fork"; /// After cleanup's fast empty-sidecar probe, before it acquires the closed /// schema/branch/table GC gate set and performs the authoritative recheck. diff --git a/crates/omnigraph/src/instrumentation.rs b/crates/omnigraph/src/instrumentation.rs index 06620c2b..81afe0ac 100644 --- a/crates/omnigraph/src/instrumentation.rs +++ b/crates/omnigraph/src/instrumentation.rs @@ -90,6 +90,27 @@ pub struct QueryIoProbes { /// invisible to the manifest/data IO counters — a cost test asserts it so a /// future forward-child projection (the bounded-visit fix) is measurable. pub feed_commits_visited: Arc, + /// Adjacent-version transaction files read while classifying CDC candidate + /// intervals. Wider intervals must fall back before incrementing this + /// counter, keeping stateless tiny-page resumes constant in history depth. + pub candidate_transaction_reads: Arc, + /// Manifest fragment entries compared or validated while deriving a CDC + /// candidate plan. This exposes the metadata CPU term that object-store I/O + /// counters cannot see. + pub candidate_fragment_metadata_steps: Arc, + /// Candidate child rows pulled by the pruned emitter. A max-changes=1 page + /// over all-changing rows should inspect only the emitted row plus one + /// continuation sentinel. + pub candidate_rows_examined: Arc, + /// Largest row/byte scanner target requested by a candidate emitter in the + /// measured operation. Both are maxima (not sums) because parent and child + /// streams use the same current-page target. + pub candidate_scan_target_rows_peak: Arc, + pub candidate_scan_target_bytes_peak: Arc, + /// Complete logical change images materialized (and therefore eligible to + /// read managed Blob payloads). A continuation sentinel must not increment + /// this counter. + pub change_images_materialized: Arc, } tokio::task_local! { @@ -217,6 +238,39 @@ pub(crate) fn record_feed_commits_visited(commits: usize) { }); } +pub(crate) fn record_candidate_transaction_read() { + let _ = current(|p| { + p.candidate_transaction_reads + .fetch_add(1, Ordering::Relaxed) + }); +} + +pub(crate) fn record_candidate_fragment_metadata_steps(steps: u64) { + if steps > 0 { + let _ = current(|p| { + p.candidate_fragment_metadata_steps + .fetch_add(steps, Ordering::Relaxed) + }); + } +} + +pub(crate) fn record_candidate_row_examined() { + let _ = current(|p| p.candidate_rows_examined.fetch_add(1, Ordering::Relaxed)); +} + +pub(crate) fn record_candidate_scan_targets(rows: usize, bytes: u64) { + let _ = current(|p| { + p.candidate_scan_target_rows_peak + .fetch_max(rows as u64, Ordering::Relaxed); + p.candidate_scan_target_bytes_peak + .fetch_max(bytes, Ordering::Relaxed); + }); +} + +pub(crate) fn record_change_image_materialized() { + let _ = current(|p| p.change_images_materialized.fetch_add(1, Ordering::Relaxed)); +} + /// Per-operation staged-write counts, installed for a task via /// [`with_merge_write_probes`]. Lets a cost-budget test assert WHICH staged-write /// primitive an operation invokes — e.g. that an append-only fast-forward merge diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs index ca8381ad..1f0e74c0 100644 --- a/crates/omnigraph/src/table_store.rs +++ b/crates/omnigraph/src/table_store.rs @@ -117,6 +117,14 @@ impl ScanTuning<'_> { self } + /// Scope the scan to exactly these physical fragments — a scan-input + /// selection like `filter_expr`, not an ordering decision, so the bounded + /// executor routing chosen by `scan_stream_with` is unaffected. + pub(crate) fn with_fragments(&mut self, fragments: Vec) -> &mut Self { + self.scanner.with_fragments(fragments); + self + } + pub(crate) fn batch_size(&mut self, batch_size: usize) -> &mut Self { self.scanner.batch_size(batch_size); self @@ -238,6 +246,46 @@ pub(crate) fn has_insert_absence_certificate(transaction: &Transaction) -> bool .is_some_and(|value| value == INSERT_ABSENCE_V1) } +/// Durable provenance marker asserting that a keyed-write `Operation::Update` +/// was produced by an OmniGraph keyed writer, whose `merge_insert` never uses a +/// delete-capable by-source arm (Lance defaults the unmatched-by-source arm to +/// Keep, and the `no_delete_capable_merge_arm_in_engine_source` guard forbids +/// such an arm in engine source). It therefore removed no unmatched-by-source +/// row, so a +/// consumer may treat the interval as row-set-preserving. Unlike +/// `insert_absence`, it is stamped on every **general keyed MergeInsert +/// update** — real upserts, known-present updates, and stream strict inserts — +/// not only pure inserts; batch/proven strict inserts carry `insert_absence` +/// instead, which consumers accept as the same no-delete proof. It is +/// read-advisory: a missing marker only forces a fall-back, never a +/// correctness change. A persisted `Update` from an external Lance merge +/// (e.g. adopted via `repair --force`) carries no marker and cannot be pruned. +pub(crate) const NO_BY_SOURCE_DELETE_PROPERTY: &str = "omnigraph.no_by_source_delete"; +pub(crate) const NO_BY_SOURCE_DELETE_V1: &str = "v1"; + +pub(crate) fn has_no_by_source_delete_marker(transaction: &Transaction) -> bool { + transaction + .transaction_properties + .as_ref() + .and_then(|properties| properties.get(NO_BY_SOURCE_DELETE_PROPERTY)) + .is_some_and(|value| value == NO_BY_SOURCE_DELETE_V1) +} + +/// Stamp [`NO_BY_SOURCE_DELETE_PROPERTY`] on a keyed-write transaction before it +/// is committed. Unconditional: every OmniGraph keyed `merge_insert` is +/// no-by-source-delete by construction. Mirrors `certify_insert_absence`'s +/// property write; the two keys are distinct, so a pure-insert upsert that also +/// earns `insert_absence` carries both. +fn stamp_no_by_source_delete(transaction: &mut Transaction) { + let properties = transaction + .transaction_properties + .get_or_insert_with(|| Arc::new(HashMap::new())); + Arc::make_mut(properties).insert( + NO_BY_SOURCE_DELETE_PROPERTY.to_string(), + NO_BY_SOURCE_DELETE_V1.to_string(), + ); +} + /// Verify one persisted link of the insertion-absence proof chain and return /// its exact physical row contribution. This is intentionally stricter than a /// property lookup: the caller must also supply the expected parent version, @@ -5588,7 +5636,7 @@ fn validate_proven_insert_source_batch(batch: &RecordBatch, table_key: &str) -> /// Preserve all conflict metadata Lance returned while exposing the physical /// fragment delta needed by read-your-writes scans. fn staged_keyed_merge_result( - uncommitted: UncommittedMergeInsert, + mut uncommitted: UncommittedMergeInsert, context: &'static str, ) -> Result { let (new_fragments, removed_fragment_ids) = match &uncommitted.transaction.operation { @@ -5609,6 +5657,13 @@ fn staged_keyed_merge_result( ))); } }; + // This is the single chokepoint for every general OmniGraph keyed + // `merge_insert` Update (upsert, known-present update, stream strict + // insert), none of which use a by-source-delete arm. Stamp the durable + // no-by-source-delete marker before the transaction is committed, so the CDC + // candidate-pruning classifier can trust this persisted `Update` removed no + // rows (an external merge adopted via `repair --force` carries no marker). + stamp_no_by_source_delete(&mut uncommitted.transaction); Ok(StagedWrite::with_commit_metadata( uncommitted.transaction, StagedCommitMetadata::affected_rows(uncommitted.affected_rows), diff --git a/crates/omnigraph/src/table_store/staged_tests.rs b/crates/omnigraph/src/table_store/staged_tests.rs index 25523356..81fe941a 100644 --- a/crates/omnigraph/src/table_store/staged_tests.rs +++ b/crates/omnigraph/src/table_store/staged_tests.rs @@ -647,6 +647,69 @@ async fn all_new_upsert_certifies_insert_absence_and_persists_it_in_history() { assert!(super::has_insert_absence_certificate(&history[0])); } +/// Every general keyed-write `Update` (here an upsert that UPDATES an existing +/// row, so it is not a pure insert and earns no `insert_absence`) carries the +/// durable no-by-source-delete marker, and it survives commit → reopen → +/// `list_transactions` (the exact read path the CDC candidate-pruning classifier +/// uses). This is the durable provenance that lets pruning trust a persisted +/// `Update` removed no rows; an external merge would carry no marker. +#[tokio::test] +async fn keyed_upsert_stamps_no_by_source_delete_marker_and_persists_it() { + let dir = tempfile::tempdir().unwrap(); + let uri = format!("{}/people.lance", dir.path().to_str().unwrap()); + let store = TableStore::new(dir.path().to_str().unwrap(), test_session()); + let ds = TableStore::write_dataset(&uri, person_pk_batch(&[("alice", Some(30))])) + .await + .unwrap(); + let base_version = ds.version().version; + + let staged = store + .stage_keyed_write( + ds.clone(), + "Person", + person_pk_batch(&[("alice", Some(31))]), + KeyedWriteSemantics::Upsert, + ) + .await + .unwrap(); + assert!( + super::has_no_by_source_delete_marker(&staged.transaction), + "every keyed-write Update carries the no-by-source-delete marker" + ); + assert!( + !super::has_insert_absence_certificate(&staged.transaction), + "an upsert that updates an existing row is not a pure insert, so no insert_absence" + ); + + let committed = store.commit_staged(Arc::new(ds), staged).await.unwrap(); + let committed_version = committed.version().version; + let reopened = Dataset::open(&uri).await.unwrap(); + let persisted = reopened + .read_transaction_by_version(committed_version) + .await + .unwrap() + .expect("committed keyed upsert transaction"); + assert!( + super::has_no_by_source_delete_marker(&persisted), + "the marker must survive commit and reopen" + ); + + let history = reopened + .delta() + .with_begin_version(base_version) + .with_end_version(committed_version) + .build() + .unwrap() + .list_transactions() + .await + .unwrap(); + assert_eq!(history.len(), 1); + assert!( + super::has_no_by_source_delete_marker(&history[0]), + "list_transactions (the candidate-pruning read path) sees the marker" + ); +} + #[tokio::test] async fn keyed_strict_insert_preflights_typed_conflict_without_changing_mode() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/omnigraph/tests/changes.rs b/crates/omnigraph/tests/changes.rs index ea4ba549..390a5b87 100644 --- a/crates/omnigraph/tests/changes.rs +++ b/crates/omnigraph/tests/changes.rs @@ -1344,6 +1344,68 @@ async fn change_feed_detects_same_length_blob_update_after_overwrite() { ); } +/// RFC-030 §4.2/§9 L0: candidate pruning must NOT be applied to an operation +/// that can remove or reuse a logical id. An overwrite that drops one id and +/// changes another is `Operation::Overwrite`, which the classifier rejects, so +/// the enumerator falls back to the exact ordered merge and reports BOTH the +/// delete and the update. A candidate scan of the child's new fragments alone +/// would never see the dropped id, silently losing the delete. +#[tokio::test] +async fn commit_changes_falls_back_for_overwrite_that_removes_an_id() { + use omnigraph::changes::{ChangeFeedScope, ChangeOpKind}; + + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = Omnigraph::init( + uri, + "node Doc {\n slug: String @key\n body: String?\n}", + ) + .await + .unwrap(); + db.load_with_receipt( + "main", + "{\"type\":\"Doc\",\"data\":{\"slug\":\"x\",\"body\":\"a\"}}\n{\"type\":\"Doc\",\"data\":{\"slug\":\"y\",\"body\":\"b\"}}", + LoadMode::Overwrite, + ) + .await + .unwrap(); + // Overwrite the whole table: y is dropped and x is changed. + let overwritten = db + .load_with_receipt( + "main", + "{\"type\":\"Doc\",\"data\":{\"slug\":\"x\",\"body\":\"c\"}}", + LoadMode::Overwrite, + ) + .await + .unwrap(); + + let page = db + .commit_changes_page( + &overwritten.commit.graph_commit_id, + &ChangeFeedScope::default(), + None, + None, + None, + ) + .await + .unwrap(); + let mut ops: Vec<(String, ChangeOpKind)> = page + .block + .changes + .iter() + .map(|change| (change.id.clone(), change.op)) + .collect(); + ops.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!( + ops, + vec![ + ("x".to_string(), ChangeOpKind::Update), + ("y".to_string(), ChangeOpKind::Delete), + ], + "overwrite must fall back to the exact merge and report the dropped id as a delete" + ); +} + #[tokio::test] async fn commit_changes_are_exact_ordered_and_bounded() { use omnigraph::changes::{ChangeEntityKind, ChangeFeedScope, ChangeOpKind}; diff --git a/crates/omnigraph/tests/changes_cost.rs b/crates/omnigraph/tests/changes_cost.rs index 75212dec..131f57c7 100644 --- a/crates/omnigraph/tests/changes_cost.rs +++ b/crates/omnigraph/tests/changes_cost.rs @@ -1,45 +1,63 @@ //! Cost-budget tests for per-commit change pages, on the shared //! `helpers::cost` harness. //! -//! The current page implementation is the exact ordered-merge fallback — the -//! authority path: for every changed table lifetime it scans BOTH pinned -//! versions, so page cost is O(table size), not O(delta). Following the -//! `merge_cost.rs` idiom, that known-non-flat term is pinned as a GROWING -//! tripwire rather than mislabeled flat; substrate candidate pruning over the -//! Lance row-version columns is the planned fix and must flip the tripwire to -//! a flat assertion when it lands. The bounded terms asserted here: +//! A per-commit page has two derivation paths. When the commit's effect on a +//! table is a proven row-set-preserving shape (RFC-030 §4.2), it is derived in +//! from the adjacent transaction's touched fragments: new child fragments and +//! only the parent fragments that transaction updated or removed are streamed +//! in id order. No secondary index is required, so absent and stale-index states +//! have the same bounded shape. When the effect is unproven (delete, overwrite, +//! …), it falls back to the exact ordered merge of both pinned versions — +//! O(dataset extent), pinned honestly as a GROWING tripwire. The terms +//! asserted here: //! //! * dataset opens per page — at most parent + child of each changed //! interval; an untouched table contributes zero opens; +//! * candidate transaction reads — exactly one for an adjacent interval and +//! zero for a wider interval (fallback happens before history I/O); +//! * pruned-path data reads — flat in dataset extent without index coverage; +//! * fragment-metadata steps — logarithmic lookup plus touched fragments, not +//! a walk of the complete parent/child manifests; +//! * fallback-path data reads — growing in dataset extent (exact merge); +//! * max-changes=1 candidate work — one emitted row plus one sentinel, with +//! scanner targets derived from the current row/byte page budget; //! * Blob payload work — proportional to emitted changes, never to the //! number of unchanged Blob rows scanned (descriptor identity short-circuit). #![recursion_limit = "512"] mod helpers; +use std::sync::Arc; + use helpers::cost::{IoCounts, assert_flat, assert_grows, cost_harness, measure}; +use lance::Dataset; +use lance::dataset::UpdateBuilder; +use omnigraph::IndexCoverage; use omnigraph::changes::ChangeFeedScope; -use omnigraph::db::Omnigraph; +use omnigraph::db::{Omnigraph, ReadTarget, RepairOptions}; use omnigraph::loader::LoadMode; +use omnigraph_compiler::ir::ParamMap; -/// One page over a Δ=1 commit: opens stay bounded by the changed interval -/// while the ordered-merge scan term grows with the changed table's physical -/// extent (its fragments), because the exact fallback reads both pinned -/// versions in full. Both sweep points publish the SAME number of graph -/// commits — the smaller point pads history with commits on the untouched -/// table — so the known `__manifest` fold term stays comparable and only the -/// scanned table's extent moves. +/// One page over a Δ=1 update commit: opens stay bounded by the changed +/// interval AND the data-read term stays flat in the changed table's physical +/// extent, because the proven interval is derived from only its new child +/// fragments and transaction-touched parent fragments. Run the same curve with +/// the `id` index absent and stale: neither state may affect the plan. Both +/// extent sweep points publish the same number of graph commits — the smaller +/// point pads history with commits on the untouched dataset — so the known +/// `__manifest` fold term stays comparable. #[tokio::test] -async fn changes_page_opens_are_bounded_and_scan_term_grows_with_table_extent() { +async fn changes_page_opens_and_data_reads_are_bounded_by_delta() { const SEED_COMMITS: u64 = 8; const ROWS_PER_COMMIT: u64 = 64; cost_harness(async { - let mut curve: Vec<(u64, IoCounts)> = Vec::new(); - for person_commits in [2u64, 8] { - let dir = tempfile::tempdir().unwrap(); - let db = Omnigraph::init( - dir.path().to_str().unwrap(), - r#" + for stale_index in [false, true] { + let mut curve: Vec<(u64, IoCounts)> = Vec::new(); + for person_commits in [2u64, 8] { + let dir = tempfile::tempdir().unwrap(); + let db = Omnigraph::init( + dir.path().to_str().unwrap(), + r#" node Person { name: String @key age: I32? @@ -48,6 +66,341 @@ node Company { slug: String @key } "#, + ) + .await + .unwrap(); + for commit in 0..SEED_COMMITS { + let batch = if commit < person_commits { + (0..ROWS_PER_COMMIT) + .map(|row| { + let name = commit * ROWS_PER_COMMIT + row; + format!( + r#"{{"type":"Person","data":{{"name":"p{name:05}","age":1}}}}"# + ) + }) + .collect::>() + .join("\n") + } else { + format!(r#"{{"type":"Company","data":{{"slug":"filler-{commit}"}}}}"#) + }; + db.load_with_receipt("main", &batch, LoadMode::Merge) + .await + .unwrap(); + } + if stale_index { + // Build full coverage, then append a new parent fragment. + // The measured update therefore runs with a normal partial + // / stale index state. The false arm never builds an index. + db.ensure_indices().await.unwrap(); + db.load_with_receipt( + "main", + r#"{"type":"Person","data":{"name":"stale-index-tail","age":1}}"#, + LoadMode::Merge, + ) + .await + .unwrap(); + } + let parent = db + .snapshot_of(ReadTarget::branch("main")) + .await + .unwrap() + .open_dataset("node:Person") + .await + .unwrap(); + assert_eq!( + parent.has_btree_index("id").await.unwrap(), + stale_index, + "the two cost cells must distinguish absent from physically present index" + ); + assert!( + matches!( + parent.index_coverage("id").await.unwrap(), + IndexCoverage::Degraded { .. } + ), + "both absent and stale/partial coverage must be normal degraded states" + ); + let updated = db + .load_with_receipt( + "main", + r#"{"type":"Person","data":{"name":"p00000","age":2}}"#, + LoadMode::Merge, + ) + .await + .unwrap(); + + let (page, io) = measure(db.commit_changes_page( + &updated.commit.graph_commit_id, + &ChangeFeedScope::default(), + None, + Some(10), + None, + )) + .await; + let page = page.unwrap(); + assert_eq!( + page.block.changes.len(), + 1, + "the measured commit is a one-row update" + ); + assert_eq!( + io.candidate_transaction_reads, 1, + "an adjacent candidate interval reads exactly its child's transaction" + ); + assert!( + io.candidate_fragment_metadata_steps <= 32, + "candidate planning must binary-search manifests rather than walk them: {io:?}" + ); + eprintln!( + "PAGE stale_index={stale_index} person_commits={person_commits}: \ + data_open_count={} data_reads={} candidate_fragment_steps={}", + io.data_open_count, io.data_reads, io.candidate_fragment_metadata_steps, + ); + curve.push((person_commits, io)); + } + for (person_commits, io) in &curve { + assert!( + io.data_open_count <= 2, + "one changed interval opens at most its parent and child pinned \ + datasets; untouched datasets contribute zero \ + (stale_index={stale_index}, person_commits={person_commits}): {io:?}" + ); + } + assert_flat(&curve, |io| io.data_open_count, 0, "changed-interval opens"); + assert_flat(&curve, |io| io.manifest_reads, 0, "manifest reads per page"); + assert_flat( + &curve, + |io| io.data_reads, + 3, + "candidate-pruned page data reads (touched fragments, not dataset extent)", + ); + // Two binary searches may grow logarithmically as fragments grow; + // a full-manifest walk would exceed this tight delta immediately. + assert_flat( + &curve, + |io| io.candidate_fragment_metadata_steps, + 8, + "candidate fragment metadata work (logarithmic + touched fragments)", + ); + } + }) + .await; +} + +/// A large all-changing delta with `max_changes=1` must not prepare the former +/// 8,192-row parent-probe chunk. The emitter reads one row for the response and +/// one look-ahead row to prove truncation; its Lance row/byte batch targets are +/// derived from this page's remaining budget. Blob descriptions make the byte +/// target meaningful without paying payload I/O for un-emitted rows. +#[tokio::test] +async fn changes_page_size_one_bounds_large_candidate_delta() { + const DELTA_ROWS: usize = 2_048; + const PAGE_BYTES: u64 = 4 * 1_024; + + cost_harness(async { + let dir = tempfile::tempdir().unwrap(); + let db = Omnigraph::init( + dir.path().to_str().unwrap(), + r#" +node Document { + slug: String @key + payload: Blob? +} +"#, + ) + .await + .unwrap(); + let batch = (0..DELTA_ROWS) + .map(|row| { + format!( + r#"{{"type":"Document","data":{{"slug":"d{row:05}","payload":"base64:QQ=="}}}}"# + ) + }) + .collect::>() + .join("\n"); + let inserted = db + .load_with_receipt("main", &batch, LoadMode::Merge) + .await + .unwrap(); + + let (page, io) = measure(db.commit_changes_page( + &inserted.commit.graph_commit_id, + &ChangeFeedScope::default(), + None, + Some(1), + Some(PAGE_BYTES), + )) + .await; + let first = page.unwrap(); + assert_eq!(first.block.changes.len(), 1); + let token = first + .next_page_token + .expect("the large delta must continue"); + let (second, second_io) = measure(db.commit_changes_page( + &inserted.commit.graph_commit_id, + &ChangeFeedScope::default(), + Some(&token), + Some(1), + Some(PAGE_BYTES), + )) + .await; + let second = second.unwrap(); + assert_eq!(second.block.changes.len(), 1); + assert_ne!(first.block.changes[0].id, second.block.changes[0].id); + + for io in [io, second_io] { + assert_eq!(io.candidate_transaction_reads, 1); + assert_eq!( + io.candidate_rows_examined, 2, + "one emitted candidate plus one continuation sentinel; no 8,192-row queue" + ); + assert_eq!( + io.candidate_scan_target_rows_peak, 2, + "candidate scanner row target follows max_changes + one sentinel" + ); + assert_eq!( + io.candidate_scan_target_bytes_peak, PAGE_BYTES, + "candidate scanner byte target follows the current page budget" + ); + assert_eq!( + io.change_images_materialized, 1, + "the continuation sentinel must not materialize JSON or Blob payloads" + ); + } + }) + .await; +} + +/// Multi-version graph intervals are intentionally outside the candidate +/// contract. The fixture advances one physical dataset twice, then adopts both +/// logical updates in one forced repair graph commit. Two stateless +/// `max_changes=1` pages must each fall back before reading either transaction; +/// the page count cannot multiply a transaction-history walk. +#[tokio::test] +async fn changes_page_size_one_skips_transaction_history_for_multi_version_intervals() { + cost_harness(async { + let dir = tempfile::tempdir().unwrap(); + let db = Omnigraph::init( + dir.path().to_str().unwrap(), + "node Person {\n name: String @key\n age: I32?\n}", + ) + .await + .unwrap(); + db.load_with_receipt( + "main", + concat!( + r#"{"type":"Person","data":{"name":"alice","age":1}}"#, + "\n", + r#"{"type":"Person","data":{"name":"bob","age":1}}"#, + ), + LoadMode::Merge, + ) + .await + .unwrap(); + let snapshot = db.snapshot_of(ReadTarget::branch("main")).await.unwrap(); + let entry = snapshot.dataset("node:Person").unwrap(); + let before = entry.published_dataset_version; + let dataset_uri = format!( + "{}/{}", + db.uri().trim_end_matches('/'), + entry.dataset_path.trim_start_matches('/') + ); + + let dataset = Dataset::open(&dataset_uri).await.unwrap(); + let dataset = UpdateBuilder::new(Arc::new(dataset)) + .update_where("name = 'alice'") + .unwrap() + .set("age", "2") + .unwrap() + .build() + .unwrap() + .execute() + .await + .unwrap() + .new_dataset; + let dataset = UpdateBuilder::new(dataset) + .update_where("name = 'bob'") + .unwrap() + .set("age", "3") + .unwrap() + .build() + .unwrap() + .execute() + .await + .unwrap() + .new_dataset; + assert_eq!(dataset.version().version, before + 2); + + let repair = db + .repair(RepairOptions { + confirm: true, + force: true, + }) + .await + .unwrap(); + assert!(repair.graph_manifest_version.is_some()); + let repair_commit = db + .list_commits(None) + .await + .unwrap() + .first() + .expect("repair publishes a graph lineage commit") + .graph_commit_id + .clone(); + + let (first, first_io) = measure(db.commit_changes_page( + &repair_commit, + &ChangeFeedScope::default(), + None, + Some(1), + None, + )) + .await; + let first = first.unwrap(); + assert_eq!(first.block.changes.len(), 1); + let token = first.next_page_token.expect("two updates need two pages"); + assert_eq!( + first_io.candidate_transaction_reads, 0, + "first multi-version page must fall back before transaction-history I/O" + ); + + let (second, second_io) = measure(db.commit_changes_page( + &repair_commit, + &ChangeFeedScope::default(), + Some(&token), + Some(1), + None, + )) + .await; + let second = second.unwrap(); + assert_eq!(second.block.changes.len(), 1); + assert!(second.next_page_token.is_none()); + assert_ne!(first.block.changes[0].id, second.block.changes[0].id); + for io in [first_io, second_io] { + assert_eq!( + io.candidate_transaction_reads, 0, + "every stateless multi-version page must skip transaction-history I/O" + ); + } + }) + .await; +} + +/// The fallback path stays honestly pinned: an unproven operation (a delete) +/// forces the exact ordered merge of both pinned versions, so page data reads +/// grow with the table's physical extent even at Δ=1. This is the counterpart +/// to the pruned flat assertion above — if a future change mistakenly pruned an +/// unproven op, this tripwire would go flat and fail. +#[tokio::test] +async fn changes_page_unproven_op_scan_term_grows_with_table_extent() { + const SEED_COMMITS: u64 = 8; + const ROWS_PER_COMMIT: u64 = 64; + cost_harness(async { + let mut curve: Vec<(u64, IoCounts)> = Vec::new(); + for person_commits in [2u64, 8] { + let dir = tempfile::tempdir().unwrap(); + let db = Omnigraph::init( + dir.path().to_str().unwrap(), + "node Person {\n name: String @key\n age: I32?\n}\nnode Company {\n slug: String @key\n}\n", ) .await .unwrap(); @@ -67,17 +420,24 @@ node Company { .await .unwrap(); } - let updated = db - .load_with_receipt( + // A delete is an unproven (row-removing) operation, so the enumerator + // falls back to the exact ordered merge of both pinned versions. + let deleted = db + .mutate_with_receipt( "main", - r#"{"type":"Person","data":{"name":"p00000","age":2}}"#, - LoadMode::Merge, + "query del() { delete Person where name = \"p00000\" }", + "del", + &ParamMap::new(), ) .await .unwrap(); + let commit_id = deleted + .commit + .expect("a row-removing delete publishes one commit") + .graph_commit_id; let (page, io) = measure(db.commit_changes_page( - &updated.commit.graph_commit_id, + &commit_id, &ChangeFeedScope::default(), None, Some(10), @@ -85,45 +445,14 @@ node Company { )) .await; let page = page.unwrap(); - assert_eq!( - page.block.changes.len(), - 1, - "the measured commit is a one-row update" - ); - eprintln!( - "PAGE person_commits={person_commits}: data_open_count={} data_reads={} \ - manifest_reads={} manifest_scan_count={} internal_open_count={}", - io.data_open_count, - io.data_reads, - io.manifest_reads, - io.manifest_scan_count, - io.internal_open_count, - ); + assert_eq!(page.block.changes.len(), 1, "the measured commit is one delete"); curve.push((person_commits, io)); } - for (person_commits, io) in &curve { - assert!( - io.data_open_count <= 2, - "one changed interval opens at most its parent and child pinned \ - datasets; untouched tables contribute zero \ - (person_commits={person_commits}): {io:?}" - ); - } - assert_flat(&curve, |io| io.data_open_count, 0, "changed-interval opens"); - // Graph-commit depth is identical at both points, so manifest work - // must not move with the scanned table's extent. - assert_flat(&curve, |io| io.manifest_reads, 0, "manifest reads per page"); - // The honest non-flat pin: the exact ordered merge reads both pinned - // table versions in full, so data reads grow with the table's physical - // extent at fixed Δ. Candidate pruning over - // `_row_last_updated_at_version` plus exact parent membership probes is - // the planned fix; when it lands, replace this tripwire with an - // `assert_flat`. assert_grows( &curve, |io| io.data_reads, 1, - "exact ordered-merge full-table scan term (O(table extent), not O(delta))", + "unproven-op fallback still reads both pinned versions (O(dataset extent))", ); }) .await; @@ -204,7 +533,7 @@ node Document { /// A caught-up poll examines zero commits: it captures the cut, proves the /// cursor current, and touches NO data tables — so data work is flat (zero -/// opens) regardless of table extent. The manifest capture term is printed as +/// opens) regardless of dataset extent. The manifest capture term is printed as /// a recorded diagnostic; it is the known `__manifest` fold cost, not claimed /// flat here. #[tokio::test] diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs index a1c67828..32d8a83b 100644 --- a/crates/omnigraph/tests/failpoints.rs +++ b/crates/omnigraph/tests/failpoints.rs @@ -11315,6 +11315,149 @@ node Document { ); } +/// The THIRD ABA window: after the final post-open logical head witness, no +/// step of the poll may read the branch's numeric-path history live. Version +/// manifests sit at replaceable numeric paths (unlike UUID-named data and +/// transaction files), so a future `(begin, end]` history walk placed after +/// `CHANGE_FEED_POST_HEAD_WITNESS` could classify replacement history. The +/// current adjacent classifier reads the transaction referenced by the pinned +/// child manifest and stores its complete plan before this witness; this cell +/// locks that placement against a future widening. When replacement history at +/// the same versions is row-set-preserving while the ORIGINAL commit carried a +/// delete, a live post-witness classifier would silently omit that delete. The +/// poll may instead fail loudly (reader survival across branch recreation is +/// not promised), but any page it returns must carry the original delete. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn change_feed_poll_classifies_intervals_before_the_head_witness() { + use omnigraph::changes::{ + ChangeFeedPosition, ChangeFeedRequest, ChangeFeedScope, ChangeFeedStart, ChangeOpKind, + }; + + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let schema = r#" +node Document { + title: String @key + content: String +} +"#; + let setup = Omnigraph::init(&uri, schema).await.unwrap(); + setup + .load( + "main", + r#"{"type":"Document","data":{"title":"base","content":"base"}}"#, + LoadMode::Overwrite, + ) + .await + .unwrap(); + setup.branch_create("feature").await.unwrap(); + // One load -> one fragment holding both rows. The follow-up delete then + // only adds a deletion vector, so across the delete commit's interval the + // retained manifests' fragment sets are identical — a wrongly-pruned + // enumeration derives an EMPTY candidate set and emits nothing. + let insert = setup + .load_with_receipt( + "feature", + concat!( + r#"{"type":"Document","data":{"title":"keep","content":"kept"}}"#, + "\n", + r#"{"type":"Document","data":{"title":"victim","content":"doomed"}}"#, + ), + LoadMode::Merge, + ) + .await + .unwrap(); + let insert_commit_id = insert.commit.graph_commit_id.clone(); + setup + .mutate( + "feature", + r#" +query remove_victim() { + delete Document where title = "victim" +} +"#, + "remove_victim", + &mixed_params(&[], &[]), + ) + .await + .unwrap(); + drop(setup); + + let reader = Omnigraph::open(&uri).await.unwrap(); + let control = Omnigraph::open(&uri).await.unwrap(); + let old_entry = control + .snapshot_of(ReadTarget::branch("feature")) + .await + .unwrap() + .dataset("node:Document") + .unwrap() + .clone(); + + let rendezvous = + helpers::failpoint::Rendezvous::park_first(names::CHANGE_FEED_POST_HEAD_WITNESS); + let request = ChangeFeedRequest { + branch: Some("feature".to_string()), + position: ChangeFeedPosition::Start(ChangeFeedStart::AfterCommit(insert_commit_id)), + scope: ChangeFeedScope::default(), + max_changes: None, + max_bytes: None, + max_commits: None, + }; + let poll_task = tokio::spawn(async move { reader.poll_change_feed(request).await }); + rendezvous.wait_until_reached().await; + + // Replace the branch with history that is provably row-set-preserving at + // the same table versions: two marker-carrying keyed loads, landing the + // replacement's numeric version manifests exactly where the original + // insert + delete commits left theirs. + let replacement = async { + control.branch_delete("feature").await?; + control.branch_create("feature").await?; + control + .load( + "feature", + r#"{"type":"Document","data":{"title":"repl-one","content":"one"}}"#, + LoadMode::Merge, + ) + .await?; + control + .load( + "feature", + r#"{"type":"Document","data":{"title":"repl-two","content":"two"}}"#, + LoadMode::Merge, + ) + .await?; + control.snapshot_of(ReadTarget::branch("feature")).await + } + .await; + rendezvous.release(); + + let new_snapshot = replacement.expect("delete/recreate replacement must complete"); + let new_entry = new_snapshot.dataset("node:Document").unwrap(); + assert_eq!( + new_entry.published_dataset_version, old_entry.published_dataset_version, + "the regression must exercise same-version branch ABA" + ); + + if let Ok(page) = poll_task.await.unwrap() { + let carries_original_delete = page + .blocks + .iter() + .flat_map(|block| block.changes.iter()) + .any(|change| change.op == ChangeOpKind::Delete && change.id.contains("victim")); + assert!( + carries_original_delete, + "a page returned across the in-poll delete/recreate must still carry the \ + original commit's delete; omitting it silently is the classification ABA \ + this cell pins: {page:?}" + ); + } + // A loud refusal is acceptable: reader survival across branch recreation + // is not promised — only never-silent retargeting. +} + async fn setup_diverged_merge_branches(dir: &tempfile::TempDir) -> (String, usize) { let uri = dir.path().to_str().unwrap().to_string(); let db = helpers::init_and_load(dir).await; diff --git a/crates/omnigraph/tests/forbidden_apis.rs b/crates/omnigraph/tests/forbidden_apis.rs index 1d6710ca..4be6a68c 100644 --- a/crates/omnigraph/tests/forbidden_apis.rs +++ b/crates/omnigraph/tests/forbidden_apis.rs @@ -805,11 +805,14 @@ durable_calls! { ("db/omnigraph.rs", ".dataset()", 1, WriteProtocol::ReadOnlyAccess), ("db/omnigraph/table_ops.rs", ".dataset()", 1, WriteProtocol::ReadOnlyAccess), ("db/omnigraph/export.rs", ".dataset()", 1, WriteProtocol::ReadOnlyAccess), - // Commit-change enumeration: pinned parent/child handles for typed row - // comparison, lazy image materialization, and descriptor-tie payload - // reads. Read-only by construction — the enumerator stages no transaction - // and publishes nothing. - ("changes/enumerate.rs", ".dataset()", 6, WriteProtocol::ReadOnlyAccess), + // Commit-change enumeration: pinned parent/child handles for the ordered + // merge's typed row comparison. Read-only by construction — the enumerator + // stages no transaction and publishes nothing. + ("changes/enumerate.rs", ".dataset()", 2, WriteProtocol::ReadOnlyAccess), + // Candidate-pruning emitter: pinned parent/child handles for the O(delta) + // candidate scan + parent before-image probe and the full-merge fallback. + // Read-only — it stages and publishes nothing. + ("changes/candidate_scan.rs", ".dataset()", 4, WriteProtocol::ReadOnlyAccess), // Net-diff cross-branch path: the same typed row comparison over two // pinned snapshot handles. Read-only — the diff stages and publishes // nothing. @@ -1903,6 +1906,38 @@ fn proven_insert_capability_has_one_production_mint_site() { ); } +/// The CDC candidate-pruning classifier (`changes::candidate_scan`) treats every +/// persisted `Operation::Update` as row-set-preserving, deriving the change feed +/// from a candidate scan without a delete pass. That is sound only because +/// OmniGraph's `merge_insert` never deletes an unmatched-by-source row — Lance +/// defaults the by-source arm to Keep and no engine code sets it otherwise. If a +/// delete-capable by-source merge arm were ever introduced, a persisted +/// `Operation::Update` could remove rows and the feed would silently drop that +/// delete. Lock the floor: the by-source merge arm must be absent from engine +/// source (production or test), so introducing one forces this classifier to be +/// re-gated first. +#[test] +fn no_delete_capable_merge_arm_in_engine_source() { + let src = engine_src_root(); + let mut offenders: Vec = Vec::new(); + for file in walk_rust_files(&src) { + let contents = std::fs::read_to_string(&file) + .unwrap_or_else(|error| panic!("failed to read {}: {error}", file.display())); + if contents.contains("WhenNotMatchedBySource") + || contents.contains("when_not_matched_by_source") + { + offenders.push(relative_to_src(&src, &file)); + } + } + assert!( + offenders.is_empty(), + "a by-source merge arm appeared in engine source, which can make an \ + Operation::Update delete rows. The CDC candidate-pruning classifier \ + (changes::candidate_scan) assumes every Update is non-deleting; re-gate \ + it before introducing this. Found in: {offenders:?}" + ); +} + #[test] fn graph_visible_write_chokepoints_are_registered() { let src = engine_src_root(); diff --git a/crates/omnigraph/tests/helpers/cost.rs b/crates/omnigraph/tests/helpers/cost.rs index 8286a2e9..9cb990b3 100644 --- a/crates/omnigraph/tests/helpers/cost.rs +++ b/crates/omnigraph/tests/helpers/cost.rs @@ -256,6 +256,15 @@ pub struct IoCounts { /// CPU/allocation term that grows with the backlog independently of the /// manifest/data IO counters. pub feed_commits_visited: u64, + /// CDC candidate-planning and bounded-page CPU/memory proxies. These are + /// explicit production seams because object-store read counts cannot reveal + /// repeated transaction-history walks or full-manifest iteration. + pub candidate_transaction_reads: u64, + pub candidate_fragment_metadata_steps: u64, + pub candidate_rows_examined: u64, + pub candidate_scan_target_rows_peak: u64, + pub candidate_scan_target_bytes_peak: u64, + pub change_images_materialized: u64, } impl IoCounts { @@ -503,6 +512,12 @@ struct OpProbes { internal_open_count: Arc, manifest_scan_count: Arc, feed_commits_visited: Arc, + candidate_transaction_reads: Arc, + candidate_fragment_metadata_steps: Arc, + candidate_rows_examined: Arc, + candidate_scan_target_rows_peak: Arc, + candidate_scan_target_bytes_peak: Arc, + change_images_materialized: Arc, } impl OpProbes { @@ -523,6 +538,12 @@ impl OpProbes { internal_open_count: Arc::new(AtomicU64::new(0)), manifest_scan_count: Arc::new(AtomicU64::new(0)), feed_commits_visited: Arc::new(AtomicU64::new(0)), + candidate_transaction_reads: Arc::new(AtomicU64::new(0)), + candidate_fragment_metadata_steps: Arc::new(AtomicU64::new(0)), + candidate_rows_examined: Arc::new(AtomicU64::new(0)), + candidate_scan_target_rows_peak: Arc::new(AtomicU64::new(0)), + candidate_scan_target_bytes_peak: Arc::new(AtomicU64::new(0)), + change_images_materialized: Arc::new(AtomicU64::new(0)), }; let probes = QueryIoProbes { manifest_wrapper: Some(Arc::new(h.manifest.clone()) as Arc), @@ -532,6 +553,12 @@ impl OpProbes { internal_open_count: Arc::clone(&h.internal_open_count), manifest_scan_count: Arc::clone(&h.manifest_scan_count), feed_commits_visited: Arc::clone(&h.feed_commits_visited), + candidate_transaction_reads: Arc::clone(&h.candidate_transaction_reads), + candidate_fragment_metadata_steps: Arc::clone(&h.candidate_fragment_metadata_steps), + candidate_rows_examined: Arc::clone(&h.candidate_rows_examined), + candidate_scan_target_rows_peak: Arc::clone(&h.candidate_scan_target_rows_peak), + candidate_scan_target_bytes_peak: Arc::clone(&h.candidate_scan_target_bytes_peak), + change_images_materialized: Arc::clone(&h.change_images_materialized), // graph_build_count / graph_edges_built unused by this harness. ..Default::default() }; @@ -564,6 +591,18 @@ impl OpProbes { internal_open_count: self.internal_open_count.load(Ordering::Relaxed), manifest_scan_count: self.manifest_scan_count.load(Ordering::Relaxed), feed_commits_visited: self.feed_commits_visited.load(Ordering::Relaxed), + candidate_transaction_reads: self.candidate_transaction_reads.load(Ordering::Relaxed), + candidate_fragment_metadata_steps: self + .candidate_fragment_metadata_steps + .load(Ordering::Relaxed), + candidate_rows_examined: self.candidate_rows_examined.load(Ordering::Relaxed), + candidate_scan_target_rows_peak: self + .candidate_scan_target_rows_peak + .load(Ordering::Relaxed), + candidate_scan_target_bytes_peak: self + .candidate_scan_target_bytes_peak + .load(Ordering::Relaxed), + change_images_materialized: self.change_images_materialized.load(Ordering::Relaxed), } } } diff --git a/docs/dev/testing.md b/docs/dev/testing.md index f243410d..e1e2255f 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -147,7 +147,7 @@ it is not inferred from a local syntax check. See [ci.md](ci.md). | `lifecycle.rs` | Graph lifecycle and schema state, including the v6 creation invariant that every fresh node/edge dataset declares exactly physical non-null `id` as Lance's unenforced primary key. The same fresh-dataset matrix proves every graph user field carries its catalog `omnigraph.stable_property_id`, while `id`/`src`/`dst` carry none. `open_accepts_historical_body_unique_blob_but_init_rejects_it` constructs a coherent old accepted contract and pins the narrow compatibility boundary: the root opens, while new admission rejects the same source. | | `point_in_time.rs` | Snapshots and time travel (`snapshot_at_graph_manifest_version`, `entity_at`) | | `changes.rs` | `diff_between` / `diff_commits`, including immutable-identity dataset pairing: pure renames stay empty while drop/re-add under one alias remains two dataset lifetimes. The write-receipt cell proves an effectful mutation and Load return the exact durable commit from publication, while a zero-entity mutation returns no commit and leaves branch lineage unchanged. Also the owner of the change surfaces: per-commit entity pages (id-ordered nodes-before-edges emission, exact update before/after images including null-vs-empty and per-image edge endpoints, commit-era schema decoding with rename-stable opaque type ids, unchanged-Blob suppression and empty physical-only blocks, typed parentless/schema-boundary/page-token refusals, reclaim → typed feed gap) and the durable feed (block-boundary-only cursors with pinned cuts, start modes with named-branch inheritance, first-parent merge blocks carrying the merged parent, cursor scope/witness/genesis rejections incl. warm named-branch delete/recreate ABA, commit-ceiling-bounded sparse polls, stateless cross-handle resume, and the gap → baseline-reset handshake with its failing-writer no-cursor guarantee). Long shared-prefix IDs pin exact-or-prefix/digest continuation positions, arbitrarily long branch names pin fixed-size branch scopes, and both resume commit/feed pages without duplicates; a positive remainder at a commit boundary pins the page-wide solo-change exception. `commit_changes_detects_same_length_blob_update_after_overwrite` and its feed twin pin the data-file-path Blob identity: a same-length managed-Blob update through a full-dataset Overwrite (which resets fragment ids) is detected, where the retired fragment-id qualifier aliased it as unchanged. `change_feed_poll_follows_commits_from_another_handle` pins that a warm handle's live-read refresh re-reads lineage when the durable head is a commit its projection lacks (a state-only refresh permanently broke later polls with a missing-commit error), and `change_feed_byte_budget_admits_one_solo_oversized_change_per_page` pins that an exhausted byte budget stops at the block boundary instead of force-emitting one oversized change per remaining commit. | -| `changes_cost.rs` | Cost budgets for the change surfaces on `helpers::cost`: per-page dataset opens bounded at two per changed interval and Blob payload work tracking emitted changes (flat), the exact ordered-merge full-dataset scan term pinned as a GROWING tripwire (O(dataset extent), not O(delta) — substrate candidate pruning must flip it), caught-up feed polls data-flat AND (swept over commit-history depth, not entities) manifest-reads-flat — the warm-coordinator reuse that keeps a caught-up same-branch poll from paying an O(history) `__manifest` fold — and the backlog walk's one-graph-manifest-snapshot-per-commit term pinned as growing with per-commit opens bounded. The backlog cell pins `feed_commits_visited` (the chain-walk CPU term, invisible to the IO counters) equal to the backlog for an unbounded-ceiling poll, and `change_feed_small_ceiling_poll_is_bounded_across_backlog_depths` pins the bounded forward-child projection: a `max_commits=1` poll walks exactly two commits (one emitted + one sentinel) with graph-manifest reads and dataset opens flat across backlog depths. | +| `changes_cost.rs` | Cost budgets for the change surfaces on `helpers::cost`: per-page dataset opens bounded at two per changed interval and Blob payload work tracking emitted changes (flat). The adjacent proven path scans transaction-touched fragments without a secondary-index dependency: absent-index and stale/partial-index extent curves keep data reads flat, `candidate_transaction_reads` is exactly one, and `candidate_fragment_metadata_steps` permits only logarithmic manifest lookup plus touched fragments; the conservative exact ordered-merge fallback remains a GROWING extent tripwire. `changes_page_size_one_bounds_large_candidate_delta` resumes a 2,048-row Blob delta across two stateless `max_changes=1` pages and on each pins two candidates examined (one emission + sentinel), row target 2, the caller's byte target, and exactly one materialized image (the sentinel performs no JSON/Blob work); `changes_page_size_one_skips_transaction_history_for_multi_version_intervals` adopts two raw logical updates in one forced-repair graph commit, resumes it across two stateless size-one pages, and pins zero candidate transaction reads on both. Caught-up feed polls stay data-flat and manifest-read-flat over history depth; backlog cells pin `feed_commits_visited` to backlog for an unbounded poll and exactly two commits (one emitted + sentinel) for `max_commits=1`, with graph-manifest reads and dataset opens flat across backlog depths. | | `src/db/graph_coordinator.rs` | Crate-internal coordinator classification, including RFC-030's direct/reversed/arbitrary/merge range matrix: only the child's persisted first-parent pointer creates a `FirstParentEdge`; a merged parent remains provenance and classifies as an arbitrary endpoint range. | | `src/table_store.rs` | The ordered-scan unit owner forces a global `id` sort through a 2 MiB pool, proves nonzero spill count/bytes/rows and stable ordering, then proves a one-byte scratch quota emits no row and survives the Lance stream boundary as a typed resource error. The same cell pins fail-closed behavior when spilling is disabled. | | `consistency.rs` | Cross-dataset snapshot isolation and atomic publish; RFC-023 cells prove `LoadMode::Append` is strict (existing `id` rejected without update/version movement), pin the inclusive 8,192-entity keyed-load ceiling with a one-over pre-effect refusal, prove that refusal does not poison a following strict Overwrite above the keyed ceiling, reject an input above 32 MiB through the shared Mutation/Load staging seam with raw dataset HEAD/graph-manifest/sidecar unchanged, and pin the external-source failure ladder on a lazy branch: default deny returns typed policy failure without a probe, an allowed missing object returns typed source failure, an allowed oversized object is rejected from metadata before payload access/ref creation/sidecar arm, and two individually valid half-limit sources selected for different datasets share one operation-wide 32 MiB copy budget and are both refused before either payload read. The same owner distinguishes the generic external-ingress bound from the keyed entity cap: Overwrite accepts exactly 8,192 external URI cells with one normalized HEAD and no payload GET, while 8,193 cells split across two individually legal datasets return typed `resource_limit` before preflight, lazy-ref creation, dataset/graph-manifest movement, or recovery arm. A barrier-synchronized stress cell over 16 pre-opened handles proves one same-key winner, 15 typed `KeyConflict` losers, exactly one stored entity carrying the winner's value, and survival of disjoint IDs. | diff --git a/docs/rfcs/0030-cdc-time-travel.md b/docs/rfcs/0030-cdc-time-travel.md index b63f0773..54051a24 100644 --- a/docs/rfcs/0030-cdc-time-travel.md +++ b/docs/rfcs/0030-cdc-time-travel.md @@ -368,10 +368,15 @@ snapshot. An optimization may skip this comparison only after inspecting every exact table transaction in the interval and proving that every operation is a mature, -row-set-preserving shape used by OmniGraph. Missing transaction files, cleaned -version holes, `Overwrite`, `Restore`, delete-capable `Update`, unknown/new -operation variants, and experimental Lance operations all mean **unknown** and -fall back to the exact ID comparison. The optimization is never authority. +row-set-preserving shape used by OmniGraph. A `RewriteRows` `Update` proves this +only by carrying a durable OmniGraph provenance marker (`omnigraph.no_by_source_delete`, +stamped on every keyed write, or the `insert_absence` certificate) — the op +shape alone is insufficient because `repair --force` can adopt an external Lance +merge that persists a delete-capable `Update{RewriteRows}`. Missing transaction +files, cleaned version holes, `Overwrite`, `Restore`, an unmarked or +delete-capable `Update`, unknown/new operation variants, and experimental Lance +operations all mean **unknown** and fall back to the exact ID comparison. The +optimization is never authority. Forbidden delete shortcuts: @@ -658,6 +663,11 @@ Extend existing owners before adding a new test silo: `changes.rs`, merge-delete, compaction, and missing transaction files. - A source-walk or exhaustive match makes new Lance `Operation` variants fall back to exact ID comparison until reviewed. +- A `RewriteRows` `Update` prunes only with a durable OmniGraph provenance proof + (`omnigraph.no_by_source_delete` marker or `insert_absence`); a marker-less + `Update` (an external delete-capable merge adopted via `repair --force`) falls + back. The write path stamps the marker at the one keyed merge chokepoint and it + survives commit → `list_transactions`. ### G0 — graph semantics @@ -783,8 +793,8 @@ compatible graph schema established out of band. ## 11. Format and compatibility audit -The C0–C4 core below persists nothing and therefore requires no internal-schema -or recovery-schema bump: +The C0–C4 read surfaces below persist nothing and therefore require no +internal-schema or recovery-schema bump: - lineage and table pins already exist; - graph type identity already exists in accepted SchemaIR and is projected @@ -793,12 +803,26 @@ or recovery-schema bump: - page tokens and cursors are caller-owned wire values; - typed errors and new read APIs are additive. +One write-path addition accompanies the candidate-pruning optimization and is +audited here as §11 requires for any persisted operation summary: every +**general keyed MergeInsert update** stamps the advisory +`omnigraph.no_by_source_delete=v1` Lance transaction property (proven strict +inserts carry the `insert_absence` certificate instead — a separate, +pre-existing property). The conclusion stands — **no format bump** — because +the marker is read-advisory in every direction: a missing marker only forces +the exact-merge fallback, never a correctness change, so pre-marker history and +foreign transactions degrade to the authority path; older binaries ignore +unknown transaction properties; and neither recovery nor publication consults +it. It is an optimization-eligibility proof carried inside Lance's existing +transaction-property surface, not a stored watermark, feed offset, or +tombstone. + Opaque page tokens and cursors have separate wire versions and decoders. An unsupported version or cross-use is a typed error, not best-effort decoding. -Any implementation that proposes a stored watermark, feed offset, operation -summary, delete tombstone, or historical SchemaIR changes this conclusion and -must return to this RFC's format audit before landing. +Any implementation that proposes a stored watermark, feed offset, a NON-advisory +operation summary, delete tombstone, or historical SchemaIR changes this +conclusion and must return to this RFC's format audit before landing. ## 12. Phasing @@ -847,13 +871,66 @@ whose cost is justified by C1. C0 through C3 shipped on the surveyed contract. Details frozen by the implementation, recorded here so later phases inherit them: -- **v1 derivation is the exact ordered-merge authority path only.** No - row-version candidate pruning and no transaction-interval no-delete proof - shipped; both remain the sanctioned optimizations of §4.2/§4.3. The cost - instrument (`changes_cost.rs`) pins the O(table-extent) scan term as a - growing tripwire that the pruning slice must flip to a flat assertion, and - pins bounded per-page opens, Blob-lazy payload work, data-flat caught-up - polls, and the one-manifest-snapshot-per-commit backlog term. +- **Candidate pruning shipped (§4.2/§4.3).** The exact ordered merge remains the + authority path. The optimized contract is deliberately narrower: only one + **adjacent** Lance version (`end == begin + 1`) can prune. Classification reads + the already-open child's one transaction, requires its `read_version` to name + the pinned parent exactly, and accepts only `Append` or a `RewriteRows` merge + `Update` (an exhaustive, wildcard-free `Operation` match, so a new Lance + variant compile-errors into review). A wider graph-visible interval falls back + **before any transaction read**; stateless page-size-one resumes therefore + never replay a 1,024-version history walk. + + The accepted plan is derived from the transaction footprint, not a full + manifest diff and not a secondary index. Lance manifests store fragments in + id order, fragment ids never recycle, and Append/Update assigns new fragments + consecutively above the parent high-water mark. The implementation + binary-searches the child for that suffix, verifies its count, consecutive ids, + and resulting high-water mark against the transaction, then binary-searches + the parent for the Update's `removed_fragment_ids` and `updated_fragments`. + Candidate rows are the new child fragments filtered by + `_row_last_updated_at_version ∈ (begin, end]`; before-images come from an + id-ordered scan of only those transaction-touched parent fragments. The two + streams merge one row at a time through the same typed + `rows_equal`/`emitted_image` machinery. There is no `id IN` probe and no BTREE + coverage precondition, so absent and partially covered indexes have the same + execution shape. Metadata planning costs + `O(log F + T log F)` for `F` manifest fragments and `T` touched parent/new + fragments; data work is bounded by the new fragments plus rows in touched + parent fragments, not total dataset extent. This is the precise shipped claim + (not unconditional `O(delta)` for an arbitrarily large touched fragment). + + A `RewriteRows` `Update` is trusted as delete-free only with a **durable + per-transaction provenance proof** — the `omnigraph.no_by_source_delete` + marker every general keyed MergeInsert update stamps + (`table_store::stamp_no_by_source_delete` at the one keyed merge chokepoint), + or the RFC-023 `insert_absence` certificate carried by proven strict inserts. + The op shape plus D2 and the retained `forbidden_apis.rs` source guard + (`no_delete_capable_merge_arm_in_engine_source`, now defense-in-depth) prove + only that current engine code builds no by-source-delete arm; they cannot + authenticate a persisted `Update` adopted through `repair --force --confirm`. + An Update carrying neither proof therefore falls back, as does any removing, + unknown, non-adjacent, missing-transaction, identity, or branch mismatch. + Pruning additionally requires every new child fragment's + `_row_last_updated_at_version` sequence to be inline, decodable, and exactly + `physical_rows` long; pinned Lance 10 otherwise silently supplies version 1, + which could hide a real update for `begin > 1`. + + Candidate scans derive their row target from the remaining page rows plus one + continuation sentinel and their byte target from the remaining page bytes. + `OrderedRows` retains one scanner batch and prepares only the current row; the + former 8,192-`RawRow` queue, cloned id vector, parent HashMap, and queued Emits + are gone. An exhausted page consumes one sentinel to prove continuation but + does not materialize its JSON or Blob payload. `changes_cost.rs` pins all of + these terms: absent and stale-index extent curves remain flat in data reads; + fragment-metadata steps stay logarithmic; adjacent intervals read exactly one + transaction; a forced-repair graph commit adopting two raw logical updates + resumes across two size-one pages with zero candidate transaction reads on + both; and each of two stateless size-one pages over a 2,048-row Blob delta + examines exactly two candidates with row target 2 and the caller's byte target + while materializing only the emitted image (the sentinel performs no JSON or + Blob payload work). + The exact-merge fallback retains a companion growing extent tripwire. - **Typed structural equality** uses Arrow logical equality on one-row slices for non-Blob user columns and physical descriptor identity with an exact payload tie-break for Blob columns. Float comparison is bitwise. diff --git a/tools/omnigraph-vocabulary-guard/graph-vocabulary-inventory.tsv b/tools/omnigraph-vocabulary-guard/graph-vocabulary-inventory.tsv index 4f15bf8e..234f1aa8 100644 --- a/tools/omnigraph-vocabulary-guard/graph-vocabulary-inventory.tsv +++ b/tools/omnigraph-vocabulary-guard/graph-vocabulary-inventory.tsv @@ -77,6 +77,7 @@ schema_version occurrence_id surface source_path boundary site_kind current_term 2 public_rust:omnigraph-engine:1149c140cabadc7bb2e45ae78b08a7186357118cfe77aa9f2339021226018135:SIDECAR:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::ENSURE_INDICES_POST_SIDECAR_PRE_FORK: &str public_signature SIDECAR retain physical_storage retain accurate recovery-storage symbol none omnigraph-vocabulary-guard::public_rust_inventory_matches Sidecar names identify durable recovery artifacts or their failpoint boundaries. 2 public_rust:omnigraph-engine:189ca9381613b4aa8847c85a798ec59be9df1b71dc225d678569ae08938d77d9:column:1:1 public_rust crates/omnigraph/Cargo.toml pub async fn omnigraph::db::manifest::SnapshotDataset::index_coverage(&self, column: &str) -> omnigraph::error::Result public_signature column retain physical_storage retain exact physical execution API none omnigraph-vocabulary-guard::public_rust_inventory_matches The parameter or field describes a backing Lance dataset column or persisted physical row count. 2 public_rust:omnigraph-engine:189ca9381613b4aa8847c85a798ec59be9df1b71dc225d678569ae08938d77d9:column:2:1 public_rust crates/omnigraph/Cargo.toml pub async fn omnigraph::db::manifest::SnapshotDataset::index_coverage(&self, column: &str) -> omnigraph::error::Result public_signature column retain physical_storage retain exact physical execution API none omnigraph-vocabulary-guard::public_rust_inventory_matches The parameter or field describes a backing Lance dataset column or persisted physical row count. +2 public_rust:omnigraph-engine:19b7f1a155cea5a636471facf54958dbd5a58f59749e8ffdeaa11474e4fc6657:rows:1:1 public_rust crates/omnigraph/Cargo.toml pub omnigraph::instrumentation::QueryIoProbes::candidate_rows_examined: alloc::sync::Arc public_signature rows retain physical_storage retain exact physical execution API none omnigraph-vocabulary-guard::public_rust_inventory_matches The symbol refers to a physical scanner, manifest row, execution counter, or backing-store hook. 2 public_rust:omnigraph-engine:242d25727dcdfd6f4ee5221fc92e7d8c9f7db632d8e52f4356c098f98ad9e91c:TABLE:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::INIT_TABLE_CREATE_ACK_LOST: &str public_signature TABLE retain physical_storage retain exact physical failpoint name none omnigraph-vocabulary-guard::public_rust_inventory_matches The exported failpoint names an internal backing-dataset effect boundary. 2 public_rust:omnigraph-engine:34f5d948b5069130b51b4393505f8620473aa107bd5ff85fa7f338831c94dd20:rows:1:1 public_rust crates/omnigraph/Cargo.toml pub omnigraph::instrumentation::MergeWriteProbes::ordered_cursor_batch_rows: alloc::sync::Arc public_signature rows retain physical_storage retain exact physical execution API none omnigraph-vocabulary-guard::public_rust_inventory_matches The symbol refers to a physical scanner, manifest row, execution counter, or backing-store hook. 2 public_rust:omnigraph-engine:3fa08f31a6874e00c7c736d699247043fb125a878781a45cc5bbe89fd13c3fa2:TABLE:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::CHANGE_FEED_PRE_TABLE_OPEN: &str public_signature TABLE retain physical_storage retain exact physical failpoint name none omnigraph-vocabulary-guard::public_rust_inventory_matches The exported failpoint names an internal backing-dataset effect boundary. @@ -91,6 +92,7 @@ schema_version occurrence_id surface source_path boundary site_kind current_term 2 public_rust:omnigraph-engine:6a3f5498e95f931240dee3fd0206826bff8ad59a21aa88c1cb868988ef279f98:TABLE:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::RECOVERY_POST_TABLE_RESTORE_PRE_PUBLISH: &str public_signature TABLE retain physical_storage retain exact physical failpoint name none omnigraph-vocabulary-guard::public_rust_inventory_matches The exported failpoint names an internal backing-dataset effect boundary. 2 public_rust:omnigraph-engine:707df5ce587b6230e46594bcbd56a6bf53af9bbc5cfc7fc4660661ad7e19f328:TABLE:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::BRANCH_DELETE_POST_TABLE_GATES: &str public_signature TABLE retain physical_storage retain exact physical failpoint name none omnigraph-vocabulary-guard::public_rust_inventory_matches The exported failpoint names an internal backing-dataset effect boundary. 2 public_rust:omnigraph-engine:74bbbec0e92fea105d49b3398aa8150dd60b03a9b96db534cd9c25f139fd0f9c:TABLE:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::MUTATION_POST_TABLE_COMMIT: &str public_signature TABLE retain physical_storage retain exact physical failpoint name none omnigraph-vocabulary-guard::public_rust_inventory_matches The exported failpoint names an internal backing-dataset effect boundary. +2 public_rust:omnigraph-engine:8b0b104018c1f633a5139bf1e57c950dac499fa39d04736a8f8147fa26e9fd72:rows:1:1 public_rust crates/omnigraph/Cargo.toml pub omnigraph::instrumentation::QueryIoProbes::candidate_scan_target_rows_peak: alloc::sync::Arc public_signature rows retain physical_storage retain exact physical execution API none omnigraph-vocabulary-guard::public_rust_inventory_matches The symbol refers to a physical scanner, manifest row, execution counter, or backing-store hook. 2 public_rust:omnigraph-engine:8e7b9c358276c87d5b1a6320dc6738b69af35e7e7ed540f127ee5eb195c40dbd:table:1:1 public_rust crates/omnigraph/Cargo.toml pub omnigraph::instrumentation::QueryIoProbes::table_wrapper: core::option::Option> public_signature table retain physical_storage retain exact physical execution API none omnigraph-vocabulary-guard::public_rust_inventory_matches The symbol refers to a physical scanner, manifest row, execution counter, or backing-store hook. 2 public_rust:omnigraph-engine:941854de670442d85d8da1e7443b876441519de07a731c912861c4a37d20d906:SIDECAR:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::RECOVERY_POST_SIDECAR_LIST_PRE_READ: &str public_signature SIDECAR retain physical_storage retain accurate recovery-storage symbol none omnigraph-vocabulary-guard::public_rust_inventory_matches Sidecar names identify durable recovery artifacts or their failpoint boundaries. 2 public_rust:omnigraph-engine:94e906e27d7b2c45a84f2c93d63c6092287dc7e47916f5987033c185d531cfbd:SIDECAR:1:1 public_rust crates/omnigraph/Cargo.toml pub const omnigraph::failpoints::names::RECOVERY_SIDECAR_LIST: &str public_signature SIDECAR retain physical_storage retain accurate recovery-storage symbol none omnigraph-vocabulary-guard::public_rust_inventory_matches Sidecar names identify durable recovery artifacts or their failpoint boundaries. @@ -119,17 +121,27 @@ schema_version occurrence_id surface source_path boundary site_kind current_term 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fblob.rs:%3Cmodule%3E%3A%3Aimpl%20Omnigraph%3A%3Afn%20read_blob_at:error_constructor:4012c881aa26714d8a3355f00fb8012e96485e0714c412fde195da7d13f8b468:row:1:1 rust_string crates/omnigraph/src/blob.rs ::impl Omnigraph::fn read_blob_at :: error_constructor :: 4012c881aa26714d8a3355f00fb8012e96485e0714c412fde195da7d13f8b468 #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fblob.rs:%3Cmodule%3E%3A%3Aimpl%20Omnigraph%3A%3Afn%20read_blob_at:error_constructor:755216c1043784f11d0f2e4eb0061d37e4ea3e8183962675428b4ee06e268409:row:1:1 rust_string crates/omnigraph/src/blob.rs ::impl Omnigraph::fn read_blob_at :: error_constructor :: 755216c1043784f11d0f2e4eb0061d37e4ea3e8183962675428b4ee06e268409 #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fblob.rs:%3Cmodule%3E%3A%3Astruct%20BlobEtag:rustdoc:2fb6853f38341f9acbcb3a2db0063eced906356081b1cf533ddac1e4eb9c79fb:table:1:1 rust_string crates/omnigraph/src/blob.rs ::struct BlobEtag :: rustdoc :: 2fb6853f38341f9acbcb3a2db0063eced906356081b1cf533ddac1e4eb9c79fb #1 rustdoc table table physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:1da8d623d16a86f4552c20e1e4147e7a913cf297e4d741ec1b2ef68adb7d2b27:Rows:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: 1da8d623d16a86f4552c20e1e4147e7a913cf297e4d741ec1b2ef68adb7d2b27 #1 rustdoc Rows Rows physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:1fd2876e9cd816c4671b09bf44905226bcdb54e1a7045d9f1c77f5e666ab5f08:Rows:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: 1fd2876e9cd816c4671b09bf44905226bcdb54e1a7045d9f1c77f5e666ab5f08 #1 rustdoc Rows Rows physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:1fd2876e9cd816c4671b09bf44905226bcdb54e1a7045d9f1c77f5e666ab5f08:row:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: 1fd2876e9cd816c4671b09bf44905226bcdb54e1a7045d9f1c77f5e666ab5f08 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:615d0b4fe585f0edd4a9ff6a09a4d61085bca1e042c44cf1b0efd393555cd85c:rows:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: 615d0b4fe585f0edd4a9ff6a09a4d61085bca1e042c44cf1b0efd393555cd85c #1 rustdoc rows rows physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:839b7c976901eff2e5adceadb3939984b31aba81df5629b7a2cbc0c144480931:row:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: 839b7c976901eff2e5adceadb3939984b31aba81df5629b7a2cbc0c144480931 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:86e981b16dbec56c6ca2cb1b747d982a8827ed32d6eedeab37492af56b248490:row:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: 86e981b16dbec56c6ca2cb1b747d982a8827ed32d6eedeab37492af56b248490 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:aefb3cf837246e304afabf952a4059ce2a0ff20fb750ab07fdd08b960343aee3:rows:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: aefb3cf837246e304afabf952a4059ce2a0ff20fb750ab07fdd08b960343aee3 #1 rustdoc rows rows physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:cce19026b5f1b261cdd730b75f8530dd510488e68452c556a534f2e7960e246e:row:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: cce19026b5f1b261cdd730b75f8530dd510488e68452c556a534f2e7960e246e #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fcandidate_scan.rs:%3Cmodule%3E:rustdoc:dbaf08f1fb99a23179ea281525163cde6c9c301e70ec97daaa67c2be674f8b78:row:1:1 rust_string crates/omnigraph/src/changes/candidate_scan.rs :: rustdoc :: dbaf08f1fb99a23179ea281525163cde6c9c301e70ec97daaa67c2be674f8b78 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fenumerate.rs:%3Cmodule%3E%3A%3Afn%20reprove_named_branch_heads:error_constructor:b8799d8146635c609d1b758ecf5cb581be45e1cf43571df2bbe16d5797aa0f94:table:1:1 rust_string crates/omnigraph/src/changes/enumerate.rs ::fn reprove_named_branch_heads :: error_constructor :: b8799d8146635c609d1b758ecf5cb581be45e1cf43571df2bbe16d5797aa0f94 #1 error_constructor table table physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fenumerate.rs:%3Cmodule%3E:rustdoc:5f0f709d193c9b10647df754c1c7d182d483387a765047ecafb6aaf8d27e8564:columns:1:1 rust_string crates/omnigraph/src/changes/enumerate.rs :: rustdoc :: 5f0f709d193c9b10647df754c1c7d182d483387a765047ecafb6aaf8d27e8564 #1 rustdoc columns columns physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fenumerate.rs:%3Cmodule%3E:rustdoc:e7e44d205e5a560e6a025cc9db4e98111c0279b9d0375e392329f0a8d76ab7d3:table:1:1 rust_string crates/omnigraph/src/changes/enumerate.rs :: rustdoc :: e7e44d205e5a560e6a025cc9db4e98111c0279b9d0375e392329f0a8d76ab7d3 #1 rustdoc table table physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fmodel.rs:%3Cmodule%3E:rustdoc:e804c4b033d09c66577e0e0e1ac6e7091f00718e5113597c6caa3b780bb8a915:row:1:1 rust_string crates/omnigraph/src/changes/model.rs :: rustdoc :: e804c4b033d09c66577e0e0e1ac6e7091f00718e5113597c6caa3b780bb8a915 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Fmodel.rs:%3Cmodule%3E:rustdoc:e804c4b033d09c66577e0e0e1ac6e7091f00718e5113597c6caa3b780bb8a915:table:1:1 rust_string crates/omnigraph/src/changes/model.rs :: rustdoc :: e804c4b033d09c66577e0e0e1ac6e7091f00718e5113597c6caa3b780bb8a915 #1 rustdoc table table physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Afn%20blob_values_for:error_constructor:9105d269e339a795cb324502455cbe9819558467f5a233b9367c75cdbf6283dc:row:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::fn blob_values_for :: error_constructor :: 9105d269e339a795cb324502455cbe9819558467f5a233b9367c75cdbf6283dc #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. -2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Afn%20prepare_batch:error_constructor:15a6fb48e432ed40ee6e2b945d259531c18a5fce6495aab40d8b513d11f03af2:column:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::fn prepare_batch :: error_constructor :: 15a6fb48e432ed40ee6e2b945d259531c18a5fce6495aab40d8b513d11f03af2 #1 error_constructor column column physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. -2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Afn%20prepare_batch:error_constructor:b5b584c2c9397fefc2c820b44dcdb16a825609b54761b099bd6affb0c1fedaea:row:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::fn prepare_batch :: error_constructor :: b5b584c2c9397fefc2c820b44dcdb16a825609b54761b099bd6affb0c1fedaea #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. -2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Afn%20prepare_batch:error_constructor:d7d533c1f1a78e511dd9d7bde2b6587d6bb3b7a809c336c4bc80e169f7b80ced:column:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::fn prepare_batch :: error_constructor :: d7d533c1f1a78e511dd9d7bde2b6587d6bb3b7a809c336c4bc80e169f7b80ced #1 error_constructor column column physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Afn%20rows_equal:error_constructor:12e03a6522e97e7178a996f86a4b0e727397725dab59079027eeca4eefa6de35:column:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::fn rows_equal :: error_constructor :: 12e03a6522e97e7178a996f86a4b0e727397725dab59079027eeca4eefa6de35 #1 error_constructor column column physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Afn%20rows_equal:error_constructor:12e03a6522e97e7178a996f86a4b0e727397725dab59079027eeca4eefa6de35:row:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::fn rows_equal :: error_constructor :: 12e03a6522e97e7178a996f86a4b0e727397725dab59079027eeca4eefa6de35 #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Aimpl%20BatchCursor%3A%3Afn%20try_new:error_constructor:15a6fb48e432ed40ee6e2b945d259531c18a5fce6495aab40d8b513d11f03af2:column:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::impl BatchCursor::fn try_new :: error_constructor :: 15a6fb48e432ed40ee6e2b945d259531c18a5fce6495aab40d8b513d11f03af2 #1 error_constructor column column physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Aimpl%20BatchCursor%3A%3Afn%20try_new:error_constructor:b5b584c2c9397fefc2c820b44dcdb16a825609b54761b099bd6affb0c1fedaea:row:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::impl BatchCursor::fn try_new :: error_constructor :: b5b584c2c9397fefc2c820b44dcdb16a825609b54761b099bd6affb0c1fedaea #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Aimpl%20BatchCursor%3A%3Afn%20try_new:error_constructor:b5b584c2c9397fefc2c820b44dcdb16a825609b54761b099bd6affb0c1fedaea:row:2:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::impl BatchCursor::fn try_new :: error_constructor :: b5b584c2c9397fefc2c820b44dcdb16a825609b54761b099bd6affb0c1fedaea #2 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Aimpl%20BatchCursor%3A%3Afn%20try_new:error_constructor:d7d533c1f1a78e511dd9d7bde2b6587d6bb3b7a809c336c4bc80e169f7b80ced:column:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::impl BatchCursor::fn try_new :: error_constructor :: d7d533c1f1a78e511dd9d7bde2b6587d6bb3b7a809c336c4bc80e169f7b80ced #1 error_constructor column column physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Aimpl%20RawRow%3A%3Afn%20single:error_constructor:dc342d97570182ad29a1f350e34601b88799c783158e6cfcc485676adee1a8ab:row:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs ::impl RawRow::fn single :: error_constructor :: dc342d97570182ad29a1f350e34601b88799c783158e6cfcc485676adee1a8ab #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E%3A%3Aimpl%20RawRow%3A%3Afn%20single:error_constructor:dc342d97570182ad29a1f350e34601b88799c783158e6cfcc485676adee1a8ab:row:1:2 rust_string crates/omnigraph/src/changes/row_compare.rs ::impl RawRow::fn single :: error_constructor :: dc342d97570182ad29a1f350e34601b88799c783158e6cfcc485676adee1a8ab #1 error_constructor row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Fchanges%2Frow_compare.rs:%3Cmodule%3E:rustdoc:037b5d7240a4b6465bcc1e2b4e30489a3315915344b258f820ca59d5a60ed187:row:1:1 rust_string crates/omnigraph/src/changes/row_compare.rs :: rustdoc :: 037b5d7240a4b6465bcc1e2b4e30489a3315915344b258f820ca59d5a60ed187 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. @@ -673,6 +685,10 @@ schema_version occurrence_id surface source_path boundary site_kind current_term 2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20MergeWriteProbes%3A%3Afield%20ordered_cursor_scan_calls:rustdoc:1e304cab8b39050587ef676c82622285ac75381920075b31c0cb2182a9e2bf13:row:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct MergeWriteProbes::field ordered_cursor_scan_calls :: rustdoc :: 1e304cab8b39050587ef676c82622285ac75381920075b31c0cb2182a9e2bf13 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20MergeWriteProbes%3A%3Afield%20stage_fenced_insert_calls:rustdoc:903fdbae9ad67e63d2838692f07524b553fdae7f40c9b359e922cefb74db9644:row:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct MergeWriteProbes::field stage_fenced_insert_calls :: rustdoc :: 903fdbae9ad67e63d2838692f07524b553fdae7f40c9b359e922cefb74db9644 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20MergeWriteProbes%3A%3Afield%20stage_vector_index_calls:rustdoc:d3f8c4f48c21aa4d38b455a5cdad48338fb008932c61ffcef28f8fe9624ff117:table:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct MergeWriteProbes::field stage_vector_index_calls :: rustdoc :: d3f8c4f48c21aa4d38b455a5cdad48338fb008932c61ffcef28f8fe9624ff117 #1 rustdoc table table physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20candidate_rows_examined:rustdoc:7ec685c85552106f46ad4621dfd5710899e70575854b5b7b9d1a8cafe2d969c4:row:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field candidate_rows_examined :: rustdoc :: 7ec685c85552106f46ad4621dfd5710899e70575854b5b7b9d1a8cafe2d969c4 #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20candidate_rows_examined:rustdoc:7ec685c85552106f46ad4621dfd5710899e70575854b5b7b9d1a8cafe2d969c4:rows:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field candidate_rows_examined :: rustdoc :: 7ec685c85552106f46ad4621dfd5710899e70575854b5b7b9d1a8cafe2d969c4 #1 rustdoc rows rows physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20candidate_rows_examined:rustdoc:915b63cbcf93389cbf193b861c33c3f76707b6cdd9c52721f0bc7978e948c0fd:rows:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field candidate_rows_examined :: rustdoc :: 915b63cbcf93389cbf193b861c33c3f76707b6cdd9c52721f0bc7978e948c0fd #1 rustdoc rows rows physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. +2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20candidate_scan_target_rows_peak:rustdoc:f65133c9a12e8083640690a790b5e75db16971e991d0c6b5bc87f194a00b40da:row:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field candidate_scan_target_rows_peak :: rustdoc :: f65133c9a12e8083640690a790b5e75db16971e991d0c6b5bc87f194a00b40da #1 rustdoc row row physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20data_open_count:rustdoc:1198ff770d6982ea46d3312f03b28d8337ac9bb601d274b8565123d7abb057cf:tables:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field data_open_count :: rustdoc :: 1198ff770d6982ea46d3312f03b28d8337ac9bb601d274b8565123d7abb057cf #1 rustdoc tables tables physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20data_open_count:rustdoc:1352673c40f9d64f8a784e4cbdb25af693fec573125a629ef8674f3135c07c3f:table:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field data_open_count :: rustdoc :: 1352673c40f9d64f8a784e4cbdb25af693fec573125a629ef8674f3135c07c3f #1 rustdoc table table physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology. 2 rust_string:crates%2Fomnigraph%2Fsrc%2Finstrumentation.rs:%3Cmodule%3E%3A%3Astruct%20QueryIoProbes%3A%3Afield%20data_open_count:rustdoc:4033d9f8054f02f762e07a3b2b5907d38ea1c8c27e737d9e72bd1b137a520991:tables:1:1 rust_string crates/omnigraph/src/instrumentation.rs ::struct QueryIoProbes::field data_open_count :: rustdoc :: 4033d9f8054f02f762e07a3b2b5907d38ea1c8c27e737d9e72bd1b137a520991 #1 rustdoc tables tables physical_storage retain accurate physical/storage terminology none omnigraph-engine owning tests plus G2 guard Reviewed as accurate Lance, Arrow, manifest, recovery, or physical execution terminology.