apunkt/omnigraph
1
0
Fork 0
mirror of https://github.com/ModernRelay/omnigraph.git synced 2026-07-12 03:12:11 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(engine): blob tables compact — remove the LANCE_SUPPORTS_BLOB_COMPACTION gate

Browse source 
Lance 8.0.0 shipped full blob-v2 compaction (upstream #7017, hardened by
#7618 in 9.0.0-beta.15), executing this gate's documented removal plan:
the constant, the optimize skip branch, and
SkipReason::BlobColumnsUnsupportedByLance are deleted. The surface guard
inverts to a positive pin (compact_files_succeeds_on_blob_columns) and
the maintenance test now asserts a multi-fragment blob table compacts,
publishes, and keeps every row
(optimize_compacts_blob_table_alongside_plain_table).
This commit is contained in:
aaltshuler 2026-07-05 18:29:29 +03:00 • committed by Andrew Altshuler
parent 9361218a3f
commit 3fe96f2239
2 changed files with 28 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

54
crates/omnigraph/src/db/omnigraph/optimize.rs
View file

@ -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<Vec<TableOptimizeStat
let snapshot = db.fresh_snapshot_for_branch(None).await?;
// Compute per-table state (path + whether it has blob columns) up front, in
// a scope that drops the catalog handle before the async stream starts.
let table_tasks: Vec<(String, String, bool)> = {
// 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<Vec<TableOptimizeStat
continue;
};
let full_path = format!("{}/{}", db.root_uri, entry.table_path);
let has_blob = !blob_properties_for_table_key(&catalog, &table_key)?.is_empty();
tasks.push((table_key, full_path, has_blob));
tasks.push((table_key, full_path));
}
tasks
};
@ -253,8 +230,8 @@ pub async fn optimize_all_tables(db: &Omnigraph) -> Result<Vec<TableOptimizeStat
let concurrency = maint_concurrency().min(table_tasks.len()).max(1);
let stats: Vec<Result<TableOptimizeStats>> = 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<TableOptimizeStats> {
// 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

38
crates/omnigraph/tests/maintenance.rs
View file

@ -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