diff --git a/rust/lance-namespace-impls/src/dir/manifest.rs b/rust/lance-namespace-impls/src/dir/manifest.rs index 6dce81ef50b..751474ed015 100644 --- a/rust/lance-namespace-impls/src/dir/manifest.rs +++ b/rust/lance-namespace-impls/src/dir/manifest.rs @@ -101,6 +101,17 @@ static BASE_OBJECTS_VALUE_FIELD: LazyLock = LazyLock::new(|| { // commit retry budget so multi-process namespace writes can make progress. const DEFAULT_MANIFEST_REWRITE_COMMIT_RETRIES: u32 = 20; const MANIFEST_INDEX_BATCH_SIZE: usize = 8192; +// Dataset config keys of Lance's automatic old-version cleanup policy. Lance +// releases before 7.0 wrote this policy into every new dataset by default, so +// directory manifests created by those releases still carry it. +const AUTO_CLEANUP_CONFIG_PREFIX: &str = "lance.auto_cleanup."; + +fn has_auto_cleanup_config(manifest: &Manifest) -> bool { + manifest + .config + .keys() + .any(|key| key.starts_with(AUTO_CLEANUP_CONFIG_PREFIX)) +} /// Object types that can be stored in the manifest #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -597,23 +608,20 @@ pub struct NamespaceInfo { /// A wrapper around a Dataset that provides concurrent access. /// /// This can be cloned cheaply. It supports concurrent reads or exclusive writes. -/// The manifest dataset uses contiguous attached versions and this module never -/// runs old-version cleanup on it, allowing reads to check only the immediate -/// successor manifest before deciding whether a reload is needed. +/// +/// Manifest rewrites never run old-version cleanup or add an auto-cleanup +/// policy. Policy-free manifests therefore keep contiguous attached versions, +/// so reloads only need to probe the immediate successor manifest. A manifest +/// created by an older Lance release can still carry an auto-cleanup policy, +/// and older writers act on it by removing old versions, which leaves gaps +/// after the held version. While that version carries the policy, reloads +/// resolve the latest version instead. #[derive(Debug, Clone)] pub struct DatasetConsistencyWrapper(Arc>); impl DatasetConsistencyWrapper { /// Create a new wrapper with the given dataset. pub fn new(dataset: Dataset) -> Self { - debug_assert!( - !dataset - .manifest() - .config - .keys() - .any(|key| key.starts_with("lance.auto_cleanup.")), - "the directory manifest dataset must not enable old-version cleanup" - ); Self(Arc::new(RwLock::new(dataset))) } @@ -679,25 +687,17 @@ impl DatasetConsistencyWrapper { dataset_uri, current_version ); - // The directory manifest table uses contiguous attached versions and - // does not run old-version cleanup, so the immediate successor probe is - // enough to detect changes without resolving or loading the latest - // manifest on every namespace read. - let has_successor_version = read_guard.has_successor_version().await.map_err(|e| { - lance_core::Error::from(NamespaceError::Internal { - message: format!("Failed to check dataset staleness: {:?}", e), - }) - })?; + let has_newer_version = Self::has_newer_version(&read_guard).await?; log::debug!( - "Reload checked successor_version_exists={} for uri={}, current_version={}", - has_successor_version, + "Reload checked has_newer_version={} for uri={}, current_version={}", + has_newer_version, dataset_uri, current_version ); drop(read_guard); // If already up-to-date, return early - if !has_successor_version { + if !has_newer_version { log::debug!("Already up-to-date for uri={}", dataset_uri); return Ok(()); } @@ -719,13 +719,7 @@ impl DatasetConsistencyWrapper { current_version ); - let has_successor_version = dataset.has_successor_version().await.map_err(|e| { - lance_core::Error::from(NamespaceError::Internal { - message: format!("Failed to check dataset staleness: {:?}", e), - }) - })?; - - if has_successor_version { + if Self::has_newer_version(dataset).await? { dataset.checkout_latest().await.map_err(|e| { lance_core::Error::from(NamespaceError::Internal { message: format!("Failed to checkout latest: {:?}", e), @@ -735,6 +729,25 @@ impl DatasetConsistencyWrapper { Ok(()) } + + /// Return whether a version newer than the held one has been committed. + /// + /// Probing the immediate successor avoids resolving the latest manifest on + /// every namespace read, but misses newer versions once cleanup has removed + /// the successor. A held version with an auto-cleanup policy can have such + /// gaps after it, so it resolves the latest version instead. + async fn has_newer_version(dataset: &Dataset) -> Result { + let has_newer_version = if has_auto_cleanup_config(dataset.manifest()) { + dataset.is_stale().await + } else { + dataset.has_successor_version().await + }; + has_newer_version.map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!("Failed to check dataset staleness: {:?}", e), + }) + }) + } } pub struct DatasetReadGuard<'a> { @@ -1340,7 +1353,13 @@ impl ManifestNamespace { }) .collect::>(); fragments.sort_by_key(|fragment| fragment.id); - Manifest::new_from_previous(previous, schema, Arc::new(fragments)) + let mut manifest = Manifest::new_from_previous(previous, schema, Arc::new(fragments)); + // Drop a legacy auto-cleanup policy so later commits, including those of + // older writers that inherit this config, stop removing old versions. + manifest + .config + .retain(|key, _| !key.starts_with(AUTO_CLEANUP_CONFIG_PREFIX)); + manifest } async fn build_manifest_indices( @@ -3863,16 +3882,18 @@ impl LanceNamespace for ManifestNamespace { #[cfg(test)] mod tests { use super::{ - BASE_OBJECTS_INDEX_NAME, ConflictResolution, CopyOnWriteMutation, DeleteObjectMutation, - LANCE_DATA_DIR, LANCE_INDICES_DIR, MANIFEST_TABLE_NAME, ManifestBatchBuilder, - ManifestEntry, ManifestIndexAccumulator, ManifestNamespace, ManifestOutputRow, - ManifestRowValue, ManifestStreamMutation, OBJECT_ID_INDEX_NAME, OBJECT_TYPE_INDEX_NAME, - ObjectType, + BASE_OBJECTS_INDEX_NAME, ConflictResolution, CopyOnWriteMutation, + DatasetConsistencyWrapper, DeleteObjectMutation, LANCE_DATA_DIR, LANCE_INDICES_DIR, + MANIFEST_TABLE_NAME, ManifestBatchBuilder, ManifestEntry, ManifestIndexAccumulator, + ManifestNamespace, ManifestOutputRow, ManifestRowValue, ManifestStreamMutation, + OBJECT_ID_INDEX_NAME, OBJECT_TYPE_INDEX_NAME, ObjectType, has_auto_cleanup_config, }; - use crate::DirectoryNamespaceBuilder; + use crate::{DirectoryNamespace, DirectoryNamespaceBuilder}; use arrow::datatypes::DataType; use bytes::Bytes; use futures::StreamExt; + use lance::dataset::builder::DatasetBuilder; + use lance::dataset::cleanup::CleanupPolicyBuilder; use lance::index::DatasetIndexExt; use lance_core::utils::tempfile::TempStdDir; use lance_io::object_store::{ObjectStore, ObjectStoreParams, ObjectStoreRegistry}; @@ -4263,6 +4284,142 @@ mod tests { .collect() } + async fn create_manifest_only_table(ns: &DirectoryNamespace, name: &str) { + let mut create_request = CreateTableRequest::new(); + create_request.id = Some(vec![name.to_string()]); + ns.create_table(create_request, Bytes::from(create_test_ipc_data())) + .await + .unwrap(); + } + + fn copy_test_data_to_tmp(table_path: &str) -> std::io::Result { + let source = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../test_data") + .join(table_path); + let temp_dir = TempStdDir::default(); + let mut directories = vec![(source, temp_dir.to_path_buf())]; + while let Some((source, destination)) = directories.pop() { + std::fs::create_dir_all(&destination)?; + for entry in std::fs::read_dir(source)? { + let entry = entry?; + let target = destination.join(entry.file_name()); + if entry.file_type()?.is_dir() { + directories.push((entry.path(), target)); + } else { + std::fs::copy(entry.path(), target)?; + } + } + } + Ok(temp_dir) + } + + #[derive(Debug)] + enum ManifestAccess { + Get, + GetRefreshed, + GetMut, + } + + /// Readers and writers holding a manifest created by a released old writer + /// must observe newer commits after cleanup removes the immediate successor. + #[rstest] + #[case::get(ManifestAccess::Get)] + #[case::get_refreshed(ManifestAccess::GetRefreshed)] + #[case::get_mut(ManifestAccess::GetMut)] + #[tokio::test] + async fn test_legacy_auto_cleanup_manifest_reload_skips_removed_versions( + #[case] access: ManifestAccess, + ) { + let temp_dir = copy_test_data_to_tmp("v6.0.1/dir_manifest/namespace").unwrap(); + let temp_path = temp_dir.to_str().unwrap(); + let manifest_uri = format!("{}/{}", temp_path, MANIFEST_TABLE_NAME); + // Directory listing would surface tables the manifest misses, so the + // namespaces read only the manifest. + let build_namespace = || { + DirectoryNamespaceBuilder::new(temp_path) + .manifest_enabled(true) + .dir_listing_enabled(false) + .build() + }; + + let manifest_ds = DatasetBuilder::from_uri(&manifest_uri) + .load() + .await + .unwrap(); + assert!(has_auto_cleanup_config(manifest_ds.manifest())); + assert_eq!( + manifest_ds.manifest().config["lance.auto_cleanup.interval"], + "20" + ); + assert_eq!( + manifest_ds.manifest().config["lance.auto_cleanup.older_than"], + "14days" + ); + let legacy_version = manifest_ds.version().version; + let wrapper = DatasetConsistencyWrapper::new(manifest_ds); + let reader = build_namespace().await.unwrap(); + let writer = build_namespace().await.unwrap(); + create_manifest_only_table(&writer, "t2").await; + create_manifest_only_table(&writer, "t3").await; + + let latest = DatasetBuilder::from_uri(&manifest_uri) + .load() + .await + .unwrap(); + let latest_version = latest.version().version; + assert_eq!(latest_version, legacy_version + 2); + assert!(!has_auto_cleanup_config(latest.manifest())); + let policy = CleanupPolicyBuilder::default() + .retain_n_versions(&latest, 1) + .await + .unwrap() + .delete_unverified(true) + .build(); + latest.cleanup_with_policy(policy).await.unwrap(); + let error = latest + .checkout_version(legacy_version + 1) + .await + .unwrap_err(); + assert!( + matches!(error, lance_core::Error::DatasetNotFound { .. }), + "unexpected missing-version error: {error:?}" + ); + assert!(error.to_string().contains("not found")); + + let reloaded_version = match access { + ManifestAccess::Get => wrapper.get().await.unwrap().version().version, + ManifestAccess::GetRefreshed => { + wrapper.get_refreshed().await.unwrap().version().version + } + ManifestAccess::GetMut => wrapper.get_mut().await.unwrap().version().version, + }; + assert_eq!(reloaded_version, latest_version); + assert!(!has_auto_cleanup_config( + wrapper.get().await.unwrap().manifest() + )); + + let mut list_request = ListTablesRequest::new(); + list_request.id = Some(vec![]); + let mut tables = reader + .list_tables(list_request.clone()) + .await + .unwrap() + .tables; + tables.sort(); + assert_eq!(tables, vec!["t1", "t2", "t3"]); + + // Once refreshed to a policy-free version, the successor fast path must + // still observe subsequent writes. + create_manifest_only_table(&writer, "t4").await; + assert_eq!( + wrapper.get().await.unwrap().version().version, + latest_version + 1 + ); + let mut tables = reader.list_tables(list_request).await.unwrap().tables; + tables.sort(); + assert_eq!(tables, vec!["t1", "t2", "t3", "t4"]); + } + #[tokio::test] async fn test_manifest_rewrite_preserves_utf8_metadata_and_base_objects() { let temp_dir = TempStdDir::default(); diff --git a/test_data/v6.0.1/dir_manifest/datagen.py b/test_data/v6.0.1/dir_manifest/datagen.py new file mode 100644 index 00000000000..b6bd8149a74 --- /dev/null +++ b/test_data/v6.0.1/dir_manifest/datagen.py @@ -0,0 +1,28 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The Lance Authors + +# Generate with: +# uv run --isolated --no-project --python 3.12 --with pylance==6.0.1 datagen.py +import shutil +from pathlib import Path + +import lance +import pyarrow as pa +from lance.namespace import DirectoryNamespace +from lance_namespace import CreateTableRequest + +assert lance.__version__ == "6.0.1" + +root = Path(__file__).parent / "namespace" +shutil.rmtree(root, ignore_errors=True) +namespace = DirectoryNamespace( + root=str(root), manifest_enabled="true", dir_listing_enabled="false" +) +batch = pa.record_batch({"id": [1]}) +sink = pa.BufferOutputStream() +with pa.ipc.new_stream(sink, batch.schema) as writer: + writer.write_batch(batch) +namespace.create_table(CreateTableRequest(id=["t1"]), sink.getvalue().to_pybytes()) +manifest = lance.dataset(root / "__manifest") +assert manifest.config()["lance.auto_cleanup.interval"] == "20" +assert manifest.config()["lance.auto_cleanup.older_than"] == "14days" diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/3fd5246b-80ed-4455-a551-49b2cccc138c/page_data.lance b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/3fd5246b-80ed-4455-a551-49b2cccc138c/page_data.lance new file mode 100644 index 00000000000..fd7198b35f7 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/3fd5246b-80ed-4455-a551-49b2cccc138c/page_data.lance differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/3fd5246b-80ed-4455-a551-49b2cccc138c/page_lookup.lance b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/3fd5246b-80ed-4455-a551-49b2cccc138c/page_lookup.lance new file mode 100644 index 00000000000..d99fed84900 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/3fd5246b-80ed-4455-a551-49b2cccc138c/page_lookup.lance differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/90a4f46e-95b4-4ad0-b3ae-66c71c9270ad/bitmap_page_lookup.lance b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/90a4f46e-95b4-4ad0-b3ae-66c71c9270ad/bitmap_page_lookup.lance new file mode 100644 index 00000000000..1f85d1ccab3 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/90a4f46e-95b4-4ad0-b3ae-66c71c9270ad/bitmap_page_lookup.lance differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/9f61f26d-f349-43e8-8d66-b6ab1d70a513/bitmap_page_lookup.lance b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/9f61f26d-f349-43e8-8d66-b6ab1d70a513/bitmap_page_lookup.lance new file mode 100644 index 00000000000..2cf940fadf8 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_indices/9f61f26d-f349-43e8-8d66-b6ab1d70a513/bitmap_page_lookup.lance differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/0-f8b69967-f69b-430b-bd19-6de4ecc6661e.txn b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/0-f8b69967-f69b-430b-bd19-6de4ecc6661e.txn new file mode 100644 index 00000000000..b8193dc3e9c --- /dev/null +++ b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/0-f8b69967-f69b-430b-bd19-6de4ecc6661e.txn @@ -0,0 +1,4 @@ +$f8b69967-f69b-430b-bd19-6de4ecc6661e²×U object_id ÿÿÿÿÿÿÿÿÿ*string8R1 +,lance-schema:unenforced-primary-key:position0`$ object_type ÿÿÿÿÿÿÿÿÿ*string8#location ÿÿÿÿÿÿÿÿÿ*string08#metadata ÿÿÿÿÿÿÿÿÿ*string08% base_objects ÿÿÿÿÿÿÿÿÿ*list08 object_id *string08"! +lance.auto_cleanup.interval20"' +lance.auto_cleanup.older_than14days \ No newline at end of file diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/1-442b76c1-0432-4290-b9a5-0ba180d74a9f.txn b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/1-442b76c1-0432-4290-b9a5-0ba180d74a9f.txn new file mode 100644 index 00000000000..ae63945d73d Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/1-442b76c1-0432-4290-b9a5-0ba180d74a9f.txn differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/2-67e892d9-dadf-43d3-bee7-ab6698b37177.txn b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/2-67e892d9-dadf-43d3-bee7-ab6698b37177.txn new file mode 100644 index 00000000000..b9c35fb4f4c Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/2-67e892d9-dadf-43d3-bee7-ab6698b37177.txn differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/3-fc6b6d69-d8f2-49ed-88e1-66eedf788abe.txn b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/3-fc6b6d69-d8f2-49ed-88e1-66eedf788abe.txn new file mode 100644 index 00000000000..d7112b95219 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/3-fc6b6d69-d8f2-49ed-88e1-66eedf788abe.txn differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/4-28073150-91be-40c6-89de-12d2f3df294b.txn b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/4-28073150-91be-40c6-89de-12d2f3df294b.txn new file mode 100644 index 00000000000..5a2974434e3 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_transactions/4-28073150-91be-40c6-89de-12d2f3df294b.txn differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551610.manifest b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551610.manifest new file mode 100644 index 00000000000..a63257f841d Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551610.manifest differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551611.manifest b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551611.manifest new file mode 100644 index 00000000000..f7f3b0797f5 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551611.manifest differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551612.manifest b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551612.manifest new file mode 100644 index 00000000000..18a55c782d9 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551612.manifest differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551613.manifest b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551613.manifest new file mode 100644 index 00000000000..c64725c7bba Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551613.manifest differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551614.manifest b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551614.manifest new file mode 100644 index 00000000000..b5551071aaf Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/_versions/18446744073709551614.manifest differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/__manifest/data/11111001010111101010011142f3754cac8700dbea13187b61.lance b/test_data/v6.0.1/dir_manifest/namespace/__manifest/data/11111001010111101010011142f3754cac8700dbea13187b61.lance new file mode 100644 index 00000000000..6cf55d44a4f Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/__manifest/data/11111001010111101010011142f3754cac8700dbea13187b61.lance differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/_transactions/0-0fe569c3-9e67-4e19-8ad5-3892453cfd91.txn b/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/_transactions/0-0fe569c3-9e67-4e19-8ad5-3892453cfd91.txn new file mode 100644 index 00000000000..66e847d432a Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/_transactions/0-0fe569c3-9e67-4e19-8ad5-3892453cfd91.txn differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/_versions/18446744073709551614.manifest b/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/_versions/18446744073709551614.manifest new file mode 100644 index 00000000000..ec2c14ec8a4 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/_versions/18446744073709551614.manifest differ diff --git a/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/data/1100011110010100001111006f4e6e40fea5335ff45a638719.lance b/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/data/1100011110010100001111006f4e6e40fea5335ff45a638719.lance new file mode 100644 index 00000000000..098855a6e68 Binary files /dev/null and b/test_data/v6.0.1/dir_manifest/namespace/def4bfb4_t1/data/1100011110010100001111006f4e6e40fea5335ff45a638719.lance differ