refactor: inline the schema contract in the catalog

.github/workflows/ci.yml

@@ -1279,7 +1279,7 @@ jobs:
           done
           run_cell azure_write_detached_pin_survives_next_write -p omnigraph-engine --features failpoints --test failpoints
           run_cell azure_optimize_detached_pin_survives_next_write -p omnigraph-engine --features failpoints --test failpoints
-          run_cell azure_schema_apply_recovers_source_and_destination_after_partial_rename -p omnigraph-engine --features failpoints --test failpoints
+          run_cell azure_schema_apply_post_commit_failure_leaves_readable_contract -p omnigraph-engine --features failpoints --test failpoints
           run_cell azure_cluster_full_lifecycle_import_apply_serve_evolve_delete -p omnigraph-cluster --test s3_cluster
           run_cell server_opens_azure_graph_directly_and_serves_snapshot_and_read -p omnigraph-server --test s3
           run_cell server_boots_azure_cluster_from_bare_storage_uri_and_serves_query -p omnigraph-server --test s3

AGENTS.md

@@ -104,8 +104,8 @@ working memory:
    fresh rather than mixing old and new authority.
 3. A mutation, load, schema apply, merge, or maintenance batch publishes once.
 4. Pre-publication durable effects stay unreachable (detached commits, staged
-   files); what is published carries the identity that finishes it (pins,
-   the staged contract's publishing commit); ambiguity fails closed.
+   files). Table pins and the accepted schema contract are complete in one
+   manifest publication; ambiguity fails closed.
 5. Stable schema identity survives supported renames, not drop/re-add. Never
    infer identity from names, paths, versions, field IDs, or branch refs.
 6. Indexes, caches, topology, fragment layout, and compaction are derived

changelog.d/schema-contract-inlining.breaking.md

@@ -0,0 +1,9 @@
+- Schema apply and system-column upgrades now publish the schema source,
+  compiled schema and identity atomically with their table references in the
+  catalog. Storage format 13 requires an explicit offline upgrade from supported
+  older formats; ordinary open refuses those layouts. Stop readers, writers and
+  maintenance, retain a whole-root backup, and follow the
+  [storage upgrade guide][schema-contract-inlining-upgrade]. Fresh graphs no
+  longer create separate root schema files.
+
+[schema-contract-inlining-upgrade]: ../docs/user/operations/upgrade.md

crates/omnigraph-catalog/src/commit.rs

@@ -26,14 +26,14 @@ use std::sync::Arc;
 
 use arrow_array::{BooleanArray, RecordBatch};
 use arrow_schema::Schema;
-use datafusion::arrow::compute::filter_record_batch;
+use datafusion::arrow::compute::{concat_batches, filter_record_batch};
 use lance::Dataset;
 use lance::dataset::{CommitBuilder, InsertBuilder, WriteMode, WriteParams};
 use lance_file::version::LanceFileVersion;
 
 use crate::error::{OmniError, Result};
 use crate::migrations::{INTERNAL_MANIFEST_SCHEMA_VERSION, read_stamp, stamp_entry};
-use crate::publisher::map_lance_publish_error;
+use crate::publisher::{PUBLISHED_ROWS_CACHE_BYTES, map_lance_publish_error};
 use crate::record::{
     StoredShape, compact_to_storage, flat_manifest_schema, flat_to_storage,
     manifest_storage_schema, written_shape,
@@ -49,7 +49,7 @@ pub(crate) async fn overwrite(
     dataset: Dataset,
     pending: RecordBatch,
     live_rows: Vec<RecordBatch>,
-) -> Result<Dataset> {
+) -> Result<(Dataset, Option<RecordBatch>)> {
     let replaced: HashSet<&str> = string_column(&pending, "object_id")?
         .iter()
         .flatten()
@@ -90,7 +90,17 @@ pub(crate) async fn overwrite(
         }
     }
     batches.push(to_storage(&pending)?);
-    commit_overwrite(dataset, batches).await
+    let retained_bytes = batches.iter().fold(0usize, |bytes, batch| {
+        bytes.saturating_add(batch.get_array_memory_size())
+    });
+    let retained = if shape == StoredShape::Packed && retained_bytes <= PUBLISHED_ROWS_CACHE_BYTES {
+        let empty = RecordBatch::new_empty(schema.clone());
+        concat_batches(&schema, batches.iter().chain([&empty])).ok()
+    } else {
+        None
+    };
+    let dataset = commit_overwrite(dataset, batches).await?;
+    Ok((dataset, retained))
 }
 
 /// Commit `batches` as the whole stored row set at `dataset`'s version + 1 (the module doc: zero

crates/omnigraph-catalog/src/graph.rs

@@ -19,8 +19,8 @@ use super::layout::{
 use super::metadata::TableVersionMetadata;
 use super::migrations::{guard_stamp, stamp_entry, stamp_for_system_columns};
 use super::state::{
-    DatasetEntry, GraphLineageRow, ManifestState, entries_to_batch, graph_lineage_row_parts,
-    read_manifest_state, read_manifest_state_and_lineage,
+    DatasetEntry, GraphLineageRow, ManifestState, SchemaContractRow, entries_to_batch,
+    graph_lineage_row_parts, read_manifest_state, read_manifest_state_and_lineage,
 };
 use super::{TableIdentity, table_path_for_identity};
 use crate::record::{compact_to_storage, manifest_storage_schema};
@@ -113,13 +113,15 @@ impl From<ManifestInitError> for OmniError {
     }
 }
 
-/// Assembles the initial entries, genesis lineage, and internal-schema stamp,
-/// and lands them all in the one `__manifest` Create commit. A returned error
-/// from the Create call is acknowledgement-unknown; reading the state back is
+/// Assembles the initial entries, genesis lineage, the `schema_contract` row
+/// and the internal-schema stamp, and lands them all in the one `__manifest`
+/// Create commit. A returned error from the Create call is
+/// acknowledgement-unknown; reading the state back is
 /// `load_initial_manifest_state`, on the confirmed post-commit side.
 pub(crate) async fn init_manifest_graph(
     root_uri: &str,
     catalog: &Catalog,
+    contract: &SchemaContractRow,
     control_session: &Arc<lance::session::Session>,
     attempt: &GenesisManifestAttempt,
 ) -> std::result::Result<Dataset, ManifestInitError> {
@@ -131,7 +133,12 @@ pub(crate) async fn init_manifest_graph(
     // commit as every table entry and the internal-schema stamp.
     let genesis_lineage = graph_lineage_row_parts(attempt.lineage(), None)?;
 
-    let manifest_batch = entries_to_batch(&entries, &version_metadata, &genesis_lineage)?;
+    let manifest_batch = entries_to_batch(
+        &entries,
+        &version_metadata,
+        &genesis_lineage,
+        Some(contract),
+    )?;
     // The internal-schema stamp rides the Create write's schema metadata, so
     // the stamp is atomic with manifest birth: no crash window can leave
     // `__manifest` durable but unstamped. The Create commit is the manifest's
@@ -236,6 +243,12 @@ pub(crate) async fn open_exact_genesis_manifest(
             "main graph head does not match this initialization attempt",
         ));
     }
+    if known_state.schema_contract.is_none() {
+        return Err(genesis_probe_mismatch(
+            root_uri,
+            "genesis carries no schema_contract row",
+        ));
+    }
     Ok((dataset, known_state, lineage_rows))
 }
 
@@ -308,7 +321,7 @@ pub(crate) async fn snapshot_state_at(
     root_uri: &str,
     branch: Option<&str>,
     version: u64,
-) -> Result<ManifestState> {
+) -> Result<(Dataset, ManifestState)> {
     let control_session = crate::lance_access::control_session();
     let dataset = open_manifest_dataset_with_session(
         root_uri.trim_end_matches('/'),
@@ -320,7 +333,8 @@ pub(crate) async fn snapshot_state_at(
         .checkout_version(version)
         .await
         .map_err(OmniError::storage)?;
-    read_manifest_state(&dataset).await
+    let state = read_manifest_state(&dataset).await?;
+    Ok((dataset, state))
 }
 
 async fn build_initial_entries(