fix(engine): historical queries bind properties by stable identity

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

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<SnapshotReadCaches>,
+    /// 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<String>,
 }
 
 #[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::<Vec<_>>();
 
+        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(),
         }
     }
 

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)]

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, &current).unwrap();
+    assert!(snapshot.contract_covers("edge:Link"));
 }
 
 #[tokio::test]

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))
 }

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<Arc<Catalog>> {
+        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<Catalog>, 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<Catalog>)> {
         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))
     }
 

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

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()),

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)?;

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::<datafusion::prelude::Expr>(ids.len())

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,

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, &params, catalog, None).unwrap();
+        NodeRead::for_index_split(ds.clone(), "Person", true, &filters, &params, catalog, None)
+            .unwrap();
     let (plan, filter) = read.plan_with_filter().await.unwrap();
     if finalized {
         assert!(matches!(

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,