diff --git a/changelog.d/historical-query-property-identity.fixed.md b/changelog.d/historical-query-property-identity.fixed.md new file mode 100644 index 000000000..98115b7ad --- /dev/null +++ b/changelog.d/historical-query-property-identity.fixed.md @@ -0,0 +1,9 @@ +- A historical query no longer reads a dropped property's values through a + property later re-added with the same name. At a commit that stores a property + under another lifetime or another name, or not at all, the query is refused + with a request error naming the property, where before it returned the dropped + values or failed with an internal error. Commits that read table data written + before v0.10, which records no property identity, are refused the same way + unless that data is still current. See [historical reads][historical-query-property-identity-reads]. + +[historical-query-property-identity-reads]: ../docs/user/branching/index.md#historical-reads diff --git a/crates/omnigraph-catalog/src/lib.rs b/crates/omnigraph-catalog/src/lib.rs index 6e8afd676..469b296f4 100644 --- a/crates/omnigraph-catalog/src/lib.rs +++ b/crates/omnigraph-catalog/src/lib.rs @@ -264,6 +264,10 @@ pub struct Snapshot { /// snapshots, time-travel / Snapshot-id reads, and directly-built test /// snapshots, which fall back to a plain open. read_caches: Option, + /// Table keys this view pins at a version the accepted contract it plans + /// against was not published with: set by `bind_catalog_aliases` on a + /// historical query view, empty on every other view. + uncovered_tables: HashSet, } #[derive(Clone)] @@ -605,7 +609,13 @@ impl Snapshot { /// table lifetime under its current name. Conversely, a reused name whose /// current identity is absent from the historical snapshot is removed from /// the execution view, preventing cross-incarnation adoption. - pub fn bind_catalog_aliases(&mut self, catalog: &Catalog) -> Result<()> { + /// + /// `live` is the live view `catalog` was accepted against. A bound table + /// whose pinned version is not exactly the one `live` pins is recorded as + /// uncovered (`contract_covers`): the catalog's property names were not + /// published with that version, so a read there resolves each property by + /// its persisted stable property id instead. + pub fn bind_catalog_aliases(&mut self, catalog: &Catalog, live: &Snapshot) -> Result<()> { // Freeze the historical identity map before changing any aliases. A // renamed-away alias may later be reused by a new live type; resolving // that replacement first must not remove the only entry needed to bind @@ -640,6 +650,7 @@ impl Snapshot { })) .collect::>(); + self.uncovered_tables.clear(); for (table_key, stable_table_id, incarnation_id) in table_aliases { let identity = TableIdentity::new(stable_table_id, incarnation_id)?; @@ -648,12 +659,27 @@ impl Snapshot { // historical name. self.entries.remove(&table_key); if let Some(entry) = historical_by_identity.get(&identity) { + let covered = live + .entries + .get(&table_key) + .is_some_and(|live_entry| live_entry.same_physical_version(entry)); + if !covered { + self.uncovered_tables.insert(table_key.clone()); + } self.entries.insert(table_key, entry.clone()); } } Ok(()) } + /// Whether the accepted contract this view plans against was published + /// with the version it pins for `type_key`, so its property names name + /// that version's columns. False only on a historical query view, for a + /// table pinned at a version other than the live one. + pub fn contract_covers(&self, type_key: &str) -> bool { + !self.uncovered_tables.contains(type_key) + } + /// Open a backing dataset at its pinned version by qualified graph type key. With read caches present /// (live Branch reads), reuse a held handle through the cache (0 open IO on a /// warm repeat) and the shared `Session`; otherwise plain-open (Fix 2). @@ -1301,6 +1327,7 @@ impl ManifestCoordinator { graph_branch: None, native_branch: None, read_caches: None, + uncovered_tables: HashSet::new(), } } diff --git a/crates/omnigraph-catalog/src/state.rs b/crates/omnigraph-catalog/src/state.rs index de4f5f385..14b1ba654 100644 --- a/crates/omnigraph-catalog/src/state.rs +++ b/crates/omnigraph-catalog/src/state.rs @@ -56,6 +56,18 @@ impl DatasetEntry { && self.version_metadata == other.version_metadata && self.manifest_version == other.manifest_version } + + /// The same table lifetime pinned at the same physical Lance version, + /// whatever alias or publication registered it. + pub fn same_physical_version(&self, other: &DatasetEntry) -> bool { + self.identity == other.identity + && self.dataset_path == other.dataset_path + && self.published_dataset_version == other.published_dataset_version + && self.native_dataset_branch == other.native_dataset_branch + && self + .version_metadata + .same_read_witness(&other.version_metadata) + } } #[derive(Debug, Clone)] diff --git a/crates/omnigraph-catalog/src/tests.rs b/crates/omnigraph-catalog/src/tests.rs index c4f036276..4895d03a5 100644 --- a/crates/omnigraph-catalog/src/tests.rs +++ b/crates/omnigraph-catalog/src/tests.rs @@ -556,7 +556,14 @@ async fn historical_alias_binding_keeps_same_name_node_and_edge_identities_disti let edge_identity = snapshot.dataset("edge:Link").unwrap().identity; assert_ne!(node_identity, edge_identity); - snapshot.bind_catalog_aliases(&catalog).unwrap(); + // The live view pins the node table at the same version and the edge table + // at another one, so only the edge table is read outside the contract. + let mut live = snapshot.clone(); + live.entries + .get_mut("edge:Link") + .unwrap() + .published_dataset_version += 1; + snapshot.bind_catalog_aliases(&catalog, &live).unwrap(); assert_eq!( snapshot.dataset("node:Link").unwrap().identity, @@ -566,6 +573,12 @@ async fn historical_alias_binding_keeps_same_name_node_and_edge_identities_disti snapshot.dataset("edge:Link").unwrap().identity, edge_identity ); + assert!(snapshot.contract_covers("node:Link")); + assert!(!snapshot.contract_covers("edge:Link")); + + let current = snapshot.clone(); + snapshot.bind_catalog_aliases(&catalog, ¤t).unwrap(); + assert!(snapshot.contract_covers("edge:Link")); } #[tokio::test] diff --git a/crates/omnigraph-core/src/error.rs b/crates/omnigraph-core/src/error.rs index 7a15a386c..0f6ec20a8 100644 --- a/crates/omnigraph-core/src/error.rs +++ b/crates/omnigraph-core/src/error.rs @@ -124,6 +124,38 @@ pub fn missing_graph_type_at_snapshot(table_key: &str) -> String { ) } +/// The current property's lifetime is not in the table version a historical +/// read pins: it was added since, or the column of that name there belongs to +/// a dropped property. +pub fn missing_graph_property_at_snapshot(table_key: &str, property: &str) -> String { + format!( + "property '{property}' of {} does not exist at this snapshot", + graph_type_subject(table_key) + ) +} + +/// The current property's lifetime is in the pinned table version under the +/// name it had before a rename. +pub fn renamed_graph_property_at_snapshot( + table_key: &str, + property: &str, + stored_name: &str, +) -> String { + format!( + "property '{property}' of {} is named '{stored_name}' at this snapshot; a historical query cannot read a property across a rename", + graph_type_subject(table_key) + ) +} + +/// The pinned table version stores the column without a stable property id, +/// so it cannot show which property lifetime the column belongs to. +pub fn unwitnessed_graph_property_at_snapshot(table_key: &str, property: &str) -> String { + format!( + "property '{property}' of {} has no persisted property-lifetime witness at this snapshot", + graph_type_subject(table_key) + ) +} + fn format_key_conflict(type_key: &str) -> String { format!("{} already has this id", graph_type_subject(type_key)) } diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index bc01905d8..0b0beb987 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -1007,6 +1007,12 @@ impl Omnigraph { /// caller MUST already hold a schema permit (either side); this helper does /// not acquire one because the gate is non-reentrant on one task. pub(crate) async fn load_accepted_catalog_with_schema_gate_held(&self) -> Result> { + Ok(self.load_accepted_view_with_schema_gate_held().await?.0) + } + + /// The accepted catalog and the live view of the current branch it was + /// accepted against. + async fn load_accepted_view_with_schema_gate_held(&self) -> Result<(Arc, Snapshot)> { let current_branch = self .coordinator .read() @@ -1020,7 +1026,7 @@ impl Omnigraph { .snapshot; let (catalog, _) = self.accepted_catalog_for_snapshot(&snapshot).await?; validate_bound_catalog_against_snapshot(&catalog, &snapshot)?; - Ok(catalog) + Ok((catalog, snapshot)) } /// Accept the contract bytes in this snapshot before reusing its catalog. @@ -2154,12 +2160,12 @@ impl Omnigraph { validate_bound_catalog_against_snapshot(&catalog, &resolved.snapshot)?; return Ok((resolved, catalog)); } - let catalog = self.load_accepted_catalog_with_schema_gate_held().await?; + let (catalog, live) = self.load_accepted_view_with_schema_gate_held().await?; if bind_historical_aliases { let catalog = self .catalog_for_image_vintage(&resolved.snapshot, catalog) .await?; - resolved.snapshot.bind_catalog_aliases(&catalog)?; + resolved.snapshot.bind_catalog_aliases(&catalog, &live)?; return Ok((resolved, catalog)); } Ok((resolved, catalog)) @@ -2189,7 +2195,7 @@ impl Omnigraph { version: u64, ) -> Result<(Snapshot, Arc)> { let _schema_permit = self.write_queue().acquire_schema_shared().await; - let catalog = self.load_accepted_catalog_with_schema_gate_held().await?; + let (catalog, live) = self.load_accepted_view_with_schema_gate_held().await?; let branch = self .coordinator .read() @@ -2205,7 +2211,7 @@ impl Omnigraph { .await?, ); let catalog = self.catalog_for_image_vintage(&snapshot, catalog).await?; - snapshot.bind_catalog_aliases(&catalog)?; + snapshot.bind_catalog_aliases(&catalog, &live)?; Ok((snapshot, catalog)) } diff --git a/crates/omnigraph/src/db/snapshot.rs b/crates/omnigraph/src/db/snapshot.rs index d5b5fe061..99ea82e08 100644 --- a/crates/omnigraph/src/db/snapshot.rs +++ b/crates/omnigraph/src/db/snapshot.rs @@ -274,8 +274,18 @@ impl Snapshot { .map(SchemaContractIdentity::from) } - pub(crate) fn bind_catalog_aliases(&mut self, catalog: &Catalog) -> Result<()> { - self.inner.bind_catalog_aliases(catalog) + pub(crate) fn bind_catalog_aliases( + &mut self, + catalog: &Catalog, + live: &Snapshot, + ) -> Result<()> { + self.inner.bind_catalog_aliases(catalog, &live.inner) + } + + /// Whether the accepted contract was published with the version this view + /// pins for `type_key` (see `omnigraph_catalog::Snapshot::contract_covers`). + pub(crate) fn contract_covers(&self, type_key: &str) -> bool { + self.inner.contract_covers(type_key) } /// Open a backing dataset at its pinned version by qualified graph type key. With read caches present diff --git a/crates/omnigraph/src/engine/graph.rs b/crates/omnigraph/src/engine/graph.rs index 6d0079acb..62c9ad216 100644 --- a/crates/omnigraph/src/engine/graph.rs +++ b/crates/omnigraph/src/engine/graph.rs @@ -491,6 +491,7 @@ pub(super) async fn prepare_selected_edges( &dataset, &env.catalog.edge_types[&member.edge_type].arrow_schema, &table, + env.snapshot.contract_covers(&table), [ env.catalog.system_columns.src, env.catalog.system_columns.dst, @@ -672,6 +673,7 @@ pub(super) async fn decide_expand_start( &edge_ds, &catalog.edge_types[edge_type].arrow_schema, &edge_table_key, + snapshot.contract_covers(&edge_table_key), [catalog.system_columns.src, catalog.system_columns.dst], )?; let coverage = named @@ -870,18 +872,16 @@ where source_memory .checkpoint() .map_err(|error| memory.error(error))?; + let edge_table_key = format!("edge:{edge_type}"); let dataset = match prepared { Some(dataset) => dataset.clone(), - None => { - snapshot - .open_lance_dataset(&format!("edge:{edge_type}")) - .await? - } + None => snapshot.open_lance_dataset(&edge_table_key).await?, }; super::typed_value::check_stored_schema( &dataset, &catalog.edge_types[edge_type].arrow_schema, - &format!("edge:{edge_type}"), + &edge_table_key, + snapshot.contract_covers(&edge_table_key), [catalog.system_columns.src, catalog.system_columns.dst] .into_iter() .chain(attach_columns.iter().copied()), diff --git a/crates/omnigraph/src/engine/operators/hydrate.rs b/crates/omnigraph/src/engine/operators/hydrate.rs index 95e57f7d5..01d736906 100644 --- a/crates/omnigraph/src/engine/operators/hydrate.rs +++ b/crates/omnigraph/src/engine/operators/hydrate.rs @@ -360,6 +360,7 @@ async fn hydrate( &dataset, declared, &binding.table.type_key, + plan.snapshot.contract_covers(&binding.table.type_key), properties.iter().copied(), ) .map_err(external)?; diff --git a/crates/omnigraph/src/engine/operators/scan/input.rs b/crates/omnigraph/src/engine/operators/scan/input.rs index 9fbbabc14..397f1c7ae 100644 --- a/crates/omnigraph/src/engine/operators/scan/input.rs +++ b/crates/omnigraph/src/engine/operators/scan/input.rs @@ -251,16 +251,24 @@ async fn hydrate_nodes( let table_key = format!("node:{}", type_name); let ds = snapshot.open_lance_dataset(&table_key).await?; + let contract_covers = snapshot.contract_covers(&table_key); crate::engine::typed_value::check_stored_schema( &ds, &node_type.arrow_schema, &table_key, + contract_covers, columns .stored_columns() .into_iter() .chain([catalog.system_columns.id]), )?; - crate::engine::typed_value::check_scan_leaves(&ds, dst_filters)?; + crate::engine::typed_value::check_scan_leaves( + &ds, + &node_type.arrow_schema, + &table_key, + contract_covers, + dst_filters, + )?; memory .entries::(ids.len()) diff --git a/crates/omnigraph/src/engine/plan_source/scan_access.rs b/crates/omnigraph/src/engine/plan_source/scan_access.rs index 32896ea6d..2ce09f84e 100644 --- a/crates/omnigraph/src/engine/plan_source/scan_access.rs +++ b/crates/omnigraph/src/engine/plan_source/scan_access.rs @@ -41,6 +41,7 @@ pub(super) async fn index_split(source: &QuerySource<'_>, spec: &ScanSpec) -> Re let read = NodeRead::for_index_split( dataset.clone(), type_name, + source.snapshot.contract_covers(&spec.table.type_key), &filters, source.params.shared(), source.catalog, diff --git a/crates/omnigraph/src/engine/plan_source/scan_access/planning_cost.rs b/crates/omnigraph/src/engine/plan_source/scan_access/planning_cost.rs index b7452aaa9..6819f6b2d 100644 --- a/crates/omnigraph/src/engine/plan_source/scan_access/planning_cost.rs +++ b/crates/omnigraph/src/engine/plan_source/scan_access/planning_cost.rs @@ -392,7 +392,8 @@ async fn measure( )]; let params = ParamMap::new(); let read = - NodeRead::for_index_split(ds.clone(), "Person", &filters, ¶ms, catalog, None).unwrap(); + NodeRead::for_index_split(ds.clone(), "Person", true, &filters, ¶ms, catalog, None) + .unwrap(); let (plan, filter) = read.plan_with_filter().await.unwrap(); if finalized { assert!(matches!( diff --git a/crates/omnigraph/src/engine/scan.rs b/crates/omnigraph/src/engine/scan.rs index 06142e16e..217e3d56e 100644 --- a/crates/omnigraph/src/engine/scan.rs +++ b/crates/omnigraph/src/engine/scan.rs @@ -61,6 +61,7 @@ impl<'n> NodeRead<'n> { ds: &Dataset, node_type: &omnigraph_compiler::catalog::NodeType, table_key: &str, + contract_covers: bool, filters: &[IRExpr], catalog: &Catalog, search_mode: &SearchMode, @@ -70,6 +71,7 @@ impl<'n> NodeRead<'n> { ds, &node_type.arrow_schema, table_key, + contract_covers, read_columns .stored_columns() .into_iter() @@ -87,13 +89,20 @@ impl<'n> NodeRead<'n> { .map(|target| target.property.as_str()), ), )?; - super::typed_value::check_scan_leaves(ds, filters)?; + super::typed_value::check_scan_leaves( + ds, + &node_type.arrow_schema, + table_key, + contract_covers, + filters, + )?; Ok(()) } pub(super) fn for_index_split( ds: Dataset, type_name: &str, + contract_covers: bool, filters: &[IRExpr], params: &ParamMap, catalog: &'n Catalog, @@ -110,6 +119,7 @@ impl<'n> NodeRead<'n> { &ds, node_type, &format!("node:{type_name}"), + contract_covers, filters, catalog, &SearchMode::default(), @@ -145,6 +155,7 @@ impl<'n> NodeRead<'n> { &ds, node_type, &table_key, + snapshot.contract_covers(&table_key), filters, catalog, search_mode, diff --git a/crates/omnigraph/src/engine/typed_value.rs b/crates/omnigraph/src/engine/typed_value.rs index 7d97bff4c..f5ddea3a1 100644 --- a/crates/omnigraph/src/engine/typed_value.rs +++ b/crates/omnigraph/src/engine/typed_value.rs @@ -7,6 +7,11 @@ use datafusion::scalar::ScalarValue; use omnigraph_compiler::ir::IRParam; use omnigraph_compiler::types::ExprType; +use crate::error::{ + missing_graph_property_at_snapshot, renamed_graph_property_at_snapshot, + unwitnessed_graph_property_at_snapshot, +}; + /// Fill omitted nullable declarations and refuse omitted required ones. /// Bound values, including the invocation's sampled clock, stay unchanged. pub(crate) fn fill_declared_params(params: &mut ParamMap, declared: &[IRParam]) -> Result<()> { @@ -380,33 +385,33 @@ pub(super) fn check_output_schema(actual: &Schema, declared: &Schema, hidden: bo /// Check the schema of the dataset that was actually opened at the pinned /// version, for every consumed physical column. A historical dataset may omit -/// unrelated newer declarations. Consumed fields must match type and nullability. +/// unrelated newer declarations. Consumed fields must match type and nullability, +/// and a consumed property must be the declared property lifetime +/// ([`stored_field`]); `contract_covers` is the view's +/// `Snapshot::contract_covers` for `table`. pub(super) fn check_stored_schema<'a>( dataset: &Dataset, declared: &Schema, table: &str, + contract_covers: bool, consumed: impl IntoIterator, ) -> Result<()> { let stored = Schema::from(dataset.schema()); - check_stored_fields(&stored, declared, table, consumed) + check_stored_fields(&stored, declared, table, contract_covers, consumed) } fn check_stored_fields<'a>( stored: &Schema, declared: &Schema, table: &str, + contract_covers: bool, consumed: impl IntoIterator, ) -> Result<()> { for name in consumed { let expected = declared.field_with_name(name).map_err(|_| { OmniError::manifest_internal(format!("consumed {table}.{name} has no declared field")) })?; - let actual = stored.field_with_name(expected.name()).map_err(|_| { - OmniError::manifest_internal(format!( - "stored {table} has no declared field '{}'", - expected.name() - )) - })?; + let actual = stored_field(stored, expected, table, contract_covers)?; if actual.data_type() != expected.data_type() || actual.is_nullable() != expected.is_nullable() { @@ -423,9 +428,72 @@ fn check_stored_fields<'a>( Ok(()) } +/// The stored field a declared column reads in the opened table version. +/// +/// A property is its persisted stable property id, never its name. Where the +/// accepted contract was published with the opened version, its names name +/// that version's columns, and a column that is not the declared lifetime is +/// an integrity failure. A historical version is read under the current +/// contract, so the column of the declared name may be a dropped property's +/// (a same-name re-add) and the declared property may sit under an earlier +/// name (a rename since) or nowhere (an addition since): each refuses like a +/// type absent at the snapshot. A column written before 0.10 carries no id +/// and proves no lifetime, so a historical version refuses it, as Blob reads +/// do. System columns carry no id and resolve by name. +fn stored_field<'s>( + stored: &'s Schema, + expected: &Field, + table: &str, + contract_covers: bool, +) -> Result<&'s Field> { + let name = expected.name(); + let internal = + || OmniError::manifest_internal(format!("stored {table} has no declared field '{name}'")); + let Some(declared_id) = expected + .metadata() + .get(crate::db::STABLE_PROPERTY_ID_METADATA_KEY) + else { + return stored.field_with_name(name).map_err(|_| internal()); + }; + if let Ok(actual) = stored.field_with_name(name) { + match actual + .metadata() + .get(crate::db::STABLE_PROPERTY_ID_METADATA_KEY) + { + Some(stored_id) if stored_id == declared_id => return Ok(actual), + None if contract_covers => return Ok(actual), + None => { + return Err(OmniError::manifest(unwitnessed_graph_property_at_snapshot( + table, name, + ))); + } + Some(_) => {} + } + } + if contract_covers { + return Err(internal()); + } + let renamed = stored.fields().iter().find(|field| { + field + .metadata() + .get(crate::db::STABLE_PROPERTY_ID_METADATA_KEY) + == Some(declared_id) + }); + Err(OmniError::manifest(match renamed { + Some(field) => renamed_graph_property_at_snapshot(table, name, field.name()), + None => missing_graph_property_at_snapshot(table, name), + })) +} + /// A pushed filter consumes physical columns directly, so validate its stored /// leaf declarations against the opened dataset before pushdown can hide them. -pub(super) fn check_scan_leaves(dataset: &Dataset, filters: &[IRExpr]) -> Result<()> { +pub(super) fn check_scan_leaves( + dataset: &Dataset, + declared: &Schema, + table: &str, + contract_covers: bool, + filters: &[IRExpr], +) -> Result<()> { let stored = Schema::from(dataset.schema()); for filter in filters { let mut result = Ok(()); @@ -437,9 +505,10 @@ pub(super) fn check_scan_leaves(dataset: &Dataset, filters: &[IRExpr]) -> Result return; }; result = (|| { - let actual = stored + let expected = declared .field_with_name(property) .map_err(OmniError::arrow_internal)?; + let actual = stored_field(&stored, expected, table, contract_covers)?; let ExprType::Value { nullable, .. } = ty else { return Err(OmniError::manifest_internal( "pushed property carries a node type", @@ -839,6 +908,7 @@ pub(super) fn validate_plan_values( #[cfg(test)] mod tests { use super::*; + use crate::error::ManifestErrorKind; use omnigraph_compiler::query::ast::Param; use omnigraph_compiler::types::PropType; @@ -1019,12 +1089,82 @@ mod tests { Field::new("title", DataType::Utf8, false), Field::new("legacy", DataType::Utf8, false), ]); - assert!(check_stored_fields(&stored, &declared, "node:Doc", ["title"]).is_ok()); + assert!(check_stored_fields(&stored, &declared, "node:Doc", false, ["title"]).is_ok()); for consumed in ["rating", "legacy", "unknown"] { - assert!(check_stored_fields(&stored, &declared, "node:Doc", [consumed]).is_err()); + assert!( + check_stored_fields(&stored, &declared, "node:Doc", false, [consumed]).is_err() + ); } let widened = Schema::new(vec![Field::new("title", DataType::Utf8, true)]); - assert!(check_stored_fields(&widened, &declared, "node:Doc", ["title"]).is_err()); + assert!(check_stored_fields(&widened, &declared, "node:Doc", false, ["title"]).is_err()); + } + + /// A property resolves by its stable property id. A version the accepted + /// contract was published with must hold every declared lifetime under its + /// name; a historical version may not, and refuses each mismatch typed. + #[test] + fn historical_schema_binds_properties_by_stable_property_id() { + let marked = |name: &str, id: Option<&str>| { + let field = Field::new(name, DataType::Utf8, true); + match id { + Some(id) => field.with_metadata(HashMap::from([( + crate::db::STABLE_PROPERTY_ID_METADATA_KEY.to_string(), + id.to_string(), + )])), + None => field, + } + }; + let declared = Schema::new(vec![ + Field::new("__id", DataType::Utf8, false), + marked("title", Some("1")), + marked("note", Some("7")), + marked("remark", Some("5")), + marked("age", Some("9")), + ]); + let stored = Schema::new(vec![ + Field::new("__id", DataType::Utf8, false), + marked("title", Some("1")), + marked("note", Some("3")), + marked("memo", Some("5")), + ]); + let check = |stored: &Schema, covers: bool, column: &str| { + check_stored_fields(stored, &declared, "node:Doc", covers, [column]) + }; + let refusal = |stored: &Schema, column: &str| match check(stored, false, column) { + Err(OmniError::Manifest(error)) if error.kind == ManifestErrorKind::BadRequest => { + error.message + } + other => panic!("historical {column} must refuse as BadRequest; got {other:?}"), + }; + for covers in [true, false] { + assert!(check(&stored, covers, "__id").is_ok()); + assert!(check(&stored, covers, "title").is_ok()); + } + assert_eq!( + refusal(&stored, "note"), + "property 'note' of node type 'Doc' does not exist at this snapshot" + ); + assert_eq!( + refusal(&stored, "remark"), + "property 'remark' of node type 'Doc' is named 'memo' at this snapshot; a historical query cannot read a property across a rename" + ); + assert_eq!( + refusal(&stored, "age"), + "property 'age' of node type 'Doc' does not exist at this snapshot" + ); + for column in ["note", "remark", "age"] { + assert!(matches!( + check(&stored, true, column), + Err(OmniError::Manifest(ref error)) if error.kind == ManifestErrorKind::Internal + )); + } + + let unmarked = Schema::new(vec![marked("note", None)]); + assert!(check(&unmarked, true, "note").is_ok()); + assert_eq!( + refusal(&unmarked, "note"), + "property 'note' of node type 'Doc' has no persisted property-lifetime witness at this snapshot" + ); } #[test] @@ -1032,9 +1172,14 @@ mod tests { let declared = Schema::new(vec![Field::new("amount", DataType::Int32, false)]); let widened = Schema::new(vec![Field::new("amount", DataType::Int32, true)]); let changed = Schema::new(vec![Field::new("amount", DataType::Int64, false)]); - assert!(check_stored_fields(&declared, &declared, "node:Account", ["amount"]).is_ok()); - assert!(check_stored_fields(&widened, &declared, "node:Account", ["amount"]).is_err()); - assert!(check_stored_fields(&changed, &declared, "node:Account", ["amount"]).is_err()); + for covers in [true, false] { + let check = |stored: &Schema| { + check_stored_fields(stored, &declared, "node:Account", covers, ["amount"]) + }; + assert!(check(&declared).is_ok()); + assert!(check(&widened).is_err()); + assert!(check(&changed).is_err()); + } assert!(check_output_schema(&widened, &declared, false).is_ok()); assert!(check_output_schema(&changed, &declared, false).is_err()); let nulls: ArrayRef = Arc::new(arrow_array::Int32Array::from(vec![None])); diff --git a/crates/omnigraph/tests/integration/point_in_time.rs b/crates/omnigraph/tests/integration/point_in_time.rs index e1b7c6e4e..6f088bd5c 100644 --- a/crates/omnigraph/tests/integration/point_in_time.rs +++ b/crates/omnigraph/tests/integration/point_in_time.rs @@ -3,6 +3,7 @@ use crate::helpers; use crate::helpers::*; use arrow_array::{Array, Int32Array}; use omnigraph::db::{Omnigraph, ReadTarget}; +use omnigraph::error::{ManifestErrorKind, OmniError}; use omnigraph::loader::LoadMode; use omnigraph_compiler::ir::ParamMap; @@ -397,6 +398,153 @@ async fn historical_query_resolves_rename_by_identity_but_rejects_reincarnation( ); } +/// A historical query binds a current property by its stable property id, as +/// it binds a type by its table identity. A same-name re-add is a new property +/// lifetime and must not read the dropped one's values; a rename since and an +/// addition since have no column of the current lifetime under the current +/// name. Each refuses as BadRequest instead of answering from another lifetime. +#[tokio::test] +async fn historical_query_binds_properties_by_identity_not_name() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap(); + let db = session( + Omnigraph::init( + uri, + "node Person {\n name: String @key\n note: String?\n memo: String?\n}\n\ + edge Knows: Person -> Person {\n since: I32?\n}\n", + ) + .await + .unwrap(), + ); + db.load( + "main", + concat!( + r#"{"type":"Person","data":{"name":"Alice","note":"secret","memo":"kept"}}"#, + "\n", + r#"{"type":"Person","data":{"name":"Bob"}}"#, + "\n", + r#"{"edge":"Knows","from":"Alice","to":"Bob","data":{"since":2020}}"#, + ), + LoadMode::Merge, + ) + .await + .unwrap(); + let before = version_main(&db).await.unwrap(); + let before_snapshot = db.resolve_snapshot("main").await.unwrap(); + + db.apply_schema( + "node Person {\n name: String @key\n note: String?\n remark: String? @rename_from(\"memo\")\n}\n\ + edge Knows: Person -> Person {\n since: I32?\n}\n", + ) + .await + .unwrap(); + db.apply_schema( + "node Person {\n name: String @key\n remark: String?\n}\n\ + edge Knows: Person -> Person\n", + ) + .await + .unwrap(); + db.apply_schema( + "node Person {\n name: String @key\n remark: String?\n note: String?\n age: I32?\n}\n\ + edge Knows: Person -> Person {\n since: I32?\n}\n", + ) + .await + .unwrap(); + + let note = "query q() { match { $p: Person { name: \"Alice\" } } return { $p.note } }"; + let live = query_main(&db, note, "q", &ParamMap::new()).await.unwrap(); + assert!( + collect_column_strings(live.batches(), "p.note").is_empty(), + "the re-added property starts empty on the live branch" + ); + + let missing = |property: &str, owner: &str| { + format!("property '{property}' of {owner} does not exist at this snapshot") + }; + for (source, expected) in [ + (note.to_string(), missing("note", "node type 'Person'")), + ( + "query q() { match { $p: Person\n $p.note = \"secret\" } return { $p.name } }" + .to_string(), + missing("note", "node type 'Person'"), + ), + ( + "query q() { match { $p: Person } return { $p.name, $p.remark } }".to_string(), + "property 'remark' of node type 'Person' is named 'memo' at this snapshot; \ + a historical query cannot read a property across a rename" + .to_string(), + ), + ( + "query q() { match { $p: Person } return { $p.name, $p.age } }".to_string(), + missing("age", "node type 'Person'"), + ), + ( + "query q() { match { $p: Person $p $k:knows $f } return { $f.name, $k.since } }" + .to_string(), + missing("since", "edge type 'Knows'"), + ), + ] { + let at_version = db + .run_query_at(before, &source, "q", &ParamMap::new()) + .await + .expect_err("a property lifetime absent at the snapshot must refuse"); + let at_snapshot = db + .query( + ReadTarget::Snapshot(before_snapshot.clone()), + &source, + "q", + &ParamMap::new(), + ) + .await + .expect_err("a property lifetime absent at the snapshot must refuse"); + for error in [at_version, at_snapshot] { + assert!( + matches!( + error, + OmniError::Manifest(ref error) + if error.kind == ManifestErrorKind::BadRequest + && error.message == expected + ), + "{source}: expected BadRequest {expected:?}; got {error:?}" + ); + } + } + + let whole_node = db + .run_query_at( + before, + "query q() { match { $p: Person } return { $p } }", + "q", + &ParamMap::new(), + ) + .await + .expect_err("a whole node reads every current property"); + assert!( + matches!( + whole_node, + OmniError::Manifest(ref error) + if error.kind == ManifestErrorKind::BadRequest + && error.message.contains(" at this snapshot") + ), + "{whole_node:?}" + ); + + let names = db + .run_query_at( + before, + "query q() { match { $p: Person } return { $p.name } order { $p.name asc } }", + "q", + &ParamMap::new(), + ) + .await + .unwrap(); + assert_eq!( + collect_column_strings(names.batches(), "p.name"), + vec!["Alice".to_string(), "Bob".to_string()], + "a property kept under its own name and lifetime still reads at the snapshot" + ); +} + #[tokio::test] async fn run_query_at_traversal_uses_historical_graph_index() { let dir = tempfile::tempdir().unwrap(); diff --git a/docs/dev/execution.md b/docs/dev/execution.md index e6533d684..b38e78b0b 100644 --- a/docs/dev/execution.md +++ b/docs/dev/execution.md @@ -34,6 +34,18 @@ The executor never refreshes a mutable branch head midway through a query. Historical reads build from the requested immutable table versions; current branch reads may reuse version-keyed derived state. +A historical read plans against the current accepted catalog. Its tables bind +by table identity (`Snapshot::bind_catalog_aliases`) and its properties by the +stable property id each physical column persists as +`omnigraph.stable_property_id` (`stored_field` in `engine/typed_value.rs`, +behind every scan, hydration and edge read). The binding records each table +whose pinned version differs from the live one (`Snapshot::contract_covers`); +only there can the column of a property's name hold another lifetime. On such +a table a property refuses when that column carries another id (a same-name +re-add), when the id sits under an earlier name (a rename since) or nowhere +(an addition since), and when the column carries no id (written before 0.10), +as Blob reads do. Crossing a rename is not supported. + Stable code owners: | Concern | Owner | diff --git a/docs/user/branching/index.md b/docs/user/branching/index.md index 4711789d0..48bcf5992 100644 --- a/docs/user/branching/index.md +++ b/docs/user/branching/index.md @@ -157,9 +157,21 @@ omnigraph query sources_for_claim --query queries.gq \ A query stays on one snapshot for its entire lifetime. Wildcard traversal is refused on an explicit historical target because historical schema membership is not reconstructed. Explicit edge alternatives retain named-edge historical -read rules. Historical reads can -eventually fail after destructive cleanup removes the versions that commit -needs. Branch deletion can likewise end access to branch-only history. +read rules. + +A historical query is written against the current schema. A type renamed since +the commit still reads under its current name, but a property reads only where +the commit stores that same property under its current name. A property added +since the commit, renamed since, or dropped and re-added with the same name is +refused at that commit, naming the property, rather than answering with null or +with the dropped property's values. A whole-node `return { $p }` reads every +current property, so it is refused at commits made before a property was added. +Table data written before v0.10 records no property identity, so a commit that +reads such data refuses its properties too, unless that data is still the +table's current version. + +Historical reads can eventually fail after destructive cleanup removes the +versions that commit needs. Branch deletion can likewise end access to branch-only history. ## Changes and feeds diff --git a/docs/user/schema/index.md b/docs/user/schema/index.md index 90a17ac3c..62fda9669 100644 --- a/docs/user/schema/index.md +++ b/docs/user/schema/index.md @@ -174,7 +174,9 @@ rejected. The plan reports the exact unsupported step before anything changes. A drop removes the declaration from the current schema and reclaims no storage at apply. Older commits still read the dropped data until `omnigraph cleanup` -stops retaining them. A dropped type's data is reclaimed by that cleanup. A +stops retaining them. Re-adding a property under a dropped name starts a new +property, so a historical query never reads the dropped values through it; see +[historical reads](../branching/index.md#historical-reads). A dropped type's data is reclaimed by that cleanup. A dropped property's values also stay in the table's current data files, beside the properties that remain, until the next `omnigraph optimize`: every optimize rewrites each fragment of a table that still stores a dropped