perf(engine): halve per-write __manifest scans

crates/omnigraph/Cargo.toml

@@ -55,6 +55,8 @@ arc-swap = { workspace = true }
 omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.7.2" }
 tokio = { workspace = true }
 lance-namespace-impls = { workspace = true }
-lance-io = "7.0.0"
+# test-util gates IoStats.requests + assert_io_eq! (failure diagnostics only); dev-dep,
+# so production builds (which exclude dev-deps) never see it.
+lance-io = { version = "7.0.0", features = ["test-util"] }
 serial_test = "3"
 proptest = "1"

crates/omnigraph/src/db/manifest.rs

@@ -579,13 +579,16 @@ impl ManifestCoordinator {
         let PublishOutcome {
             dataset,
             parent_commit_id,
+            known_state,
         } = self
             .publisher
             .publish(changes, expected_table_versions, lineage)
             .await?;
+        // RFC-013 PR2 #1b: the publisher folded the new visible state in-memory
+        // (byte-identical to a re-scan via the shared `assemble_manifest_state`),
+        // so adopt it directly instead of an O(fragments) `read_manifest_state`.
         self.dataset = dataset;
-
-        self.known_state = read_manifest_state(&self.dataset).await?;
+        self.known_state = known_state;
         Ok(CommitOutcome {
             version: self.version(),
             parent_commit_id,

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

@@ -32,11 +32,12 @@ use crate::error::{OmniError, Result};
 #[cfg(test)]
 use super::SubTableUpdate;
 use super::layout::{open_manifest_dataset, tombstone_object_id, version_object_id};
-use super::metadata::parse_namespace_version_request;
+use super::metadata::{TableVersionMetadata, parse_namespace_version_request};
 use super::migrations::migrate_internal_schema;
 use super::state::{
-    GraphLineageRow, GraphLineageRowPart, graph_lineage_row_parts, head_lineage_row,
-    manifest_rows_batch, manifest_schema, read_publish_scan,
+    GraphLineageRow, GraphLineageRowPart, ManifestState, assemble_manifest_state,
+    graph_lineage_row_parts, head_lineage_row, manifest_rows_batch, manifest_schema,
+    read_manifest_state, read_publish_scan,
 };
 use super::{
     ManifestChange, OBJECT_TYPE_TABLE, OBJECT_TYPE_TABLE_TOMBSTONE, OBJECT_TYPE_TABLE_VERSION,
@@ -82,6 +83,11 @@ pub(super) struct PublishOutcome {
     /// in-memory commit cache without a re-read. `None` when no lineage was
     /// recorded, or when the commit is the genesis (no parent).
     pub parent_commit_id: Option<String>,
+    /// The new visible per-table state, folded in-memory from the pre-publish
+    /// state ∪ the committed rows (RFC-013 PR2 #1b). Returned so the caller skips
+    /// the O(fragments) post-publish `read_manifest_state` re-scan. Byte-identical
+    /// to that re-scan: built through the same `assemble_manifest_state` reduction.
+    pub known_state: ManifestState,
 }
 
 #[async_trait]
@@ -451,6 +457,101 @@ impl GraphNamespacePublisher {
         latest
     }
 
+    /// Build the inputs for [`assemble_manifest_state`] from the pre-publish state
+    /// unioned with the pending rows about to be committed — the in-memory basis
+    /// for the post-publish `known_state` fold (RFC-013 PR2 #1b), so the caller
+    /// skips the O(fragments) re-scan. Mirrors `read_manifest_scan`'s row handling
+    /// exactly so the result is byte-identical: `table_path` resolves through
+    /// `table_locations` = `registered_tables` UNION the pending `OBJECT_TYPE_TABLE`
+    /// rows (a freshly-registered table is not yet in `registered_tables`);
+    /// `version_metadata` parses the SAME JSON string a re-scan would read. Pending
+    /// `OBJECT_TYPE_TABLE` rows feed only `table_locations`; lineage rows
+    /// (`graph_commit`/`graph_head`) are not manifest-state entries.
+    fn fold_inputs(
+        existing_versions: &HashMap<(String, u64), SubTableEntry>,
+        existing_tombstones: &HashMap<(String, u64), ()>,
+        rows: &[PendingVersionRow],
+        registered_tables: &HashMap<String, String>,
+    ) -> Result<(Vec<SubTableEntry>, Vec<(String, u64)>)> {
+        let mut table_locations: HashMap<String, String> = registered_tables.clone();
+        for row in rows {
+            if row.object_type == OBJECT_TYPE_TABLE {
+                if let Some(location) = &row.location {
+                    table_locations.insert(row.table_key.clone(), location.clone());
+                }
+            }
+        }
+
+        // Key version entries by `(table_key, table_version)` so a pending row at
+        // the SAME version REPLACES the pre-publish entry — modelling merge-insert
+        // `UpdateAll` on the shared, deterministic `version_object_id(table_key,
+        // version)`. Load-bearing for the owner-branch handoff
+        // (`is_owner_branch_handoff`): a handoff updates a `table_version` row in
+        // place at the same version with a new `table_branch`, so `__manifest` ends
+        // with ONE row carrying the new branch and a re-scan reflects it; appending
+        // the pending row instead (and letting `assemble_manifest_state` keep the
+        // first equal-version entry) would leave `known_state` on the stale fork.
+        let mut version_map: HashMap<(String, u64), SubTableEntry> = existing_versions.clone();
+        let mut tombstones: Vec<(String, u64)> = existing_tombstones
+            .keys()
+            .map(|(key, version)| (key.clone(), *version))
+            .collect();
+
+        for row in rows {
+            match row.object_type.as_str() {
+                OBJECT_TYPE_TABLE_VERSION => {
+                    let table_version = row.table_version.ok_or_else(|| {
+                        OmniError::manifest_internal(format!(
+                            "post-publish fold: table_version row missing version for {}",
+                            row.table_key
+                        ))
+                    })?;
+                    let table_path =
+                        table_locations.get(&row.table_key).cloned().ok_or_else(|| {
+                            OmniError::manifest_internal(format!(
+                                "post-publish fold: missing table row for {}",
+                                row.table_key
+                            ))
+                        })?;
+                    let metadata_json = row.metadata.as_deref().ok_or_else(|| {
+                        OmniError::manifest_internal(format!(
+                            "post-publish fold: table_version row missing metadata for {}",
+                            row.table_key
+                        ))
+                    })?;
+                    version_map.insert(
+                        (row.table_key.clone(), table_version),
+                        SubTableEntry {
+                            table_key: row.table_key.clone(),
+                            table_path,
+                            table_version,
+                            table_branch: row.table_branch.clone(),
+                            row_count: row.row_count.ok_or_else(|| {
+                                OmniError::manifest_internal(format!(
+                                    "post-publish fold: table_version row missing row_count for {}",
+                                    row.table_key
+                                ))
+                            })?,
+                            version_metadata: TableVersionMetadata::from_json_str(metadata_json)?,
+                        },
+                    );
+                }
+                OBJECT_TYPE_TABLE_TOMBSTONE => {
+                    let tombstone_version = row.table_version.ok_or_else(|| {
+                        OmniError::manifest_internal(format!(
+                            "post-publish fold: tombstone row missing version for {}",
+                            row.table_key
+                        ))
+                    })?;
+                    tombstones.push((row.table_key.clone(), tombstone_version));
+                }
+                _ => {}
+            }
+        }
+
+        Ok((version_map.into_values().collect(), tombstones))
+    }
+
     /// Compare each caller-supplied expectation against the manifest's current
     /// latest visible version per table. The first mismatch is returned as a
     /// typed `ExpectedVersionMismatch` (`actual = 0` if the table isn't in the
@@ -578,9 +679,15 @@ impl ManifestBatchPublisher for GraphNamespacePublisher {
         lineage: Option<&LineageIntent>,
     ) -> Result<PublishOutcome> {
         if changes.is_empty() && expected_table_versions.is_empty() && lineage.is_none() {
+            // Defensive no-op (never reached from `commit_changes_with_lineage`,
+            // which short-circuits the all-empty case): state is unchanged, so a
+            // re-scan here is acceptable.
+            let dataset = self.dataset().await?;
+            let known_state = read_manifest_state(&dataset).await?;
             return Ok(PublishOutcome {
-                dataset: self.dataset().await?,
+                dataset,
                 parent_commit_id: None,
+                known_state,
             });
         }
 
@@ -645,18 +752,39 @@ impl ManifestBatchPublisher for GraphNamespacePublisher {
 
             if rows.is_empty() {
                 // Expected-version-only publish with no changes and no lineage:
-                // the precondition held, nothing to write.
+                // the precondition held, nothing to write. Fold the unchanged state
+                // from the loaded maps — no re-scan (RFC-013 PR2 #1b).
+                let known_state = assemble_manifest_state(
+                    dataset.version().version,
+                    existing_versions.values().cloned().collect(),
+                    existing_tombstones
+                        .keys()
+                        .map(|(key, version)| (key.clone(), *version)),
+                );
                 return Ok(PublishOutcome {
                     dataset,
                     parent_commit_id,
+                    known_state,
                 });
             }
 
+            // Build the post-publish fold inputs from the pre-publish state ∪ the
+            // rows we are about to commit, BEFORE `rows` is moved into merge_rows
+            // (RFC-013 PR2 #1b). Recomputed per attempt from freshly-loaded state.
+            let (fold_entries, fold_tombstones) =
+                Self::fold_inputs(&existing_versions, &existing_tombstones, &rows, &known_tables)?;
+
             match self.merge_rows(dataset, rows).await {
                 Ok(new_dataset) => {
+                    let known_state = assemble_manifest_state(
+                        new_dataset.version().version,
+                        fold_entries,
+                        fold_tombstones,
+                    );
                     return Ok(PublishOutcome {
                         dataset: new_dataset,
                         parent_commit_id,
+                        known_state,
                     });
                 }
                 Err(err) => {

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

@@ -131,9 +131,30 @@ pub(super) async fn read_manifest_state(dataset: &Dataset) -> Result<ManifestSta
     let version = dataset.version().version;
     // The table-state hot path never needs lineage, so don't pay its JSON decode.
     let scan = read_manifest_scan(dataset, false).await?;
-    let mut latest_versions = HashMap::<String, SubTableEntry>::new();
+    Ok(assemble_manifest_state(
+        version,
+        scan.version_entries,
+        scan.tombstones
+            .into_iter()
+            .map(|t| (t.table_key, t.tombstone_version)),
+    ))
+}
 
-    for entry in scan.version_entries {
+/// Reduce raw manifest rows to the visible per-table state: keep the latest
+/// `table_version` per `table_key`, drop any whose latest version is sealed by a
+/// tombstone (`tombstone_version >= table_version`), then sort by `table_key` for
+/// deterministic output. Shared by the scan path (`read_manifest_state`) and the
+/// in-memory post-publish fold in the publisher (RFC-013 PR2 #1b), so the two
+/// CANNOT diverge in the dedup/filter/sort — the byte-identity the fold relies on.
+/// Tombstones are passed as `(table_key, tombstone_version)` tuples so callers
+/// outside this module need not name the private `TableTombstoneEntry`.
+pub(super) fn assemble_manifest_state(
+    version: u64,
+    version_entries: Vec<SubTableEntry>,
+    tombstones: impl IntoIterator<Item = (String, u64)>,
+) -> ManifestState {
+    let mut latest_versions = HashMap::<String, SubTableEntry>::new();
+    for entry in version_entries {
         match latest_versions.get(&entry.table_key) {
             Some(existing) if existing.table_version >= entry.table_version => {}
             _ => {
@@ -142,28 +163,27 @@ pub(super) async fn read_manifest_state(dataset: &Dataset) -> Result<ManifestSta
         }
     }
 
-    let mut tombstones = HashMap::<String, u64>::new();
-    for tombstone in scan.tombstones {
-        match tombstones.get(&tombstone.table_key) {
-            Some(existing) if *existing >= tombstone.tombstone_version => {}
+    let mut tombstone_map = HashMap::<String, u64>::new();
+    for (table_key, tombstone_version) in tombstones {
+        match tombstone_map.get(&table_key) {
+            Some(existing) if *existing >= tombstone_version => {}
             _ => {
-                tombstones.insert(tombstone.table_key, tombstone.tombstone_version);
+                tombstone_map.insert(table_key, tombstone_version);
             }
         }
     }
 
     let mut entries: Vec<SubTableEntry> = latest_versions
         .into_values()
         .filter(|entry| {
-            tombstones
+            tombstone_map
                 .get(&entry.table_key)
                 .map(|tombstone_version| *tombstone_version < entry.table_version)
                 .unwrap_or(true)
         })
         .collect();
     entries.sort_by(|a, b| a.table_key.cmp(&b.table_key));
-
-    Ok(ManifestState { version, entries })
+    ManifestState { version, entries }
 }
 
 // After RFC-013 P2 folded the publish path off this accessor (it now projects
@@ -245,8 +265,29 @@ fn decode_graph_commit_row(
 }
 
 async fn read_manifest_scan(dataset: &Dataset, collect_lineage: bool) -> Result<ManifestScan> {
-    let batches: Vec<RecordBatch> = dataset
-        .scan()
+    // Project only the columns the assembly below reads (RFC-013 PR2 #1c). The
+    // table-state hot path never touches `object_id` (lineage decode only) or
+    // `base_objects` (reserved/unused — never read on any path), so reading them
+    // is wasted bytes on every `__manifest` scan — write publish AND every
+    // branch-op open. Mirrors Lance's own directory-catalog `__manifest` reads,
+    // which project to the needed columns rather than scanning all of them.
+    let mut projection: Vec<&str> = vec![
+        "object_type",
+        "location",
+        "metadata",
+        "table_key",
+        "table_version",
+        "table_branch",
+        "row_count",
+    ];
+    if collect_lineage {
+        projection.push("object_id");
+    }
+    let mut scanner = dataset.scan();
+    scanner
+        .project(&projection)
+        .map_err(|e| OmniError::Lance(e.to_string()))?;
+    let batches: Vec<RecordBatch> = scanner
         .try_into_stream()
         .await
         .map_err(|e| OmniError::Lance(e.to_string()))?