perf(schema): apply property changes as metadata-only commits; optimize erases dropped columns

benchmarks/deferred/README.md

@@ -24,10 +24,12 @@ GQT can benchmark them. GQT must not introduce separate maintenance commands.
   active instrument to guard. Ordinary engine correctness and deterministic
   cost tests remain at their owning layers.
 - `crates/omnigraph/tests/compaction_memory.rs.disabled`: the stock-Lance
-  compaction comparator and the engine optimize peak-allocation assertion.
-  The latter is deferred pending GQ optimization support; it is no longer an
-  executable resource regression test. Existing deterministic Blob compaction
-  tests remain active.
+  compaction comparator, the engine optimize peak-allocation assertion, the
+  schema-apply peak that stays flat in a table's Blob bytes, and the optimize
+  that erases a dropped Blob property within the compaction batch budget.
+  These are deferred pending GQ schema and optimization support; they are no
+  longer executable resource regression tests. Existing deterministic Blob
+  compaction, schema-evolution and erasure tests remain active.
 - `crates/omnigraph/tests/manifest_history_curve.rs.disabled`: four ignored
   acquisition instruments covering publication and schema-outcome histories,
   request/byte counts, retained storage and diagnostic timings. Restoring them

benchmarks/deferred/crates/omnigraph/tests/compaction_memory.rs.disabled

@@ -1,4 +1,4 @@
-//! Peak allocation of Blob-table compaction.
+//! Peak allocation of Blob-table compaction and schema apply.
 //!
 //! Lance 11 compaction materializes every managed Blob payload of one scanner
 //! batch, and a batch reads up to a whole fragment by default. `optimize`
@@ -9,9 +9,14 @@
 //! of a stock Lance compaction at the default and at the derived batch size,
 //! then asserts the engine's `optimize` peaks within the budget plus the two
 //! named allowances on fragments twice as wide as the derived batch. Every
-//! value is 1 MiB, a packed placement. It is ignored by default: the
-//! allocator counts the whole process, so the measurement needs the test
-//! binary to itself.
+//! value is 1 MiB, a packed placement. A second instrument measures schema
+//! apply's peak over Blob tables of two sizes: its column changes are
+//! metadata-only, so the peak must not grow with the table's Blob bytes. A
+//! third measures the `optimize` that erases a dropped Blob property: the
+//! rewrite reads only the remaining columns, so it peaks within the same
+//! bound as an ordinary compaction although every row also stored a dropped
+//! 1 MiB value. All are ignored by default: the allocator counts the whole
+//! process, so a measurement needs the test binary to itself (`--exact`).
 
 mod helpers;
 
@@ -238,3 +243,149 @@ async fn compaction_peak_allocation_on_blob_tables() {
         "optimize allocated {peak} bytes at its peak, above the {bound}-byte bound"
     );
 }
+
+/// Schema apply on a Doc table of `rows` 1 MiB Blob values: one apply adds a
+/// nullable property and renames another, a second drops it. Returns the peak
+/// heap above the level at entry across both applies.
+async fn schema_apply_peak(rows: usize) -> usize {
+    let graph = tempfile::tempdir().unwrap();
+    let uri = graph.path().to_str().unwrap();
+    let db = helpers::session(
+        Omnigraph::init(
+            uri,
+            "node Doc {\n    slug: String @key\n    content: Blob?\n    label: String?\n}\n",
+        )
+        .await
+        .unwrap(),
+    );
+    for (load, start) in (0..rows).step_by(16).enumerate() {
+        let mode = if load == 0 {
+            LoadMode::Overwrite
+        } else {
+            LoadMode::Merge
+        };
+        load_rows(&db, start..(start + 16).min(rows), mode).await;
+    }
+    let (applied, peak) = peak_above_baseline(async {
+        db.apply_schema(
+            "node Doc {\n    slug: String @key\n    content: Blob?\n    \
+             name: String? @rename_from(\"label\")\n    note: String?\n}\n",
+        )
+        .await?;
+        db.apply_schema("node Doc {\n    slug: String @key\n    content: Blob?\n}\n")
+            .await
+    })
+    .await;
+    assert!(applied.unwrap().applied);
+    peak
+}
+
+/// Everything schema apply allocates that is not table data: plan, catalog,
+/// manifest reads and publication. An allowance, not a derived figure.
+const SCHEMA_APPLY_ALLOWANCE: usize = 16 * MIB;
+
+#[tokio::test(flavor = "current_thread")]
+#[ignore = "instrument: schema apply peak allocation is flat in a table's Blob bytes"]
+async fn schema_apply_peak_allocation_is_flat_in_blob_bytes() {
+    let small_rows = 2 * DERIVED_ROWS;
+    let large_rows = 4 * DERIVED_ROWS;
+    let small = schema_apply_peak(small_rows).await;
+    let large = schema_apply_peak(large_rows).await;
+    for (rows, peak) in [(small_rows, small), (large_rows, large)] {
+        eprintln!(
+            "schema apply (add + rename, then drop) on {rows} rows of 1 MiB Blob: \
+             peak {:.1} MiB (allowance {:.1} MiB)",
+            peak as f64 / MIB as f64,
+            SCHEMA_APPLY_ALLOWANCE as f64 / MIB as f64
+        );
+    }
+    assert!(
+        large <= SCHEMA_APPLY_ALLOWANCE,
+        "schema apply allocated {large} bytes at its peak over {large_rows} MiB of Blob values"
+    );
+    assert!(
+        large <= small + 4 * MIB,
+        "schema apply's peak grew with the table's Blob bytes: {small} -> {large}"
+    );
+}
+
+#[tokio::test(flavor = "current_thread")]
+#[ignore = "instrument: erasing a dropped Blob property compacts within the batch budget"]
+async fn dropped_blob_erasure_peak_allocation_is_within_the_compaction_bound() {
+    let graph = tempfile::tempdir().unwrap();
+    let uri = graph.path().to_str().unwrap();
+    let db = helpers::session(
+        Omnigraph::init(
+            uri,
+            "node Doc {\n    slug: String @key\n    content: Blob?\n    secret: Blob?\n}\n",
+        )
+        .await
+        .unwrap(),
+    );
+    // Eight loads of 16 MiB, compacted into one fragment of 64 rows, each row
+    // holding a kept and a soon-dropped 1 MiB value.
+    for load in 0..8 {
+        let lines = (load * 8..(load + 1) * 8)
+            .map(|row| {
+                let encode = |bytes: Vec<u8>| {
+                    format!(
+                        "base64:{}",
+                        base64::engine::general_purpose::STANDARD.encode(bytes)
+                    )
+                };
+                serde_json::json!({
+                    "type": "Doc",
+                    "data": {
+                        "slug": format!("d{row:03}"),
+                        "content": encode(payload(row)),
+                        "secret": encode(payload(row + 1)),
+                    },
+                })
+                .to_string()
+            })
+            .collect::<Vec<_>>();
+        let mode = if load == 0 {
+            LoadMode::Overwrite
+        } else {
+            LoadMode::Merge
+        };
+        db.load_jsonl(&lines.join("\n"), mode).await.unwrap();
+    }
+    db.optimize().await.unwrap();
+    let doc = |stats: Vec<omnigraph::db::DatasetOptimizeStats>| {
+        stats
+            .into_iter()
+            .find(|stat| stat.type_key == "node:Doc")
+            .unwrap()
+    };
+    assert!(
+        !doc(db.optimize().await.unwrap()).committed,
+        "test precondition: nothing left to compact before the drop"
+    );
+    let fragment_rows = helpers::open_pinned_dataset_for_test(&db, "main", "node:Doc")
+        .await
+        .get_fragments()
+        .iter()
+        .map(|fragment| fragment.metadata().physical_rows.unwrap())
+        .collect::<Vec<_>>();
+    assert_eq!(fragment_rows, vec![WIDE_FRAGMENT_ROWS], "test precondition");
+    db.apply_schema("node Doc {\n    slug: String @key\n    content: Blob?\n}\n")
+        .await
+        .unwrap();
+
+    let (stats, peak) = peak_above_baseline(db.optimize()).await;
+    let doc = doc(stats.unwrap());
+    assert!(doc.committed);
+    assert_eq!(doc.fragments_removed, 1, "the measured optimize erases");
+    let bound = BUDGET + IN_FLIGHT_VALUE_ALLOWANCE + NON_PAYLOAD_ALLOWANCE;
+    eprintln!(
+        "engine optimize erasing a dropped Blob, {WIDE_FRAGMENT_ROWS} rows of 1 MiB kept \
+         + 1 MiB dropped: peak {:.1} MiB (bound {:.1} MiB)",
+        peak as f64 / MIB as f64,
+        bound as f64 / MIB as f64
+    );
+    assert!(
+        peak <= bound,
+        "the erasing optimize allocated {peak} bytes at its peak, above the {bound}-byte bound"
+    );
+}

changelog.d/dropped-property-storage.changed.md

@@ -0,0 +1,12 @@
+- Dropping a property no longer rewrites its table. Through v0.13.0 schema
+  apply rewrote the table without the dropped values, so `omnigraph cleanup` alone
+  erased them; now they stay in the table's data files until
+  `omnigraph optimize` rewrites every fragment that holds them, whatever its
+  size. To erase them, run optimize, delete every branch created from a commit
+  before it and every tag naming such a commit (merging a branch does not
+  release its files), then run `cleanup` with a retention that excludes the
+  commits before the optimize, for example `--keep 1`. Dropped types are still
+  reclaimed by `cleanup` alone. See
+  [dropping declarations][dropped-property-storage-drops].
+
+[dropped-property-storage-drops]: ../docs/user/schema/index.md#schema-changes

changelog.d/schema-apply-metadata-only.performance.md

@@ -0,0 +1,12 @@
+- Adding a nullable property, renaming a property and dropping a property no
+  longer rewrite the table. Schema apply changes only the table's metadata, so
+  its memory and I/O no longer grow with the rows and Blob values the table
+  stores, only with its metadata (fragments, columns and indexes): a schema
+  change on a large Blob table, which previously held every managed Blob value
+  of the table in memory and could exhaust a server running the deployment,
+  now reads no row and no Blob value. Existing indexes keep covering their rows
+  instead of waiting for the next `optimize`, and a stored external Blob
+  reference to a byte range, which schema apply previously refused, is kept as
+  stored. See [schema changes][schema-apply-metadata-only-changes].
+
+[schema-apply-metadata-only-changes]: ../docs/user/schema/index.md#schema-changes

crates/omnigraph-compiler/src/catalog/schema_plan.rs

@@ -97,9 +97,11 @@ pub enum SchemaMigrationStep {
         type_kind: SchemaTypeKind,
         name: String,
     },
-    /// Remove a property from an existing type: the table is rewritten
-    /// without the column. Older table versions keep it, and older graph
-    /// commits read it until `omnigraph cleanup` stops retaining them.
+    /// Remove a property from an existing type: a metadata-only commit takes
+    /// the column out of the table's schema and rewrites no row. Its values
+    /// stay in the existing data files until a compaction rewrites them, and
+    /// older graph commits read it until `omnigraph cleanup` stops retaining
+    /// them.
     DropProperty {
         type_kind: SchemaTypeKind,
         type_name: String,
@@ -716,11 +718,9 @@ fn plan_properties(
         .iter()
         .filter(|property| !consumed.contains(&property.property_id))
     {
-        // Property removed from the desired schema. Apply reuses the
-        // stage_overwrite rewrite path — batch_for_schema_apply_rewrite
-        // iterates target_schema.fields(), so the dropped column is
-        // naturally projected away. Retention, which the OG-DS-104
-        // destructive tier expects, is stated on
+        // Property removed from the desired schema. Apply drops the column
+        // from the table's schema by a metadata-only commit. Retention, which
+        // the OG-DS-104 destructive tier expects, is stated on
         // `SchemaMigrationStep::DropProperty`.
         steps.push(SchemaMigrationStep::DropProperty {
             type_kind,
@@ -1443,8 +1443,8 @@ node Account @rename_from("User") {
         // Removing a property from the desired schema emits
         // DropProperty (schema-lint v1 chassis commit #3,
         // MR-694). The plan is `supported = true` — the apply path
-        // handles the drop via the existing stage_overwrite rewrite
-        // projection. Verified at the integration level by
+        // drops the column by a metadata-only commit. Verified at the
+        // integration level by
         // `apply_schema_drops_a_nullable_property_and_preserves_prior_version`
         // in `crates/omnigraph/tests/schema_apply.rs`.
         let accepted = ir(r#"

crates/omnigraph-core/src/instrumentation.rs

@@ -1237,7 +1237,7 @@ pub struct MergeWriteProbes {
     /// while its assignments resolve must leave this at zero.
     pub mutation_table_open_calls: Arc<AtomicU64>,
     /// Blob payload values a rewrite consumed while rebuilding descriptor rows
-    /// into a logical source (a keyed write or a schema rewrite), or that
+    /// into a logical source (a keyed write or a branch merge), or that
     /// export, a change image or an entity read rendered: one per managed
     /// value, counted after the batched managed read returned it and its
     /// length matched, and one per external object read (rewrites only).

crates/omnigraph/src/blob.rs

@@ -1169,7 +1169,7 @@ impl<'a> BlobDescriptorDecoder<'a> {
     /// resolves managed (inline/packed/dedicated) bytes from the row's owning
     /// data file, then `position`/`size`/`blob_id` within it. So the identity
     /// must be qualified by *which* data file. It uses the file's stable path
-    /// (`data_file_path`) — a per-file v4 UUID Lance mints once and never
+    /// (`managed_data_file`) — a per-file v4 UUID Lance mints once and never
     /// rewrites; compaction, `Overwrite`, and per-branch writes all produce a
     /// NEW UUID. Unlike the numeric fragment id, this is globally unique: it
     /// does not restart at 0 on `Overwrite` and is not branch-local, so equal
@@ -1178,7 +1178,16 @@ impl<'a> BlobDescriptorDecoder<'a> {
     /// placement, so their identity is source-independent. Null uses the
     /// classified state, never sentinel child values, so the two physical null
     /// encodings share one identity.
-    pub(crate) fn physical_identity(&self, row: usize, data_file_path: &str) -> Result<String> {
+    ///
+    /// `managed_data_file` is called only for a managed descriptor. A null or
+    /// external identity names no file, and a fragment written before its Blob
+    /// column was added has no data file for that column at all: Lance reads
+    /// the column there as null.
+    pub(crate) fn physical_identity<'path>(
+        &self,
+        row: usize,
+        managed_data_file: impl FnOnce() -> Result<&'path str>,
+    ) -> Result<String> {
         match self.classify(row)? {
             BlobDescriptor::Null => Ok("null".to_string()),
             BlobDescriptor::External {
@@ -1190,7 +1199,8 @@ impl<'a> BlobDescriptorDecoder<'a> {
                 length.map(|len| len.to_string()).unwrap_or_default(),
             )),
             BlobDescriptor::Managed { .. } => Ok(format!(
-                "mgd:{data_file_path}:{}:{}:{}:{}",
+                "mgd:{}:{}:{}:{}:{}",
+                managed_data_file()?,
                 self.kinds.value(row),
                 self.positions.value(row),
                 self.sizes.value(row),
@@ -2406,6 +2416,55 @@ mod tests {
         );
     }
 
+    /// Only a managed identity names its owning data file. A null or external
+    /// descriptor never asks for one, since a fragment written before its Blob
+    /// column was added has none; a managed descriptor whose file cannot be
+    /// resolved keeps the resolver's integrity error.
+    #[test]
+    fn physical_identity_resolves_a_data_file_only_for_managed_descriptors() {
+        let no_file =
+            || -> Result<&'static str> { panic!("a null or external identity names no data file") };
+        let null = StructArray::new_null(fields(), 1);
+        assert_eq!(
+            BlobDescriptorDecoder::try_new(&null)
+                .unwrap()
+                .physical_identity(0, no_file)
+                .unwrap(),
+            "null"
+        );
+        let external = descriptor(
+            Some(3),
+            Some(4),
+            Some(8),
+            Some(0),
+            Some("s3://bucket/object"),
+        );
+        assert_eq!(
+            BlobDescriptorDecoder::try_new(&external)
+                .unwrap()
+                .physical_identity(0, no_file)
+                .unwrap(),
+            "ext:4:8:s3://bucket/object"
+        );
+
+        let managed = descriptor(Some(1), Some(16), Some(8), Some(2), Some(""));
+        let decoder = BlobDescriptorDecoder::try_new(&managed).unwrap();
+        assert_eq!(
+            decoder.physical_identity(0, || Ok("a.lance")).unwrap(),
+            "mgd:a.lance:1:16:8:2"
+        );
+        assert_blob_integrity(
+            decoder
+                .physical_identity(0, || {
+                    Err(OmniError::blob_integrity(
+                        "fragment 0 has no data file for blob field 3",
+                    ))
+                })
+                .unwrap_err(),
+            "has no data file",
+        );
+    }
+
     #[test]
     fn exact_v2_shape_unknown_kind_bounds_and_arithmetic_fail_closed() {
         for index in 0..5 {

crates/omnigraph/src/changes/row_compare.rs

@@ -288,15 +288,19 @@ impl BatchCursor {
                 .downcast_ref::<StructArray>()
                 .expect("Blob descriptor shape was validated when the batch arrived");
             let decoder = BlobDescriptorDecoder::try_new(descriptions)?;
-            let fragment_id = row_addresses
-                .expect("Blob batches carry row addresses")
-                .value(row)
-                >> 32;
-            let data_file_path =
-                data_file_path_for_field(dataset.fragments(), fragment_id, blob_column.field_id)?;
+            // Only a managed descriptor names a data file. A fragment written
+            // before this Blob column was added has none for it, and its
+            // descriptor is the null Lance reads there.
+            let identity = decoder.physical_identity(row, || {
+                let fragment_id = row_addresses
+                    .expect("Blob batches carry row addresses")
+                    .value(row)
+                    >> 32;
+                data_file_path_for_field(dataset.fragments(), fragment_id, blob_column.field_id)
+            })?;
             blob_signatures.push(BlobColumnSig {
                 name: blob_column.name.clone(),
-                identity: decoder.physical_identity(row, data_file_path)?,
+                identity,
                 managed: matches!(decoder.classify(row)?, BlobDescriptor::Managed { .. }),
             });
         }
@@ -429,7 +433,9 @@ impl OrderedRows {
 
 /// Resolve the immutable data-file path that holds `field_id` in the fragment
 /// owning a row — the stable UUID qualifier for a managed-Blob identity. Reads
-/// only the in-memory manifest.
+/// only the in-memory manifest. Called for managed descriptors only, so a
+/// fragment without a data file for the field is an integrity failure: the
+/// descriptor claims bytes nothing stores.
 fn data_file_path_for_field(
     fragments: &[lance_table::format::Fragment],
     fragment_id: u64,