From 7f9468238b563495871cc7d3af423ad555544cb4 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 28 Jul 2026 17:29:24 +0800 Subject: [PATCH 1/5] feat: add Dataset::referenced_files for external orphan cleanup Adds an experimental `Dataset::referenced_files()` that returns the set of storage paths referenced by all currently-present manifests, so an external distributed orphan-cleanup driver can list storage itself and delete files the set does not cover. Unlike version cleanup it unions references across every present version (not just the latest), so a file referenced only by an older-but-present version is retained. The result is an opaque `ReferencedFileSet` exposing `is_referenced(path)`, which encapsulates exact matching, the `_indices/{uuid}/` prefix rule, and the blob v2 sidecar rule (a sidecar is referenced iff its parent data file is), so callers cannot reintroduce a data-loss bug via a naive anti-join. Reads manifests concurrently and errors on datasets with branches, detached versions, external bases, or external row-id files, whose files it cannot fully represent. --- rust/lance/src/dataset.rs | 46 ++ rust/lance/src/dataset/cleanup.rs | 810 +++++++++++++++++++++++++++++- 2 files changed, 855 insertions(+), 1 deletion(-) diff --git a/rust/lance/src/dataset.rs b/rust/lance/src/dataset.rs index e73fa6233fe..f13d0d3efcb 100644 --- a/rust/lance/src/dataset.rs +++ b/rust/lance/src/dataset.rs @@ -1479,6 +1479,52 @@ impl Dataset { CleanupOperation::new(self, policy) } + /// Collect every storage path still referenced by the dataset's currently-present + /// manifests, for use by an external orphan-cleanup driver. + /// + /// **Experimental.** Intended for external distributed orphan-cleanup; the shape + /// may change. It reads the dataset's manifests and unions the files each + /// references, without listing the (potentially huge) `data/`, `_indices/`, or + /// `_deletions/` trees. The caller lists storage itself and, for each file in a + /// managed subtree (`data/`, `_deletions/`, `_transactions/`, `_indices/`, + /// `_versions/*.manifest`), deletes it only if + /// [`ReferencedFileSet::is_referenced`](crate::dataset::cleanup::ReferencedFileSet::is_referenced) + /// returns `false` and it is older than a caller-enforced age threshold. Do not + /// hand-roll an anti-join against the raw paths — see + /// [`ReferencedFileSet`](crate::dataset::cleanup::ReferencedFileSet) for + /// the full contract (blob v2 sidecars, index prefixes, never-delete categories). + /// + /// Unlike version cleanup, this considers **all present manifests** (not just the + /// latest), so a file referenced only by an older-but-present version is kept. + /// + /// Returns an error for datasets with branches, detached versions, external + /// (multi-base) fragments, or external row-id files, whose files this set cannot + /// fully represent. + /// + /// ``` + /// # use lance::{Dataset, Result}; + /// # async fn example(dataset: &Dataset, listed_paths: Vec) -> Result<()> { + /// # let is_in_managed_subtree = |_: &str| true; + /// # let older_than_threshold = |_: &str| true; + /// let referenced = dataset.referenced_files().await?; + /// for path in listed_paths { + /// // Only ever consider files in a managed subtree (never `_refs/`, + /// // `_versions/.tmp*`, or the version hint), and only past your age + /// // threshold. Then delete only what the set does not cover. + /// if is_in_managed_subtree(&path) + /// && older_than_threshold(&path) + /// && !referenced.is_referenced(&path) + /// { + /// // safe to delete `path` + /// } + /// } + /// # Ok(()) + /// # } + /// ``` + pub async fn referenced_files(&self) -> Result { + cleanup::referenced_files(self).await + } + #[allow(clippy::too_many_arguments)] async fn do_commit( base_uri: WriteDestination<'_>, diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index d8f69e7d36a..51eed620508 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -51,7 +51,7 @@ use lance_core::{ }, }; use lance_table::{ - format::{IndexMetadata, Manifest}, + format::{IndexMetadata, Manifest, RowIdMeta}, io::{ commit::ManifestLocation, deletion::deletion_file_path, @@ -79,6 +79,145 @@ struct ReferencedFiles { index_uuids: HashSet, } +/// The set of storage paths a dataset's currently-present manifests still +/// reference, for external orphan-cleanup drivers (e.g. a distributed cleanup +/// that lists storage itself and needs an authoritative "keep set"). +/// +/// **Experimental.** This API is intended for external orphan-cleanup drivers +/// and may change. It is only defined for datasets without branches, detached +/// versions, external (multi-base) fragments, or external row-id files; +/// [`Dataset::referenced_files`] errors otherwise. +/// +/// # How to use it safely +/// +/// Do **not** hand-roll an anti-join like `listed_files - exact_paths`: that +/// deletes blob sidecars, index files, tags, and staging manifests, none of +/// which are enumerated verbatim. Instead, for each file you listed under a +/// *managed subtree* (`data/`, `_deletions/`, `_transactions/`, `_indices/`, +/// and `_versions/*.manifest`), call [`is_referenced`](Self::is_referenced); +/// delete only files it returns `false` for, and only past a caller-enforced +/// age threshold (this is a point-in-time snapshot, so a file written just +/// before its commit lands is referenced by no present manifest yet). +/// +/// Never treat these as orphan candidates — this set does not describe them: +/// `_refs/` (tags/branches), staging manifests (`_versions/.tmp*`), and the +/// version-hint file. +/// +/// [`is_referenced`](Self::is_referenced) already encapsulates the blob v2 +/// sidecar rule (a sidecar `data/{key}/{blob_id}.blob` is referenced iff its +/// parent `data/{key}.lance` is) and the index-prefix rule, so callers cannot +/// get them wrong. The [`exact_paths`](Self::exact_paths) / +/// [`index_prefixes`](Self::index_prefixes) accessors exist for serializing the +/// set to distribute to workers, which then reconstruct it and match with +/// [`is_referenced`](Self::is_referenced). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReferencedFileSet { + /// Root-relative paths referenced exactly: data files (`data/{key}.lance`, + /// including data-overlay files), deletion files, transaction files, and + /// manifest files. + exact: HashSet, + /// Root-relative `_indices/{uuid}` directory prefixes (sorted). + index_prefixes: Vec, +} + +impl ReferencedFileSet { + /// Reconstruct a set from its serialized parts (see the accessors). Use this + /// on a worker after distributing [`exact_paths`](Self::exact_paths) and + /// [`index_prefixes`](Self::index_prefixes) from the driver. + pub fn new(exact_paths: Vec, index_prefixes: Vec) -> Self { + // Normalize both sides through the same `object_store::path::Path` form + // that `is_referenced` applies to its query, so a caller that + // reconstructs the set with a directory-style trailing slash (e.g. + // `_indices/{uuid}/`) or a leading slash still matches — an asymmetry + // here would silently flip a live file to unreferenced. + let mut index_prefixes: Vec = index_prefixes + .into_iter() + .map(|p| Path::from(p.as_str()).to_string()) + .collect(); + index_prefixes.sort_unstable(); + index_prefixes.dedup(); + Self { + exact: exact_paths + .into_iter() + .map(|p| Path::from(p.as_str()).to_string()) + .collect(), + index_prefixes, + } + } + + /// Whether a root-relative path is referenced by a present manifest. + /// + /// Handles the three matching rules so callers don't have to: exact match, + /// `_indices/{uuid}/` prefix match, and the blob v2 sidecar rule (a file + /// under `data/{key}/` is referenced iff `data/{key}.lance` is). A file this + /// returns `false` for — within a managed subtree and past an age threshold — + /// is an orphan. + /// + /// The path is normalized to [`object_store::path::Path`] form before + /// matching, so a caller that lists with a leading/trailing slash still + /// matches; a false negative here would delete a live file. + pub fn is_referenced(&self, root_relative_path: &str) -> bool { + // Normalize to the same `object_store::path::Path` form the producer + // stored keys in (strips leading/trailing delimiters, collapses empty + // segments), so path-shape differences can't turn a live file into a + // false negative (which the caller would delete). + let normalized = Path::from(root_relative_path).to_string(); + let path = normalized.as_str(); + + if self.exact.contains(path) { + return true; + } + // Index artifact: any file under a referenced `_indices/{uuid}/`. + if self + .index_prefixes + .iter() + .any(|prefix| is_under_prefix(path, prefix)) + { + return true; + } + // Blob v2 sidecar: `data/{key}/{blob_id}.blob` lives as long as its + // parent data file `data/{key}.lance`. Derive the parent from the + // directory name (not the sidecar file stem) and check membership. + let mut segments = path.split('/'); + if segments.next() == Some("data") + && let Some(key) = segments.next() + // A third segment means the path is *inside* `data/{key}/`. + && segments.next().is_some() + { + // Only `data/{key}/{file}` (exactly 3 segments) is the known sidecar + // layout. A deeper path doesn't match today's layout, so keep it + // conservatively rather than derive a truncated (wrong) parent and + // risk deleting a live file. + if segments.next().is_some() { + return true; + } + return self.exact.contains(&format!("data/{key}.lance")); + } + false + } + + /// Root-relative exact-match paths (data/deletion/transaction/manifest), + /// sorted. For serializing the set; use [`is_referenced`](Self::is_referenced) + /// to match a listed file. + pub fn exact_paths(&self) -> Vec { + let mut paths: Vec = self.exact.iter().cloned().collect(); + paths.sort_unstable(); + paths + } + + /// Root-relative `_indices/{uuid}` directory prefixes, sorted. For + /// serializing the set; use [`is_referenced`](Self::is_referenced) to match. + pub fn index_prefixes(&self) -> &[String] { + &self.index_prefixes + } +} + +/// Whether `path` lies strictly inside the directory `prefix` (i.e. `prefix/…`). +fn is_under_prefix(path: &str, prefix: &str) -> bool { + path.strip_prefix(prefix) + .is_some_and(|rest| rest.starts_with('/')) +} + #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct RemovalStats { pub bytes_removed: u64, @@ -1354,6 +1493,195 @@ pub async fn cleanup_old_versions( CleanupOperation::new(dataset, policy).execute().await } +/// Collect every storage path still referenced by the dataset's currently-present +/// manifests, for use by an external orphan-cleanup driver. +/// +/// This is the read-only "keep set" a distributed cleanup needs: it lists the +/// dataset's manifests and unions the files each one references, without listing +/// the (potentially huge) `data/`, `_indices/`, or `_deletions/` trees. The +/// caller lists storage itself and deletes whatever this set does not cover +/// (subject to its own age/safety checks). +/// +/// Unlike [`cleanup_old_versions`], this considers **all present manifests** +/// (not just the latest), so a file referenced only by an older-but-not-yet- +/// deleted version is included and will not be reported as an orphan. +/// +/// # Safety scope +/// +/// The returned set is only complete for datasets **without branches, detached +/// versions, external (multi-base) fragments, or external row-id files**; this +/// returns an error if any of those is present, rather than silently returning an +/// incomplete set a caller could act on: +/// +/// * Branch lineage files (referenced across `base_id`/`base_paths`) are not +/// traced, so a child branch's files could be reported as orphans. +/// * Detached versions (`d{version}.manifest`) are skipped by manifest listing, +/// so files reachable only from a detached version would be reported as orphans. +/// * External-base fragment/index paths resolve outside this dataset's root, so +/// they neither protect nor match files in another base. +/// * External row-id files are a referenced artifact this set does not enumerate +/// (matching `collect_paths`, which also rejects them). +/// +/// See [`ReferencedFileSet`] for how to interpret the result, including the +/// blob v2 sidecar rule. +pub async fn referenced_files(dataset: &Dataset) -> Result { + // Tracing branch lineage requires reading every referenced branch's + // manifests and resolving `base_id`; rather than silently under-report and + // let a caller delete a child branch's files, refuse and let them handle it. + if !dataset.branches().list().await?.is_empty() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with branches: \ + a child branch may reference files this set would omit" + .into(), + )); + } + + // External (multi-base) fragments/indices live outside this dataset's root, + // so their paths neither protect nor match files here; refuse rather than + // return a set that omits them. + if !dataset.manifest.base_paths.is_empty() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external base paths: \ + files in another base cannot be represented in this dataset's keep set" + .into(), + )); + } + + // Detached versions (`d{version}.manifest`) are skipped by the manifest + // listing below, so a set built here would omit their files; refuse if any + // exist rather than let a caller delete them. + if !dataset.list_detached_manifests().await?.is_empty() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with detached versions: \ + files reachable only from a detached version would be reported as orphans" + .into(), + )); + } + + // Collect references from every present manifest concurrently, mirroring + // `process_manifests`. Reading manifests is I/O-bound and there can be many + // present versions (the workload this API targets), so a sequential walk + // would be needlessly slow. + let collected = Mutex::new((HashSet::::new(), HashSet::::new())); + // Guard against a listing anomaly (e.g. an eventual-consistency blip or a + // concurrent cleanup that emptied `_versions/`) returning an empty keep-set: + // a raw anti-join against an empty set would treat every file as an orphan. + let manifest_count = std::sync::atomic::AtomicUsize::new(0); + + let data_dir = dataset.data_dir(); + let base = &dataset.base; + + dataset + .commit_handler + .list_manifest_locations(base, &dataset.object_store, false) + .try_for_each_concurrent(dataset.object_store.io_parallelism(), |location| { + let collected = &collected; + let data_dir = &data_dir; + let manifest_count = &manifest_count; + async move { + manifest_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let manifest = + read_manifest(&dataset.object_store, &location.path, location.size).await?; + let indexes = + read_manifest_indexes(&dataset.object_store, &location, &manifest).await?; + + let mut local_exact: Vec = Vec::new(); + // The manifest file itself is referenced (it is a present version). + local_exact.push(remove_prefix(&location.path, base).to_string()); + + for fragment in manifest.fragments.iter() { + // External row-id files are a referenced artifact we do not + // enumerate; refuse rather than under-report (matches + // `collect_paths`). Checked here so we cover every present + // version, not just the latest. + if let Some(RowIdMeta::External(external_file)) = &fragment.row_id_meta { + return Err(Error::not_supported_source( + format!( + "referenced_files is not supported on datasets with external \ + row-id files (e.g. {}): the file is referenced but not enumerated", + external_file.path + ) + .into(), + )); + } + // Base data files and data-overlay files share the + // `data/{key}.lance` namespace; both must be kept. + let data_files = fragment + .files + .iter() + .chain(fragment.overlays.iter().map(|overlay| &overlay.data_file)); + for file in data_files { + // External-base files resolve outside this root; the + // top-level `base_paths` guard only inspects the latest + // manifest, so re-check per fragment across all present + // versions rather than emit a bogus local path. + if file.base_id.is_some() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external \ + base fragments: the file lives outside this dataset's root" + .into(), + )); + } + let full = data_dir.clone().join(file.path.as_str()); + local_exact.push(remove_prefix(&full, base).to_string()); + } + if let Some(delfile) = fragment.deletion_file.as_ref() { + // Same external-base reasoning as data files above: a + // deletion file in another base resolves outside this + // root, so refuse rather than emit a phantom local path. + if delfile.base_id.is_some() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external \ + base deletion files: the file lives outside this dataset's root" + .into(), + )); + } + let delpath = deletion_file_path(base, fragment.id, delfile); + local_exact.push(remove_prefix(&delpath, base).to_string()); + } + } + + if let Some(relative_tx_path) = &manifest.transaction_file { + let tx_path = Path::parse(TRANSACTIONS_DIR)?.join(relative_tx_path.as_str()); + local_exact.push(tx_path.to_string()); + } + + let mut guard = collected.lock().unwrap(); + let (exact_paths, index_uuids) = &mut *guard; + exact_paths.extend(local_exact); + for index in &indexes { + index_uuids.insert(index.uuid.to_string()); + } + Ok(()) + } + }) + .await?; + + if manifest_count.load(std::sync::atomic::Ordering::Relaxed) == 0 { + // An opened dataset always has at least one present manifest; zero means + // a listing anomaly, not "nothing is referenced". Refuse rather than + // hand back an empty keep-set that would authorize deleting everything. + return Err(Error::not_supported_source( + "referenced_files found no manifests for an opened dataset; refusing to \ + return an empty keep-set (a cleanup driver would treat every file as an orphan)" + .into(), + )); + } + + let (exact, index_uuids) = collected.into_inner().unwrap(); + + let indices_dir = dataset.indices_dir(); + let index_prefixes: Vec = index_uuids + .into_iter() + .map(|uuid| remove_prefix(&indices_dir.clone().join(uuid.as_str()), base).to_string()) + .collect(); + + Ok(ReferencedFileSet::new( + exact.into_iter().collect(), + index_prefixes, + )) +} + /// If the dataset config has `lance.auto_cleanup` parameters set, /// this function automatically calls `dataset.cleanup_old_versions` /// every `lance.auto_cleanup.interval` versions. This function calls @@ -1538,6 +1866,7 @@ mod tests { use crate::index::DatasetIndexExt; use crate::{ dataset::transaction::{Operation, Transaction}, + dataset::write::{CommitBuilder, InsertBuilder}, dataset::{AutoCleanupParams, ReadParams, WriteMode, WriteParams, builder::DatasetBuilder}, index::vector::VectorIndexParams, }; @@ -4331,4 +4660,483 @@ mod tests { elapsed ); } + + // Collect a fixture dataset's referenced_files, keyed for easy assertions. + async fn referenced_paths(fixture: &MockDatasetFixture) -> (HashSet, Vec) { + let db = fixture.open().await.unwrap(); + let refs = db.referenced_files().await.unwrap(); + ( + refs.exact_paths().into_iter().collect(), + refs.index_prefixes().to_vec(), + ) + } + + #[tokio::test] + async fn referenced_files_keeps_older_present_version_data() { + // The heart of the contract: a file referenced ONLY by an + // older-but-not-yet-deleted version must be reported as referenced, + // otherwise an orphan-cleanup driver would delete it and break time + // travel. `overwrite` makes v1's data file unreferenced by v2 (latest) + // yet still present on disk. + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + fixture.overwrite_some_data().await.unwrap(); + + let (exact, _) = referenced_paths(&fixture).await; + + // Both versions' data files are present and must both be kept. + let data_files: Vec<_> = exact.iter().filter(|p| p.ends_with(".lance")).collect(); + assert_eq!( + data_files.len(), + 2, + "both the latest and the older-but-present version's data files must be referenced, got {exact:?}" + ); + + // Every present data file on disk must be covered by the reference set. + let registry = Arc::new(ObjectStoreRegistry::default()); + let (os, path) = + ObjectStore::from_uri_and_params(registry, &fixture.dataset_path, &fixture.os_params()) + .await + .unwrap(); + let mut stream = os.read_dir_all(&path, None); + while let Some(meta) = stream.try_next().await.unwrap() { + let rel = remove_prefix(&meta.location, &path).to_string(); + if rel.ends_with(".lance") && rel.starts_with("data/") { + assert!( + exact.contains(&rel), + "present data file {rel} was not in referenced set {exact:?}" + ); + } + } + } + + #[tokio::test] + async fn referenced_files_reports_manifests_deletions_and_transactions() { + let fixture = MockDatasetFixture::try_new().unwrap(); + let mut data_gen = BatchGenerator::new().col(Box::new( + IncrementingInt32::new().named("filter_me".to_owned()), + )); + fixture.create_with_data(data_gen.batch(16)).await.unwrap(); + fixture.delete_data("filter_me < 5").await.unwrap(); + + let (exact, _) = referenced_paths(&fixture).await; + + assert!( + exact.iter().any(|p| p.starts_with("_versions/")), + "manifest paths must be reported, got {exact:?}" + ); + assert!( + exact.iter().any(|p| p.starts_with("_transactions/")), + "transaction paths must be reported, got {exact:?}" + ); + assert!( + exact.iter().any(|p| p.starts_with("_deletions/")), + "deletion file paths must be reported, got {exact:?}" + ); + } + + #[tokio::test] + async fn referenced_files_reports_index_prefixes() { + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + fixture.create_some_index().await.unwrap(); + + let (_, index_prefixes) = referenced_paths(&fixture).await; + + assert_eq!(index_prefixes.len(), 1, "expected one index prefix"); + assert!( + index_prefixes[0].starts_with("_indices/"), + "index prefix must be under _indices/, got {index_prefixes:?}" + ); + // The prefix is a directory (a uuid), not a specific file. + assert!( + !index_prefixes[0].ends_with(".idx"), + "index prefix must be the uuid dir, not a file, got {index_prefixes:?}" + ); + + // Every present index file on disk must be covered by a prefix + "/". + let registry = Arc::new(ObjectStoreRegistry::default()); + let (os, path) = + ObjectStore::from_uri_and_params(registry, &fixture.dataset_path, &fixture.os_params()) + .await + .unwrap(); + let mut stream = os.read_dir_all(&path, None); + while let Some(meta) = stream.try_next().await.unwrap() { + let rel = remove_prefix(&meta.location, &path).to_string(); + if rel.starts_with("_indices/") { + assert!( + index_prefixes + .iter() + .any(|prefix| rel.starts_with(&format!("{prefix}/"))), + "present index file {rel} not covered by any prefix {index_prefixes:?}" + ); + } + } + } + + #[tokio::test] + async fn referenced_files_spans_multiple_fragments() { + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + fixture.append_some_data().await.unwrap(); + fixture.append_some_data().await.unwrap(); + + let (exact, _) = referenced_paths(&fixture).await; + + // Three appends => three live data files, all referenced by the union. + let data_files = exact.iter().filter(|p| p.ends_with(".lance")).count(); + assert_eq!( + data_files, 3, + "all three fragments' data files must be referenced, got {exact:?}" + ); + } + + #[tokio::test] + async fn referenced_files_keeps_blob_v2_parent_not_sidecar() { + // The sidecar contract: the parent data/{key}.lance is reported, and the + // .blob sidecar is NOT enumerated but IS covered by `is_referenced` (the + // matcher follows the parent). + let fixture = MockDatasetFixture::try_new().unwrap(); + Dataset::write( + blob_v2_batch(100 * 1024), + &fixture.dataset_path, + Some(WriteParams { + store_params: Some(fixture.os_params()), + commit_handler: Some(Arc::new(RenameCommitHandler)), + mode: WriteMode::Create, + data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2), + ..Default::default() + }), + ) + .await + .unwrap(); + assert_gt!(fixture.count_blob_files().await.unwrap(), 0); + + let db = fixture.open().await.unwrap(); + let refs = db.referenced_files().await.unwrap(); + let exact: HashSet = refs.exact_paths().into_iter().collect(); + + // Parent data file present; no sidecar path enumerated verbatim. + assert!( + exact.iter().any(|p| p.ends_with(".lance")), + "parent data file must be referenced, got {exact:?}" + ); + assert!( + !exact.iter().any(|p| p.ends_with(".blob")), + "sidecar .blob files must not be enumerated, got {exact:?}" + ); + + // Every present .blob sidecar must be covered by `is_referenced` via its + // parent, and its parent .lance must be enumerated. + let registry = Arc::new(ObjectStoreRegistry::default()); + let (os, path) = + ObjectStore::from_uri_and_params(registry, &fixture.dataset_path, &fixture.os_params()) + .await + .unwrap(); + let mut stream = os.read_dir_all(&path, None); + let mut saw_sidecar = false; + while let Some(meta) = stream.try_next().await.unwrap() { + let rel = remove_prefix(&meta.location, &path).to_string(); + if rel.ends_with(".blob") { + saw_sidecar = true; + // rel = data/{key}/{blob_id}.blob ; parent = data/{key}.lance + let parts: Vec<&str> = rel.split('/').collect(); + assert_eq!(parts.len(), 3, "unexpected sidecar layout: {rel}"); + let parent = format!("{}/{}.lance", parts[0], parts[1]); + assert!( + exact.contains(&parent), + "sidecar {rel} parent {parent} must be in the referenced set {exact:?}" + ); + // The matcher must keep the sidecar without the caller deriving + // the parent themselves. + assert!( + refs.is_referenced(&rel), + "is_referenced must cover sidecar {rel} via its parent" + ); + } + } + assert!(saw_sidecar, "test must observe at least one .blob sidecar"); + } + + #[test] + fn referenced_file_set_matcher_rules() { + // Unit-test the matcher's three rules in isolation so the caller-facing + // contract is pinned independent of dataset plumbing. + let set = ReferencedFileSet::new( + vec![ + "data/keep.lance".to_string(), + "_deletions/keep.arrow".to_string(), + ], + vec!["_indices/abc".to_string()], + ); + + // Exact match. + assert!(set.is_referenced("data/keep.lance")); + assert!(set.is_referenced("_deletions/keep.arrow")); + // Not referenced. + assert!(!set.is_referenced("data/gone.lance")); + // Index prefix: files under the dir match, the bare prefix / siblings do not. + assert!(set.is_referenced("_indices/abc/index.idx")); + assert!(set.is_referenced("_indices/abc/aux/part.bin")); + assert!(!set.is_referenced("_indices/abc")); // bare prefix, not "under" it + assert!(!set.is_referenced("_indices/abcdef/index.idx")); // sibling, not a prefix + // Blob sidecar: kept iff parent .lance is referenced. + assert!(set.is_referenced("data/keep/00000001.blob")); + assert!(!set.is_referenced("data/gone/00000001.blob")); + + // A path shape the caller might list differently (leading slash) must + // still match — otherwise a live file would be reported as an orphan. + assert!(set.is_referenced("/data/keep.lance")); + assert!(set.is_referenced("data/keep.lance/")); // trailing slash tolerated + // Deeper-than-sidecar paths under data/ are kept conservatively (unknown + // layout) rather than deriving a truncated, possibly-wrong parent. + assert!(set.is_referenced("data/keep/sub/deeper.blob")); + assert!(set.is_referenced("data/gone/sub/deeper.blob")); + + // Round-trip through the serialized accessors reproduces an equal set, + // even if a worker passes duplicate prefixes. + let round_tripped = ReferencedFileSet::new( + set.exact_paths(), + [set.index_prefixes(), set.index_prefixes()].concat(), + ); + assert_eq!(set, round_tripped); + + // A worker reconstructing the set with directory-style trailing slashes + // (a natural way to name `_indices/{uuid}/`) or leading slashes must + // still match live files — `new` normalizes its inputs symmetrically + // with the query side. + let reshaped = ReferencedFileSet::new( + vec!["/data/keep.lance".to_string()], + vec!["_indices/abc/".to_string()], + ); + assert!(reshaped.is_referenced("data/keep.lance")); + assert!(reshaped.is_referenced("_indices/abc/index.idx")); + } + + #[tokio::test] + async fn referenced_files_rejects_datasets_with_branches() { + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + let mut db = fixture.open().await.unwrap(); + fixture + .create_branch_and_load(&mut db, "dev", (None, None)) + .await + .unwrap(); + + // Reload so the main dataset observes the newly-created branch. + let db = fixture.open().await.unwrap(); + let err = db.referenced_files().await.unwrap_err(); + assert!( + matches!(err, Error::NotSupported { .. }), + "expected NotSupported for a dataset with branches, got {err:?}" + ); + assert!( + err.to_string().contains("branch"), + "error should mention branches, got {err}" + ); + } + + #[tokio::test] + async fn referenced_files_output_is_sorted_and_deterministic() { + // The accessors promise sorted output; a distributed caller may use the + // serialized set as a cache key, so equal content must compare equal. + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + fixture.append_some_data().await.unwrap(); + fixture.create_some_index().await.unwrap(); + + let db = fixture.open().await.unwrap(); + let first = db.referenced_files().await.unwrap(); + let second = db.referenced_files().await.unwrap(); + + let first_exact = first.exact_paths(); + assert!( + first_exact.windows(2).all(|w| w[0] <= w[1]), + "exact_paths must be sorted, got {first_exact:?}" + ); + assert!( + first.index_prefixes().windows(2).all(|w| w[0] <= w[1]), + "index_prefixes must be sorted, got {:?}", + first.index_prefixes() + ); + // Two calls on the same state must be equal (the set is order-independent + // and the accessors sort, so equality is meaningful). + assert_eq!( + first, second, + "referenced_files output must be deterministic" + ); + } + + #[tokio::test] + async fn referenced_files_rejects_datasets_with_detached_versions() { + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + + // Create a detached version: a committed manifest that no normal version + // references, so a keep-set built by listing normal manifests would omit it. + let db = fixture.open().await.unwrap(); + let batches: Vec = some_batch().map(|b| b.unwrap()).collect(); + let transaction = InsertBuilder::new(Arc::new(db.as_ref().clone())) + .with_params(&WriteParams { + mode: WriteMode::Append, + ..Default::default() + }) + .execute_uncommitted(batches) + .await + .unwrap(); + CommitBuilder::new(Arc::new(db.as_ref().clone())) + .with_detached(true) + .execute(transaction) + .await + .unwrap(); + + let db = fixture.open().await.unwrap(); + let err = db.referenced_files().await.unwrap_err(); + assert!( + matches!(err, Error::NotSupported { .. }), + "expected NotSupported for a dataset with detached versions, got {err:?}" + ); + assert!( + err.to_string().contains("detached"), + "error should mention detached versions, got {err}" + ); + } + + #[tokio::test] + async fn referenced_files_rejects_datasets_with_external_bases() { + use lance_table::format::BasePath; + + let base_dir = TempStrDir::default(); + let fixture = MockDatasetFixture::try_new().unwrap(); + + // Register an external base at create time so the manifest carries a + // non-empty base_paths map. + Dataset::write( + some_batch(), + &fixture.dataset_path, + Some(WriteParams { + store_params: Some(fixture.os_params()), + commit_handler: Some(Arc::new(RenameCommitHandler)), + mode: WriteMode::Create, + initial_bases: Some(vec![BasePath { + id: 1, + name: Some("external".to_string()), + is_dataset_root: false, + path: format!("file://{}", base_dir.as_str()), + }]), + ..Default::default() + }), + ) + .await + .unwrap(); + + let db = fixture.open().await.unwrap(); + let err = db.referenced_files().await.unwrap_err(); + assert!( + matches!(err, Error::NotSupported { .. }), + "expected NotSupported for a dataset with external bases, got {err:?}" + ); + assert!( + err.to_string().contains("base"), + "error should mention external bases, got {err}" + ); + } + + #[tokio::test] + async fn referenced_files_keeps_overlay_data_files() { + use crate::dataset::transaction::DataOverlayGroup; + use lance_table::format::DataFile; + use lance_table::format::overlay::{DataOverlayFile, OverlayCoverage}; + + // A data-overlay file lives only in `fragment.overlays[]`, never in + // `fragment.files`. It is a `data/{key}.lance` file, so an orphan-cleanup + // driver would delete it unless referenced_files reports it. This test is + // the tripwire: if the overlay collection is ever dropped, it fails. We + // attach the overlay as metadata (its bytes need not exist on disk for the + // reference-set computation, which reads only manifests). + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + let mut db = fixture.open().await.unwrap(); + + let fragment_id = db.get_fragments()[0].id() as u64; + let field_id = db.schema().field("indexable").unwrap().id; + let overlay_name = format!("{}.lance", Uuid::new_v4()); + let mut data_file = DataFile::new_unstarted(overlay_name.clone(), 2, 0); + data_file.fields = vec![field_id].into(); + data_file.column_indices = vec![0].into(); + + let transaction = Transaction::new( + db.manifest.version, + Operation::DataOverlay { + groups: vec![DataOverlayGroup { + fragment_id, + overlays: vec![DataOverlayFile { + data_file, + coverage: OverlayCoverage::dense([0u32].into_iter().collect()), + committed_version: 0, + }], + }], + }, + None, + ); + db.apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + let db = fixture.open().await.unwrap(); + let refs = db.referenced_files().await.unwrap(); + let overlay_rel = format!("data/{overlay_name}"); + assert!( + refs.is_referenced(&overlay_rel), + "overlay data file {overlay_rel} must be referenced, got {:?}", + refs.exact_paths() + ); + } + + #[tokio::test] + async fn referenced_files_rejects_datasets_with_external_row_ids() { + use lance_table::format::{ExternalFile, RowIdMeta}; + + // A fragment carrying an external row-id file references an artifact this + // set does not enumerate; the producer must refuse rather than under-report. + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + let db = fixture.open().await.unwrap(); + + // Re-commit the existing fragments with one carrying external row-id meta. + let mut fragments: Vec<_> = db + .get_fragments() + .iter() + .map(|f| f.metadata().clone()) + .collect(); + fragments[0].row_id_meta = Some(RowIdMeta::External(ExternalFile { + path: "_row_ids/external.rowids".to_string(), + offset: 0, + size: 16, + })); + let transaction = Transaction::new( + db.manifest.version, + Operation::Overwrite { + fragments, + schema: db.schema().clone(), + config_upsert_values: None, + initial_bases: None, + }, + None, + ); + let mut db = db; + db.apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + let err = db.referenced_files().await.unwrap_err(); + assert!( + matches!(err, Error::NotSupported { .. }), + "expected NotSupported for external row-id files, got {err:?}" + ); + assert!( + err.to_string().contains("row-id"), + "error should mention external row-id files, got {err}" + ); + } } From 9337ed65d78f00e328f34ca6f0eabf5b2f4030c7 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Thu, 30 Jul 2026 20:51:48 +0800 Subject: [PATCH 2/5] fix: adapt DataFile::new_unstarted call to new ConcreteFileVersion signature Rebased onto upstream/main, where new_unstarted takes a ConcreteFileVersion instead of (major, minor). Update the overlay test accordingly. --- rust/lance/src/dataset/cleanup.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index 51eed620508..6daf156ac64 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -5061,7 +5061,10 @@ mod tests { let fragment_id = db.get_fragments()[0].id() as u64; let field_id = db.schema().field("indexable").unwrap().id; let overlay_name = format!("{}.lance", Uuid::new_v4()); - let mut data_file = DataFile::new_unstarted(overlay_name.clone(), 2, 0); + let mut data_file = DataFile::new_unstarted( + overlay_name.clone(), + lance_file::version::ConcreteFileVersion::V2_0, + ); data_file.fields = vec![field_id].into(); data_file.column_indices = vec![0].into(); From 89e1fcf353bb68671162bd282cfdcc2dd663b467 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Fri, 31 Jul 2026 01:33:46 +0800 Subject: [PATCH 3/5] fix: keep referenced_files matching stable across serialization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Normalizing paths with `object_store::path::Path::from` was not idempotent: it percent-encodes `%`, so a key stored as `data/live%25name.lance` became `data/live%2525name.lance` on ingest and grew another `%25` every time a driver round-tripped the set through `exact_paths()` and `new()`. The worker that reconstructed the set then failed to match a live file, and a caller using this as a deletion predicate would delete it. Normalize path shape only — strip leading/trailing delimiters and collapse empty segments — and leave percent-encoding untouched, since both the producer's keys and the paths a caller lists from storage are already in object-store canonical form. Also accept a percent-decoded spelling of an exact key, which can only over-retain, and skip that retry for the ordinary alphanumeric path so the hot scan does not pay for it. Reject external row-version metadata (`created_at_version_meta`, `last_updated_at_version_meta`) alongside external row-id files: it is a root-relative referenced file this set does not enumerate, so returning `Ok` let a driver see live data as unreferenced. Reject external-base indices for the same reason the per-fragment data and deletion guards exist — the top-level `base_paths` check only inspects the latest manifest. Document `_mem_wal/` and directory-marker objects as never-delete categories. --- rust/lance/src/dataset.rs | 4 +- rust/lance/src/dataset/cleanup.rs | 222 ++++++++++++++++++++++++++---- 2 files changed, 200 insertions(+), 26 deletions(-) diff --git a/rust/lance/src/dataset.rs b/rust/lance/src/dataset.rs index f13d0d3efcb..c57865bba8d 100644 --- a/rust/lance/src/dataset.rs +++ b/rust/lance/src/dataset.rs @@ -1498,8 +1498,8 @@ impl Dataset { /// latest), so a file referenced only by an older-but-present version is kept. /// /// Returns an error for datasets with branches, detached versions, external - /// (multi-base) fragments, or external row-id files, whose files this set cannot - /// fully represent. + /// (multi-base) fragments, external row-id files, or external row-version + /// metadata, whose files this set cannot fully represent. /// /// ``` /// # use lance::{Dataset, Result}; diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index 6daf156ac64..af022b2df38 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -57,6 +57,7 @@ use lance_table::{ deletion::deletion_file_path, manifest::{read_manifest, read_manifest_indexes}, }, + rowids::version::RowDatasetVersionMeta, }; use object_store::ObjectMeta; use object_store::path::Path; @@ -85,8 +86,8 @@ struct ReferencedFiles { /// /// **Experimental.** This API is intended for external orphan-cleanup drivers /// and may change. It is only defined for datasets without branches, detached -/// versions, external (multi-base) fragments, or external row-id files; -/// [`Dataset::referenced_files`] errors otherwise. +/// versions, external (multi-base) fragments, external row-id files, or external +/// row-version metadata; [`Dataset::referenced_files`] errors otherwise. /// /// # How to use it safely /// @@ -100,8 +101,13 @@ struct ReferencedFiles { /// before its commit lands is referenced by no present manifest yet). /// /// Never treat these as orphan candidates — this set does not describe them: -/// `_refs/` (tags/branches), staging manifests (`_versions/.tmp*`), and the -/// version-hint file. +/// `_refs/` (tags/branches), staging manifests (`_versions/.tmp*`), the +/// version-hint file, and `_mem_wal/` (MemWAL entries, SSTables, and PK-index +/// sidecars, which are live data enumerated nowhere in this set). +/// +/// Directory-marker objects (zero-byte keys like `data/{key}/` or +/// `_indices/{uuid}/` that some tools create) are not referenced either — they +/// name a directory, not a file this set tracks. /// /// [`is_referenced`](Self::is_referenced) already encapsulates the blob v2 /// sidecar rule (a sidecar `data/{key}/{blob_id}.blob` is referenced iff its @@ -125,21 +131,24 @@ impl ReferencedFileSet { /// on a worker after distributing [`exact_paths`](Self::exact_paths) and /// [`index_prefixes`](Self::index_prefixes) from the driver. pub fn new(exact_paths: Vec, index_prefixes: Vec) -> Self { - // Normalize both sides through the same `object_store::path::Path` form - // that `is_referenced` applies to its query, so a caller that - // reconstructs the set with a directory-style trailing slash (e.g. - // `_indices/{uuid}/`) or a leading slash still matches — an asymmetry - // here would silently flip a live file to unreferenced. + // Normalize only path *shape* (leading/trailing delimiters, empty + // segments) — never percent-encoding. `Path::from` would percent-encode + // `%` itself (it is in object_store's INVALID set), so applying it to a + // string that is already in canonical form re-encodes it: `%25` becomes + // `%2525`. That is not idempotent, and both the producer's keys and the + // paths a caller lists from storage are already in canonical form, so + // re-encoding either side silently turns a live file into a + // false negative and the caller deletes it. let mut index_prefixes: Vec = index_prefixes .into_iter() - .map(|p| Path::from(p.as_str()).to_string()) + .map(|p| normalize_path_shape(&p)) .collect(); index_prefixes.sort_unstable(); index_prefixes.dedup(); Self { exact: exact_paths .into_iter() - .map(|p| Path::from(p.as_str()).to_string()) + .map(|p| normalize_path_shape(&p)) .collect(), index_prefixes, } @@ -153,18 +162,18 @@ impl ReferencedFileSet { /// returns `false` for — within a managed subtree and past an age threshold — /// is an orphan. /// - /// The path is normalized to [`object_store::path::Path`] form before - /// matching, so a caller that lists with a leading/trailing slash still - /// matches; a false negative here would delete a live file. + /// Leading/trailing slashes and empty path segments are normalized away, and + /// a percent-decoded spelling of a stored *exact* key is also accepted, so a + /// caller whose lister reports either form still matches. (Index prefixes are + /// `_indices/{uuid}` — hyphenated hex, so both spellings coincide.) Matching + /// errs toward "referenced": a false negative here would delete a live file. pub fn is_referenced(&self, root_relative_path: &str) -> bool { - // Normalize to the same `object_store::path::Path` form the producer - // stored keys in (strips leading/trailing delimiters, collapses empty - // segments), so path-shape differences can't turn a live file into a - // false negative (which the caller would delete). - let normalized = Path::from(root_relative_path).to_string(); + // Normalize path shape only (see `new`): percent-encoding is left alone + // so that a path listed from storage matches the key as stored. + let normalized = normalize_path_shape(root_relative_path); let path = normalized.as_str(); - if self.exact.contains(path) { + if self.contains_exact(path) { return true; } // Index artifact: any file under a referenced `_indices/{uuid}/`. @@ -191,11 +200,42 @@ impl ReferencedFileSet { if segments.next().is_some() { return true; } - return self.exact.contains(&format!("data/{key}.lance")); + return self.contains_exact(&format!("data/{key}.lance")); } false } + /// Exact-set membership, accepting either the stored spelling of a path or a + /// percent-decoded one. Keys are stored in object-store canonical (encoded) + /// form; a caller that hands back a decoded path would otherwise miss, and a + /// miss deletes a live file. Extra matches only ever over-retain. + fn contains_exact(&self, path: &str) -> bool { + if self.exact.contains(path) { + return true; + } + // Only a path that `Path::from` would rewrite can have a second + // spelling. Skip the re-encode (two allocations plus a byte scan) for + // the ordinary case: this API's caller scans every listed object, and + // Lance-generated names are alphanumeric plus `.`, `-`, `_`. The test is + // an allowlist, so an unfamiliar byte falls through to the retry rather + // than silently skipping it. + let is_already_canonical = path.split('/').all(|segment| { + // `Path::from` rewrites a bare `.`/`..` segment to `%2E`/`%2E%2E`. + segment != "." + && segment != ".." + && segment + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'_')) + }); + if is_already_canonical { + return false; + } + // Re-encode a possibly-decoded caller path and retry. `Path::from` + // percent-encodes per segment, which is exactly the producer's spelling. + let reencoded = Path::from(path).to_string(); + reencoded != path && self.exact.contains(&reencoded) + } + /// Root-relative exact-match paths (data/deletion/transaction/manifest), /// sorted. For serializing the set; use [`is_referenced`](Self::is_referenced) /// to match a listed file. @@ -212,6 +252,24 @@ impl ReferencedFileSet { } } +/// Normalize a root-relative path's *shape* only: drop leading/trailing +/// delimiters and empty segments (`a//b` → `a/b`), leaving every byte otherwise +/// untouched. +/// +/// Deliberately not [`object_store::path::Path::from`], which percent-encodes +/// `%` and so is not idempotent: keys are already stored in canonical form, and +/// re-encoding one side of a comparison would turn a live file into a false +/// negative that the caller then deletes. +fn normalize_path_shape(path: &str) -> String { + if !path.starts_with('/') && !path.ends_with('/') && !path.contains("//") { + return path.to_string(); + } + path.split('/') + .filter(|s| !s.is_empty()) + .collect::>() + .join("/") +} + /// Whether `path` lies strictly inside the directory `prefix` (i.e. `prefix/…`). fn is_under_prefix(path: &str, prefix: &str) -> bool { path.strip_prefix(prefix) @@ -1509,9 +1567,9 @@ pub async fn cleanup_old_versions( /// # Safety scope /// /// The returned set is only complete for datasets **without branches, detached -/// versions, external (multi-base) fragments, or external row-id files**; this -/// returns an error if any of those is present, rather than silently returning an -/// incomplete set a caller could act on: +/// versions, external (multi-base) fragments, external row-id files, or external +/// row-version metadata**; this returns an error if any of those is present, +/// rather than silently returning an incomplete set a caller could act on: /// /// * Branch lineage files (referenced across `base_id`/`base_paths`) are not /// traced, so a child branch's files could be reported as orphans. @@ -1521,6 +1579,9 @@ pub async fn cleanup_old_versions( /// they neither protect nor match files in another base. /// * External row-id files are a referenced artifact this set does not enumerate /// (matching `collect_paths`, which also rejects them). +/// * External row-version metadata (`created_at`/`last_updated_at`) is likewise a +/// root-relative referenced file this set does not enumerate, so a driver would +/// list it and see it as unreferenced. /// /// See [`ReferencedFileSet`] for how to interpret the result, including the /// blob v2 sidecar rule. @@ -1604,6 +1665,29 @@ pub async fn referenced_files(dataset: &Dataset) -> Result { .into(), )); } + // Same for external row-version metadata: a root-relative + // referenced file this set does not enumerate, so a driver + // would list it and see it as unreferenced. Refuse rather + // than under-report. + for (field, meta) in [ + ("created_at_version_meta", &fragment.created_at_version_meta), + ( + "last_updated_at_version_meta", + &fragment.last_updated_at_version_meta, + ), + ] { + if let Some(RowDatasetVersionMeta::External(external_file)) = meta { + return Err(Error::not_supported_source( + format!( + "referenced_files is not supported on datasets with external \ + row-version metadata ({field}, e.g. {}): the file is \ + referenced but not enumerated", + external_file.path + ) + .into(), + )); + } + } // Base data files and data-overlay files share the // `data/{key}.lance` namespace; both must be kept. let data_files = fragment @@ -1650,6 +1734,18 @@ pub async fn referenced_files(dataset: &Dataset) -> Result { let (exact_paths, index_uuids) = &mut *guard; exact_paths.extend(local_exact); for index in &indexes { + // An external-base index resolves outside this root, so its + // uuid would become a phantom local `_indices/` prefix. Same + // per-fragment reasoning as data and deletion files above: + // the top-level `base_paths` guard only sees the latest + // manifest, so re-check here across all present versions. + if index.base_id.is_some() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external \ + base indices: the index lives outside this dataset's root" + .into(), + )); + } index_uuids.insert(index.uuid.to_string()); } Ok(()) @@ -4911,6 +5007,29 @@ mod tests { ); assert!(reshaped.is_referenced("data/keep.lance")); assert!(reshaped.is_referenced("_indices/abc/index.idx")); + + // Empty segments are normalized away too, so a caller that joins a + // prefix that already ends in `/` still matches. + assert!(set.is_referenced("data//keep.lance")); + + // Percent-encoding must survive the distribute/reconstruct round trip. + // `Path::from` percent-encodes `%` itself, so normalizing with it would + // turn `%25` into `%2525` on every hop and a live object would look + // unreferenced to the worker that lists it. + let producer = ReferencedFileSet::new(vec!["data/live%25name.lance".to_string()], vec![]); + assert!(producer.is_referenced("data/live%25name.lance")); + let worker = + ReferencedFileSet::new(producer.exact_paths(), producer.index_prefixes().to_vec()); + assert!( + worker.is_referenced("data/live%25name.lance"), + "worker-side reconstruction must match the producer's keys verbatim" + ); + assert_eq!(producer, worker); + // The sidecar rule must survive the same round trip. + assert!(worker.is_referenced("data/live%25name/00000001.blob")); + // A caller whose lister reports the decoded spelling still matches: + // extra matches only over-retain, while a miss would delete a live file. + assert!(worker.is_referenced("data/live%name.lance")); } #[tokio::test] @@ -5142,4 +5261,59 @@ mod tests { "error should mention external row-id files, got {err}" ); } + + #[rstest::rstest] + #[case::created_at("created_at_version_meta")] + #[case::last_updated_at("last_updated_at_version_meta")] + #[tokio::test] + async fn referenced_files_rejects_external_row_version_metadata(#[case] field: &str) { + use lance_table::format::ExternalFile; + use lance_table::rowids::version::RowDatasetVersionMeta; + + // External row-version metadata lives under the managed `data/` prefix, so + // a cleanup driver would list it and see it as unreferenced. Since this + // set does not enumerate it, the producer must refuse. + let fixture = MockDatasetFixture::try_new().unwrap(); + fixture.create_some_data().await.unwrap(); + let db = fixture.open().await.unwrap(); + + let mut fragments: Vec<_> = db + .get_fragments() + .iter() + .map(|f| f.metadata().clone()) + .collect(); + let external = Some(RowDatasetVersionMeta::External(ExternalFile { + path: "data/external.versions".to_string(), + offset: 0, + size: 16, + })); + match field { + "created_at_version_meta" => fragments[0].created_at_version_meta = external, + _ => fragments[0].last_updated_at_version_meta = external, + } + let transaction = Transaction::new( + db.manifest.version, + Operation::Overwrite { + fragments, + schema: db.schema().clone(), + config_upsert_values: None, + initial_bases: None, + }, + None, + ); + let mut db = db; + db.apply_commit(transaction, &Default::default(), &Default::default()) + .await + .unwrap(); + + let err = db.referenced_files().await.unwrap_err(); + assert!( + matches!(err, Error::NotSupported { .. }), + "expected NotSupported for external row-version metadata, got {err:?}" + ); + assert!( + err.to_string().contains(field), + "error should name the offending field, got {err}" + ); + } } From 6250aaf57d433b9d4a3f995df7092ad86dc92ddf Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Fri, 7 Aug 2026 01:15:39 +0800 Subject: [PATCH 4/5] refactor: use Fragment::referenced_lance_files in referenced_files The accessor landed in #8267, so the hand-rolled base-plus-overlay chain here is now a second spelling of it. --- rust/lance/src/dataset/cleanup.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index 9b2880907b0..f8d684c49db 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -1708,11 +1708,7 @@ pub async fn referenced_files(dataset: &Dataset) -> Result { } // Base data files and data-overlay files share the // `data/{key}.lance` namespace; both must be kept. - let data_files = fragment - .files - .iter() - .chain(fragment.overlays.iter().map(|overlay| &overlay.data_file)); - for file in data_files { + for file in fragment.referenced_lance_files() { // External-base files resolve outside this root; the // top-level `base_paths` guard only inspects the latest // manifest, so re-check per fragment across all present From d7a33b283c3ee53ddd5dd5717a3155d27aab16f8 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 18 Aug 2026 12:52:00 +0800 Subject: [PATCH 5/5] refactor: build the keep-set on the shared manifest walk `referenced_files` had its own walk: `list_manifest_locations` fed into `try_for_each_concurrent`, with a `Mutex` around the two path sets and an atomic manifest counter. The shared walk from #8449 does the same listing and reading, so consume that instead. Two things fall out. The sets and the counter become plain locals, since the stream is consumed sequentially and the concurrency now lives in the walk. And the keep-set inherits the walk's memory budget, which it did not have before: it read manifests at `io_parallelism()` with no bound on how much manifest it held at once, on exactly the datasets this API targets. `min_version` is deliberately left unset, with the reason at the call site: a keep-set has to cover every present manifest, and skipping one would authorize deleting the files it is the last to reference. --- rust/lance/src/dataset/cleanup.rs | 230 ++++++++++++++---------------- 1 file changed, 110 insertions(+), 120 deletions(-) diff --git a/rust/lance/src/dataset/cleanup.rs b/rust/lance/src/dataset/cleanup.rs index c3f358c5800..a7a25f95400 100644 --- a/rust/lance/src/dataset/cleanup.rs +++ b/rust/lance/src/dataset/cleanup.rs @@ -35,6 +35,7 @@ use super::refs::TagContents; use crate::dataset::TRANSACTIONS_DIR; +use crate::dataset::files::scan::{ManifestScan, scan_manifests}; use crate::{Dataset, utils::temporal::utc_now}; use chrono::{DateTime, TimeDelta, Utc}; use dashmap::DashSet; @@ -1693,137 +1694,128 @@ pub async fn referenced_files(dataset: &Dataset) -> Result { )); } - // Collect references from every present manifest concurrently, mirroring - // `process_manifests`. Reading manifests is I/O-bound and there can be many - // present versions (the workload this API targets), so a sequential walk - // would be needlessly slow. - let collected = Mutex::new((HashSet::::new(), HashSet::::new())); + // Walk every present manifest on the shared scan, which bounds both read + // parallelism and the memory held by in-flight manifests. Consumed + // sequentially, so the sets below need no locking; dropping each + // `ScannedManifest` returns its share of the scan's memory budget. + // + // `min_version` is deliberately not set: a keep-set has to cover every + // present manifest, and skipping one would authorize deleting the files it + // is the last to reference. + let ManifestScan { stream, .. } = scan_manifests(dataset, None); + let mut stream = stream; + + let mut exact: HashSet = HashSet::new(); + let mut index_uuids: HashSet = HashSet::new(); // Guard against a listing anomaly (e.g. an eventual-consistency blip or a // concurrent cleanup that emptied `_versions/`) returning an empty keep-set: // a raw anti-join against an empty set would treat every file as an orphan. - let manifest_count = std::sync::atomic::AtomicUsize::new(0); + let mut manifest_count = 0usize; let data_dir = dataset.data_dir(); let base = &dataset.base; - dataset - .commit_handler - .list_manifest_locations(base, &dataset.object_store, false) - .try_for_each_concurrent(dataset.object_store.io_parallelism(), |location| { - let collected = &collected; - let data_dir = &data_dir; - let manifest_count = &manifest_count; - async move { - manifest_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let manifest = - read_manifest(&dataset.object_store, &location.path, location.size).await?; - let indexes = - read_manifest_indexes(&dataset.object_store, &location, &manifest).await?; - - let mut local_exact: Vec = Vec::new(); - // The manifest file itself is referenced (it is a present version). - local_exact.push(remove_prefix(&location.path, base).to_string()); - - for fragment in manifest.fragments.iter() { - // External row-id files are a referenced artifact we do not - // enumerate; refuse rather than under-report (matches - // `collect_paths`). Checked here so we cover every present - // version, not just the latest. - if let Some(RowIdMeta::External(external_file)) = &fragment.row_id_meta { - return Err(Error::not_supported_source( - format!( - "referenced_files is not supported on datasets with external \ - row-id files (e.g. {}): the file is referenced but not enumerated", - external_file.path - ) + while let Some(scanned) = stream.next().await { + let scanned = scanned?; + manifest_count += 1; + + // The manifest file itself is referenced (it is a present version). + exact.insert(scanned.manifest_path.clone()); + + for fragment in scanned.manifest.fragments.iter() { + // External row-id files are a referenced artifact we do not + // enumerate; refuse rather than under-report (matches + // `collect_paths`). Checked here so we cover every present + // version, not just the latest. + if let Some(RowIdMeta::External(external_file)) = &fragment.row_id_meta { + return Err(Error::not_supported_source( + format!( + "referenced_files is not supported on datasets with external \ + row-id files (e.g. {}): the file is referenced but not enumerated", + external_file.path + ) + .into(), + )); + } + // Same for external row-version metadata: a root-relative + // referenced file this set does not enumerate, so a driver + // would list it and see it as unreferenced. Refuse rather + // than under-report. + for (field, meta) in [ + ("created_at_version_meta", &fragment.created_at_version_meta), + ( + "last_updated_at_version_meta", + &fragment.last_updated_at_version_meta, + ), + ] { + if let Some(RowDatasetVersionMeta::External(external_file)) = meta { + return Err(Error::not_supported_source( + format!( + "referenced_files is not supported on datasets with external \ + row-version metadata ({field}, e.g. {}): the file is \ + referenced but not enumerated", + external_file.path + ) + .into(), + )); + } + } + // Base data files and data-overlay files share the + // `data/{key}.lance` namespace; both must be kept. + for file in fragment.referenced_lance_files() { + // External-base files resolve outside this root; the + // top-level `base_paths` guard only inspects the latest + // manifest, so re-check per fragment across all present + // versions rather than emit a bogus local path. + if file.base_id.is_some() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external \ + base fragments: the file lives outside this dataset's root" .into(), - )); - } - // Same for external row-version metadata: a root-relative - // referenced file this set does not enumerate, so a driver - // would list it and see it as unreferenced. Refuse rather - // than under-report. - for (field, meta) in [ - ("created_at_version_meta", &fragment.created_at_version_meta), - ( - "last_updated_at_version_meta", - &fragment.last_updated_at_version_meta, - ), - ] { - if let Some(RowDatasetVersionMeta::External(external_file)) = meta { - return Err(Error::not_supported_source( - format!( - "referenced_files is not supported on datasets with external \ - row-version metadata ({field}, e.g. {}): the file is \ - referenced but not enumerated", - external_file.path - ) - .into(), - )); - } - } - // Base data files and data-overlay files share the - // `data/{key}.lance` namespace; both must be kept. - for file in fragment.referenced_lance_files() { - // External-base files resolve outside this root; the - // top-level `base_paths` guard only inspects the latest - // manifest, so re-check per fragment across all present - // versions rather than emit a bogus local path. - if file.base_id.is_some() { - return Err(Error::not_supported_source( - "referenced_files is not supported on datasets with external \ - base fragments: the file lives outside this dataset's root" - .into(), - )); - } - let full = data_dir.clone().join(file.path.as_str()); - local_exact.push(remove_prefix(&full, base).to_string()); - } - if let Some(delfile) = fragment.deletion_file.as_ref() { - // Same external-base reasoning as data files above: a - // deletion file in another base resolves outside this - // root, so refuse rather than emit a phantom local path. - if delfile.base_id.is_some() { - return Err(Error::not_supported_source( - "referenced_files is not supported on datasets with external \ - base deletion files: the file lives outside this dataset's root" - .into(), - )); - } - let delpath = deletion_file_path(base, fragment.id, delfile); - local_exact.push(remove_prefix(&delpath, base).to_string()); - } + )); } - - if let Some(relative_tx_path) = &manifest.transaction_file { - let tx_path = Path::parse(TRANSACTIONS_DIR)?.join(relative_tx_path.as_str()); - local_exact.push(tx_path.to_string()); + let full = data_dir.clone().join(file.path.as_str()); + exact.insert(remove_prefix(&full, base).to_string()); + } + if let Some(delfile) = fragment.deletion_file.as_ref() { + // Same external-base reasoning as data files above: a + // deletion file in another base resolves outside this + // root, so refuse rather than emit a phantom local path. + if delfile.base_id.is_some() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external \ + base deletion files: the file lives outside this dataset's root" + .into(), + )); } + let delpath = deletion_file_path(base, fragment.id, delfile); + exact.insert(remove_prefix(&delpath, base).to_string()); + } + } - let mut guard = collected.lock().unwrap(); - let (exact_paths, index_uuids) = &mut *guard; - exact_paths.extend(local_exact); - for index in &indexes { - // An external-base index resolves outside this root, so its - // uuid would become a phantom local `_indices/` prefix. Same - // per-fragment reasoning as data and deletion files above: - // the top-level `base_paths` guard only sees the latest - // manifest, so re-check here across all present versions. - if index.base_id.is_some() { - return Err(Error::not_supported_source( - "referenced_files is not supported on datasets with external \ - base indices: the index lives outside this dataset's root" - .into(), - )); - } - index_uuids.insert(index.uuid.to_string()); - } - Ok(()) + if let Some(relative_tx_path) = &scanned.manifest.transaction_file { + let tx_path = Path::parse(TRANSACTIONS_DIR)?.join(relative_tx_path.as_str()); + exact.insert(tx_path.to_string()); + } + + for index in &scanned.indexes { + // An external-base index resolves outside this root, so its + // uuid would become a phantom local `_indices/` prefix. Same + // per-fragment reasoning as data and deletion files above: + // the top-level `base_paths` guard only sees the latest + // manifest, so re-check here across all present versions. + if index.base_id.is_some() { + return Err(Error::not_supported_source( + "referenced_files is not supported on datasets with external \ + base indices: the index lives outside this dataset's root" + .into(), + )); } - }) - .await?; + index_uuids.insert(index.uuid.to_string()); + } + } - if manifest_count.load(std::sync::atomic::Ordering::Relaxed) == 0 { + if manifest_count == 0 { // An opened dataset always has at least one present manifest; zero means // a listing anomaly, not "nothing is referenced". Refuse rather than // hand back an empty keep-set that would authorize deleting everything. @@ -1834,8 +1826,6 @@ pub async fn referenced_files(dataset: &Dataset) -> Result { )); } - let (exact, index_uuids) = collected.into_inner().unwrap(); - let indices_dir = dataset.indices_dir(); let index_prefixes: Vec = index_uuids .into_iter()