engine: RFC 0067 table pins, pinned resolution and promotion reconciler

crates/omnigraph/src/blob.rs

@@ -1077,7 +1077,9 @@ impl Omnigraph {
                 .await?
         });
         let actual_table_version = dataset.version().version;
-        if actual_table_version != expected_table_version {
+        // RFC 0067: a pending pin resolves to its staged version.
+        let pending_staged = entry.version_metadata.staged_version() == Some(actual_table_version);
+        if actual_table_version != expected_table_version && !pending_staged {
             return Err(OmniError::blob_integrity(format!(
                 "selected dataset for {} type '{}' opened at Lance version {}, expected published dataset version {}",
                 entity_label(cell.entity),
@@ -1124,9 +1126,7 @@ impl Omnigraph {
                 )));
             }
         }
-        if let Some(expected_e_tag) = entry.version_metadata.e_tag()
-            && dataset.manifest_location().e_tag.as_deref() != Some(expected_e_tag)
-        {
+        if !entry.version_metadata.witnesses(&dataset) {
             return Err(OmniError::blob_integrity(format!(
                 "selected dataset for {} type '{}' opened a different Lance manifest incarnation at published dataset version {}",
                 entity_label(cell.entity),

crates/omnigraph/src/changes/mod.rs

@@ -236,7 +236,24 @@ pub(crate) async fn diff_snapshots(
             (Some(from), None) => diff_table_removed(table_store, from, is_edge, filter).await?,
             // Fast path: version-column diff
             (Some(from), Some(to)) if same_lineage(from_entry, to_entry) => {
-                diff_table_same_lineage(table_store, from, to, is_edge, filter).await?
+                match diff_table_same_lineage(table_store, from, to, is_edge, filter).await? {
+                    Some(changes) => changes,
+                    // RFC 0067: the `to` endpoint is a pending pin's detached
+                    // version. Rows staged on a detached predecessor carry
+                    // stamps outside the linear interval, so compare exactly.
+                    None => {
+                        diff_table_cross_branch(
+                            table_store,
+                            from,
+                            to,
+                            is_edge,
+                            filter,
+                            type_name,
+                            to_commit_id.as_deref().unwrap_or_default(),
+                        )
+                        .await?
+                    }
+                }
             }
             // Cross-branch path: streaming ID-based diff
             (Some(from), Some(to)) => {
@@ -321,11 +338,14 @@ async fn diff_table_same_lineage(
     to_entry: &DatasetEntry,
     is_edge: bool,
     filter: &ChangeFilter,
-) -> Result<Vec<EntityChange>> {
+) -> Result<Option<Vec<EntityChange>>> {
     let vf = from_entry.published_dataset_version;
     let vt = to_entry.published_dataset_version;
     let storage: &dyn TableStorage = table_store;
     let to_ds = storage.open_snapshot_at_entry(to_entry).await?;
+    if TableStore::is_detached_version(to_ds.dataset().version().version) {
+        return Ok(None);
+    }
     let to_columns = system_columns_at_image(to_ds.dataset().schema(), &to_entry.type_key)?;
 
     let cols: Vec<&str> = if is_edge {
@@ -364,7 +384,7 @@ async fn diff_table_same_lineage(
         Vec::new()
     };
     if changed_rows.is_empty() && !wants_deletes {
-        return Ok(changes);
+        return Ok(Some(changes));
     }
     let from_ds = storage.open_snapshot_at_entry(from_entry).await?;
     let from_columns = system_columns_at_image(from_ds.dataset().schema(), &from_entry.type_key)?;
@@ -392,7 +412,7 @@ async fn diff_table_same_lineage(
         );
     }
 
-    Ok(changes)
+    Ok(Some(changes))
 }
 
 // ─── Cross-branch path: streaming ID-based diff ────────────────────────────

crates/omnigraph/src/db/manifest.rs

@@ -578,6 +578,8 @@ impl Snapshot {
                         entry.native_dataset_branch.as_deref(),
                         entry.published_dataset_version,
                         entry.version_metadata.e_tag(),
+                        entry.version_metadata.staged_version(),
+                        entry.version_metadata.transaction_uuid(),
                         &location,
                         Some(&caches.session),
                     )
@@ -887,9 +889,11 @@ impl DatasetEntry {
         // cached path (`Snapshot::open_lance_dataset` → handle cache) calls the same opener on
         // a miss with the shared session, so both paths count on the per-query
         // `table_wrapper`.
-        crate::instrumentation::open_dataset(
+        crate::instrumentation::open_pinned_dataset(
             &location,
-            crate::instrumentation::VersionResolution::At(self.published_dataset_version),
+            self.published_dataset_version,
+            self.version_metadata.staged_version(),
+            self.version_metadata.transaction_uuid(),
             session,
             crate::instrumentation::table_wrapper(),
         )

crates/omnigraph/src/db/manifest/metadata.rs

@@ -15,6 +15,9 @@ use super::layout::table_id_to_key;
 pub(super) const OMNIGRAPH_ROW_COUNT_KEY: &str = "omnigraph.row_count";
 const OMNIGRAPH_TABLE_BRANCH_KEY: &str = "omnigraph.table_branch";
 const OMNIGRAPH_TABLE_FORK_OWNER_KEY: &str = "omnigraph.table_fork_owner";
+/// RFC 0067 prototype pin fields, carried like the fork owner.
+const OMNIGRAPH_STAGED_VERSION_KEY: &str = "omnigraph.staged_version";
+const OMNIGRAPH_TRANSACTION_UUID_KEY: &str = "omnigraph.transaction_uuid";
 
 pub(super) fn namespace_version_metadata(
     row_count: u64,
@@ -56,6 +59,10 @@ pub(super) fn parse_namespace_version_request(
         e_tag: request.e_tag.clone(),
         naming_scheme: request.naming_scheme.clone(),
         table_fork_owner: metadata.get(OMNIGRAPH_TABLE_FORK_OWNER_KEY).cloned(),
+        staged_version: metadata
+            .get(OMNIGRAPH_STAGED_VERSION_KEY)
+            .and_then(|value| value.parse::<u64>().ok()),
+        transaction_uuid: metadata.get(OMNIGRAPH_TRANSACTION_UUID_KEY).cloned(),
     };
 
     Ok((
@@ -75,6 +82,14 @@ pub(crate) struct TableVersionMetadata {
     naming_scheme: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     table_fork_owner: Option<String>,
+    /// RFC 0067: the detached Lance version this pin was staged
+    /// as before publication. `None` on pins published by the sidecar path.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    staged_version: Option<u64>,
+    /// RFC 0067: the uuid of the staged transaction; promotion
+    /// replays it at the linear target and readers verify it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    transaction_uuid: Option<String>,
 }
 
 impl TableVersionMetadata {
@@ -93,9 +108,38 @@ impl TableVersionMetadata {
             e_tag: dataset.manifest_location().e_tag.clone(),
             naming_scheme: Some(format!("{:?}", dataset.manifest_location().naming_scheme)),
             table_fork_owner: None,
+            staged_version: None,
+            transaction_uuid: None,
         })
     }
 
+    pub(crate) fn staged_version(&self) -> Option<u64> {
+        self.staged_version
+    }
+
+    pub(crate) fn transaction_uuid(&self) -> Option<&str> {
+        self.transaction_uuid.as_deref()
+    }
+
+    /// RFC 0067: whether `dataset` is the manifest this pin
+    /// witnesses. The e_tag identifies the exact manifest object; a pin that
+    /// carries a transaction uuid also accepts the promoted twin, whose
+    /// manifest differs but whose transaction is the same.
+    pub(crate) fn witnesses(&self, dataset: &Dataset) -> bool {
+        match self.e_tag.as_deref() {
+            None => return true,
+            Some(expected) if dataset.manifest_location().e_tag.as_deref() == Some(expected) => {
+                return true;
+            }
+            Some(_) => {}
+        }
+        match self.transaction_uuid.as_deref() {
+            Some(uuid) => crate::table_store::StagedTransactionIdentity::recorded_by(dataset)
+                .is_some_and(|identity| identity.uuid == uuid),
+            None => false,
+        }
+    }
+
     pub(crate) fn table_fork_owner(&self) -> Option<&str> {
         self.table_fork_owner.as_deref()
     }
@@ -159,6 +203,12 @@ impl TableVersionMetadata {
         if let Some(owner) = &self.table_fork_owner {
             metadata.insert(OMNIGRAPH_TABLE_FORK_OWNER_KEY.to_string(), owner.clone());
         }
+        if let Some(staged) = self.staged_version {
+            metadata.insert(OMNIGRAPH_STAGED_VERSION_KEY.to_string(), staged.to_string());
+        }
+        if let Some(uuid) = &self.transaction_uuid {
+            metadata.insert(OMNIGRAPH_TRANSACTION_UUID_KEY.to_string(), uuid.clone());
+        }
         request.metadata = Some(metadata);
         request
     }
@@ -377,10 +427,27 @@ mod tests {
     /// GQT does not expose the namespace registration request and response metadata.
     #[test]
     fn namespace_pointer_roundtrip_keeps_source_ownership() {
-        let source = TableVersionMetadata::from_json_str(LEGACY_JSON)
-            .unwrap()
-            .with_table_fork_owner(Some(OWNER));
+        let source = TableVersionMetadata {
+            staged_version: Some(9_223_372_036_854_775_815),
+            transaction_uuid: Some("8b8d7a1e-6f2c-4f0e-9d3a-1c2b3a4d5e6f".to_string()),
+            ..TableVersionMetadata::from_json_str(LEGACY_JSON)
+                .unwrap()
+                .with_table_fork_owner(Some(OWNER))
+        };
         let request = source.to_create_table_version_request("node:Person", 7, 3, Some(FORK));
+        let request_metadata = request.metadata.as_ref().unwrap();
+        assert_eq!(
+            request_metadata
+                .get(OMNIGRAPH_STAGED_VERSION_KEY)
+                .map(String::as_str),
+            Some("9223372036854775815")
+        );
+        assert_eq!(
+            request_metadata
+                .get(OMNIGRAPH_TRANSACTION_UUID_KEY)
+                .map(String::as_str),
+            Some("8b8d7a1e-6f2c-4f0e-9d3a-1c2b3a4d5e6f")
+        );
         assert_eq!(
             request
                 .metadata
@@ -397,6 +464,14 @@ mod tests {
         assert_eq!(rows, 3);
         assert_eq!(native_ref.as_deref(), Some(FORK));
         assert_eq!(target_registration, source);
+        assert_eq!(
+            target_registration.staged_version(),
+            Some(9_223_372_036_854_775_815)
+        );
+        assert_eq!(
+            target_registration.transaction_uuid(),
+            Some("8b8d7a1e-6f2c-4f0e-9d3a-1c2b3a4d5e6f")
+        );
         assert!(target_registration.is_table_fork_of(FORK, OWNER));
         assert!(!target_registration.is_table_fork_of(FORK, TARGET));
 

crates/omnigraph/src/db/omnigraph.rs

@@ -32,6 +32,7 @@ use crate::table_store::TableStore;
 
 mod export;
 pub(crate) mod optimize;
+pub(crate) mod promotion;
 mod repair;
 pub(crate) mod schema_apply;
 pub(crate) mod system_column_upgrade;
@@ -2281,6 +2282,22 @@ impl Omnigraph {
                         ),
                     ));
                 }
+                // RFC 0067: a promotion that landed after this writer captured
+                // its snapshot moves HEAD one past the published version the
+                // writer holds. When the current manifest explains that HEAD,
+                // the writer's read set is merely stale and it reprepares
+                // instead of being sent to repair.
+                let graph_branch = normalized_branch.map(crate::branch_names::logical_branch_name);
+                if let Ok(current) = self.fresh_snapshot_for_branch_unchecked(graph_branch).await
+                    && let Some(entry) = current.dataset(table_key)
+                    && entry.published_dataset_version != expected_version
+                {
+                    return Err(OmniError::manifest_read_set_changed(
+                        format!("published_dataset_version:{table_key}"),
+                        Some(expected_version.to_string()),
+                        Some(entry.published_dataset_version.to_string()),
+                    ));
+                }
                 Err(OmniError::manifest_conflict(format!(
                     "{} is at Lance HEAD version {}, ahead of published dataset version {}; \
                      run `omnigraph repair` before writing",