feat(cdc): define graph change feed and land C0 foundation

crates/omnigraph/src/changes/mod.rs

@@ -1,11 +1,11 @@
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashSet};
 
 use arrow_array::{Array, RecordBatch, StringArray, UInt64Array};
 use arrow_cast::display::array_value_to_string;
 use lance::dataset::scanner::ColumnOrdering;
 
 use crate::db::SubTableEntry;
-use crate::db::manifest::Snapshot;
+use crate::db::manifest::{Snapshot, TableIdentity};
 use crate::error::Result;
 use crate::storage_layer::{SnapshotHandle, TableStorage};
 use crate::table_store::TableStore;
@@ -104,6 +104,45 @@ impl ChangeFilter {
 
 // ─── Core diff ──────────────────────────────────────────────────────────────
 
+/// One immutable table lifetime whose physical state differs between two
+/// graph snapshots.
+///
+/// Identity, not alias, pairs the endpoints. A rename therefore stays one
+/// interval (and is elided when its physical state did not move), while a
+/// drop/re-add under the same public name remains two distinct lifetimes.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct TableChangeInterval<'a> {
+    pub(crate) identity: TableIdentity,
+    pub(crate) from: Option<&'a SubTableEntry>,
+    pub(crate) to: Option<&'a SubTableEntry>,
+}
+
+/// Derive changed table lifetimes in stable immutable-identity order.
+///
+/// This is the graph-commit CDC pruning layer: later row enumeration only
+/// needs to inspect these exact endpoint pairs. It persists no parallel change
+/// log and does not infer identity from an alias, path, or Lance version.
+pub(crate) fn changed_table_intervals<'a>(
+    from: &'a Snapshot,
+    to: &'a Snapshot,
+) -> Vec<TableChangeInterval<'a>> {
+    let mut by_identity =
+        BTreeMap::<TableIdentity, (Option<&'a SubTableEntry>, Option<&'a SubTableEntry>)>::new();
+    for entry in from.entries() {
+        by_identity.entry(entry.identity).or_default().0 = Some(entry);
+    }
+    for entry in to.entries() {
+        by_identity.entry(entry.identity).or_default().1 = Some(entry);
+    }
+
+    by_identity
+        .into_iter()
+        .filter_map(|(identity, (from, to))| {
+            (!same_state(from, to)).then_some(TableChangeInterval { identity, from, to })
+        })
+        .collect()
+}
+
 /// Net-current diff between two snapshots.
 ///
 /// Uses a three-level algorithm:
@@ -117,40 +156,28 @@ pub(crate) async fn diff_snapshots(
     filter: &ChangeFilter,
     branch: Option<String>,
 ) -> Result<ChangeSet> {
-    let from_by_identity = from
-        .entries()
-        .map(|entry| (entry.identity, entry))
-        .collect::<HashMap<_, _>>();
-    let to_by_identity = to
-        .entries()
-        .map(|entry| (entry.identity, entry))
-        .collect::<HashMap<_, _>>();
-    let all_identities = from_by_identity
-        .keys()
-        .chain(to_by_identity.keys())
-        .copied()
-        .collect::<HashSet<_>>();
-
     let mut changes = Vec::new();
 
-    for identity in all_identities {
-        let from_entry = from_by_identity.get(&identity).copied();
-        let to_entry = to_by_identity.get(&identity).copied();
+    for interval in changed_table_intervals(from, to) {
+        let from_entry = interval.from;
+        let to_entry = interval.to;
         // Prefer the destination alias for a rename; a removed table has only
         // its source alias. Logical pairing never depends on either name.
         let table_key = &to_entry
             .or(from_entry)
             .expect("identity came from one snapshot")
             .table_key;
+        debug_assert!(
+            from_entry
+                .into_iter()
+                .chain(to_entry)
+                .all(|entry| entry.identity == interval.identity),
+            "table interval endpoints must retain their immutable identity"
+        );
         if !filter.matches_table(table_key) {
             continue;
         }
 
-        // Skip if both snapshots have identical state for this table
-        if same_state(from_entry, to_entry) {
-            continue;
-        }
-
         let (kind, type_name) = parse_table_key(table_key);
         let is_edge = kind == EntityKind::Edge;
 

crates/omnigraph/src/db/commit_graph.rs

@@ -31,6 +31,18 @@ impl GraphCommit {
     }
 }
 
+/// One observable graph transition on a branch's first-parent lineage.
+///
+/// CDC compares exactly these two immutable graph commits. A merge commit is
+/// compared with the branch state it landed on (`parent`); its
+/// `merged_parent_commit_id` remains provenance and is never traversed as a
+/// second feed path.
+#[derive(Debug, Clone)]
+pub(crate) struct FirstParentEdge {
+    pub(crate) parent: GraphCommit,
+    pub(crate) child: GraphCommit,
+}
+
 /// A pure projection of the graph lineage that lives in `__manifest`
 /// (`graph_commit` + `graph_head` rows, RFC-013 Phase 7). It opens NO Lance
 /// dataset (Phase B retired `_graph_commits.lance` / `_graph_commit_actors.lance`):

crates/omnigraph/src/db/graph_coordinator.rs

@@ -7,7 +7,7 @@ use crate::error::{OmniError, Result};
 use crate::failpoints;
 use crate::storage::{StorageAdapter, normalize_root_uri};
 
-use super::commit_graph::{CommitGraph, GraphCommit};
+use super::commit_graph::{CommitGraph, FirstParentEdge, GraphCommit};
 use super::is_internal_system_branch;
 use super::manifest::{
     CapturedManifestProbe, ExpectedTableVersions, LineageIntent, ManifestChange,
@@ -83,6 +83,14 @@ pub struct ResolvedTarget {
     pub snapshot: Snapshot,
 }
 
+/// Internal lineage classification for an existing two-commit diff request.
+/// Arbitrary ranges retain net-current semantics; direct adjacency is derived
+/// only from the child's persisted first-parent pointer.
+pub(crate) enum ResolvedCommitRange {
+    FirstParent(FirstParentEdge),
+    Arbitrary { from: GraphCommit, to: GraphCommit },
+}
+
 #[derive(Debug, Clone)]
 pub(crate) struct PublishedSnapshot {
     pub manifest_version: u64,
@@ -426,6 +434,28 @@ impl GraphCoordinator {
         )))
     }
 
+    /// Resolve both endpoints and classify direct first-parent adjacency from
+    /// the child's persisted parent pointer.
+    ///
+    /// This is deliberately O(1) after the two commits are resolved: it adds
+    /// no ancestry index or history walk. Arbitrary ranges retain the existing
+    /// net-current diff semantics.
+    pub(crate) async fn resolve_commit_range(
+        &self,
+        from_id: &SnapshotId,
+        to_id: &SnapshotId,
+    ) -> Result<ResolvedCommitRange> {
+        let from = self.resolve_commit(from_id).await?;
+        let to = self.resolve_commit(to_id).await?;
+        if to.parent_commit_id.as_deref() != Some(from.graph_commit_id.as_str()) {
+            return Ok(ResolvedCommitRange::Arbitrary { from, to });
+        }
+        Ok(ResolvedCommitRange::FirstParent(FirstParentEdge {
+            parent: from,
+            child: to,
+        }))
+    }
+
     pub(crate) async fn head_commit_id(&self) -> Result<Option<SnapshotId>> {
         self.commit_graph
             .head_commit_id()

crates/omnigraph/src/db/omnigraph.rs

@@ -25,7 +25,7 @@ use omnigraph_compiler::{
     plan_schema_migration,
 };
 
-use crate::db::graph_coordinator::{GraphCoordinator, PublishedSnapshot};
+use crate::db::graph_coordinator::{GraphCoordinator, PublishedSnapshot, ResolvedCommitRange};
 use crate::error::{OmniError, Result};
 use crate::runtime_cache::RuntimeCache;
 use crate::storage::{
@@ -2559,10 +2559,19 @@ impl Omnigraph {
         filter: &crate::changes::ChangeFilter,
     ) -> Result<crate::changes::ChangeSet> {
         let coord = self.coordinator.read().await;
-        let from_commit = coord
-            .resolve_commit(&SnapshotId::new(from_commit_id))
+        let range = coord
+            .resolve_commit_range(
+                &SnapshotId::new(from_commit_id),
+                &SnapshotId::new(to_commit_id),
+            )
             .await?;
-        let to_commit = coord.resolve_commit(&SnapshotId::new(to_commit_id)).await?;
+        // Classify direct adjacency from the child's persisted first-parent
+        // pointer without changing this API's net-current result shape. The
+        // future feed can reuse that relationship without an ancestry index.
+        let (from_commit, to_commit) = match range {
+            ResolvedCommitRange::FirstParent(edge) => (edge.parent, edge.child),
+            ResolvedCommitRange::Arbitrary { from, to } => (from, to),
+        };
         let from_snap = coord
             .resolve_target(&ReadTarget::Snapshot(SnapshotId::new(
                 from_commit.graph_commit_id.clone(),

crates/omnigraph/tests/forbidden_apis.rs

@@ -517,6 +517,11 @@ const LOW_LEVEL_READ_ONLY_SURFACES: &[(&str, &str, &str)] = &[
         "GraphCoordinator",
         "resolve_commit",
     ),
+    (
+        "db/graph_coordinator.rs",
+        "GraphCoordinator",
+        "resolve_commit_range",
+    ),
     (
         "db/graph_coordinator.rs",
         "GraphCoordinator",