diff --git a/crates/omnigraph/src/db/omnigraph/optimize.rs b/crates/omnigraph/src/db/omnigraph/optimize.rs index 70ee8a7e..b768ac59 100644 --- a/crates/omnigraph/src/db/omnigraph/optimize.rs +++ b/crates/omnigraph/src/db/omnigraph/optimize.rs @@ -50,20 +50,6 @@ fn maint_concurrency() -> usize { .unwrap_or(DEFAULT_MAINT_CONCURRENCY) } -/// Whether the installed Lance can compact a dataset that contains blob -/// columns. `false` today: Lance `compact_files` forces -/// `BlobHandling::AllBinary` on the read side, and the blob-v2 struct decoder -/// mis-counts columns ("there were more fields in the schema than provided -/// column indices"), failing even a pristine uniform-V2_2 multi-fragment blob -/// table. Reads are unaffected (queries use descriptor handling). -/// -/// While `false`, [`optimize_all_tables`] skips blob-bearing tables and reports -/// [`SkipReason::BlobColumnsUnsupportedByLance`] instead of aborting the whole -/// sweep. Flip to `true` once the upstream Lance fix ships — the -/// `lance_surface_guards.rs::compact_files_still_fails_on_blob_columns` guard -/// turns red on that bump and forces this flip. Tracked in `docs/dev/lance.md`. -const LANCE_SUPPORTS_BLOB_COMPACTION: bool = false; - /// Retention knobs for [`cleanup_all_tables`]. At least one must be set or /// nothing is cleaned. If both are set, Lance applies them as AND (a manifest /// is kept if it satisfies either — i.e. only manifests older than BOTH the @@ -81,10 +67,6 @@ pub struct CleanupPolicyOptions { #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[non_exhaustive] pub enum SkipReason { - /// The table has one or more `Blob` columns. Lance `compact_files` forces - /// `BlobHandling::AllBinary`, which mis-decodes blob-v2 columns; see - /// [`LANCE_SUPPORTS_BLOB_COMPACTION`] and `docs/dev/lance.md`. - BlobColumnsUnsupportedByLance, /// The Lance dataset HEAD is ahead of the version recorded in /// `__manifest`, and no recovery sidecar covers that movement. `optimize` /// cannot infer whether the drift is benign maintenance or an external @@ -98,7 +80,6 @@ impl SkipReason { /// Once emitted this is part of the output contract — keep it stable. pub fn as_str(&self) -> &'static str { match self { - SkipReason::BlobColumnsUnsupportedByLance => "blob_columns_unsupported_by_lance", SkipReason::DriftNeedsRepair => "drift_needs_repair", } } @@ -108,9 +89,6 @@ impl std::fmt::Display for SkipReason { /// Human-readable reason for CLI and log output. fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let msg = match self { - SkipReason::BlobColumnsUnsupportedByLance => { - "blob columns — Lance compaction unsupported" - } SkipReason::DriftNeedsRepair => "manifest/head drift — run omnigraph repair", }; f.write_str(msg) @@ -232,9 +210,9 @@ pub async fn optimize_all_tables(db: &Omnigraph) -> Result = { + // Compute per-table paths up front, in a scope that drops the catalog + // handle before the async stream starts. + let table_tasks: Vec<(String, String)> = { let catalog = db.catalog(); let mut tasks = Vec::new(); for table_key in all_table_keys(&catalog) { @@ -242,8 +220,7 @@ pub async fn optimize_all_tables(db: &Omnigraph) -> Result Result> = futures::stream::iter(table_tasks.into_iter()) - .map(move |(table_key, full_path, has_blob)| async move { - optimize_one_table(db, table_key, full_path, has_blob).await + .map(move |(table_key, full_path)| async move { + optimize_one_table(db, table_key, full_path).await }) .buffer_unordered(concurrency) .collect() @@ -315,26 +292,7 @@ async fn optimize_one_table( db: &Omnigraph, table_key: String, full_path: String, - has_blob: bool, ) -> Result { - // Lance `compact_files` mis-decodes blob-v2 columns under the forced - // `BlobHandling::AllBinary` read (see LANCE_SUPPORTS_BLOB_COMPACTION). Skip - // blob-bearing tables before acquiring the write queue; `repair` is the - // operator tool for full manifest/head drift classification. - if has_blob && !LANCE_SUPPORTS_BLOB_COMPACTION { - tracing::warn!( - target: "omnigraph::optimize", - table = %table_key, - "skipping compaction: table has blob columns the current Lance \ - cannot rewrite (blob-v2 AllBinary decode bug); other tables \ - unaffected — rerun after the Lance fix", - ); - return Ok(TableOptimizeStats::skipped( - table_key, - SkipReason::BlobColumnsUnsupportedByLance, - )); - } - // Serialize the whole compact→publish against concurrent mutations on this // (table, main): compaction is a Rewrite op that retryable-conflicts with a // concurrent Merge/Update/Delete on overlapping fragments, and an diff --git a/crates/omnigraph/tests/maintenance.rs b/crates/omnigraph/tests/maintenance.rs index 031de726..448a7a2d 100644 --- a/crates/omnigraph/tests/maintenance.rs +++ b/crates/omnigraph/tests/maintenance.rs @@ -419,14 +419,16 @@ node Doc { // must skip blob-bearing tables (and report the skip) rather than aborting the // whole sweep. // -// Before the skip fix, `optimize()` returned that Lance error here and aborted -// the whole sweep; it now skips the blob table (`doc.skipped == Some(..)`) -// while the sibling non-blob `Tag` table still compacts. The skip is gated by -// `LANCE_SUPPORTS_BLOB_COMPACTION`; the surface guard -// `compact_files_still_fails_on_blob_columns` flags when the upstream Lance fix -// makes the skip (and this test's blob arm) removable. +// History: through Lance 7.0.0 `compact_files` mis-decoded blob-v2 columns, so +// `optimize` skipped blob tables (`SkipReason::BlobColumnsUnsupportedByLance`) +// behind `LANCE_SUPPORTS_BLOB_COMPACTION`. Lance 8.0.0+ compacts blob-v2 +// correctly (upstream #7017/#7618), the gate and skip were removed at the +// 9.0.0-beta.15 bump, and this test now pins the POSITIVE contract: a +// multi-fragment blob table compacts in the same sweep as its non-blob +// sibling, is published, and its rows survive. The surface-guard twin is +// `lance_surface_guards.rs::compact_files_succeeds_on_blob_columns`. #[tokio::test] -async fn optimize_skips_blob_table_and_reports_skip() { +async fn optimize_compacts_blob_table_alongside_plain_table() { let dir = tempfile::tempdir().unwrap(); let uri = dir.path().to_str().unwrap(); // One Blob node type (`Doc`) + one plain node type (`Tag`): proves the blob @@ -489,17 +491,21 @@ node Tag {\n slug: String @key\n}\n"; .iter() .find(|s| s.table_key == "node:Tag") .expect("Tag stat present"); - // The blob table is skipped (and reported), not compacted. - assert_eq!( - doc.skipped, - Some(SkipReason::BlobColumnsUnsupportedByLance), - "blob table must be reported as skipped", + // The blob table compacts like any other (Lance 8+ blob-v2 compaction). + assert_eq!(doc.skipped, None, "blob table must no longer be skipped"); + assert!(doc.committed, "blob table compaction must be published"); + assert!( + doc.fragments_removed >= 2 && doc.fragments_added >= 1, + "expected a real rewrite of the multi-fragment blob table, got \ + removed={} added={}", + doc.fragments_removed, + doc.fragments_added ); - assert!(!doc.committed, "skipped blob table is not compacted"); - assert_eq!(doc.fragments_removed, 0); - assert_eq!(doc.fragments_added, 0); - // The plain (non-blob) table is unaffected by the skip. assert_eq!(tag.skipped, None, "non-blob table must not be skipped"); + + // Every blob row survives the rewrite and stays readable. + let count = count_rows(&db, "node:Doc").await; + assert_eq!(count, 4, "all blob rows must survive compaction"); } // Regression: `optimize` must publish its compaction to the `__manifest` so the