diff --git a/protos/table.proto b/protos/table.proto index 1de597938a9..e9722a780e5 100644 --- a/protos/table.proto +++ b/protos/table.proto @@ -125,6 +125,9 @@ message Manifest { // merely-carried column with an index keyed on a different column. Writers must // refuse it too: one that treats every entry of fields as keyed would maintain // the index against the wrong dependency set. + // * 1 << 8: reserved for datasets that may reference recognized V2 data files + // with different exact versions. Implementations that do not support the + // per-file exact-version contract must treat this bit as unknown. uint64 reader_feature_flags = 9; // Feature flags for writers. diff --git a/python/python/tests/test_dataset.py b/python/python/tests/test_dataset.py index 6edac488349..14af38ee180 100644 --- a/python/python/tests/test_dataset.py +++ b/python/python/tests/test_dataset.py @@ -5448,7 +5448,12 @@ def _write_overlay_file( ) -def test_data_overlay_dense(tmp_path: Path): +@pytest.fixture +def enable_unstable_data_overlay_files(monkeypatch): + monkeypatch.setenv("LANCE_ENABLE_UNSTABLE_DATA_OVERLAY_FILES", "1") + + +def test_data_overlay_dense(tmp_path: Path, enable_unstable_data_overlay_files): base_dir = tmp_path / "test" table = pa.table( { @@ -5480,7 +5485,7 @@ def test_data_overlay_dense(tmp_path: Path): assert result.column("id").to_pylist() == list(range(10)) -def test_data_overlay_newest_wins(tmp_path: Path): +def test_data_overlay_newest_wins(tmp_path: Path, enable_unstable_data_overlay_files): base_dir = tmp_path / "test" table = pa.table( { @@ -5534,7 +5539,9 @@ def test_data_overlay_newest_wins(tmp_path: Path): assert val[4] == 444 # only the older overlay covers offset 4 -def test_data_overlay_sparse_per_field(tmp_path: Path): +def test_data_overlay_sparse_per_field( + tmp_path: Path, enable_unstable_data_overlay_files +): base_dir = tmp_path / "test" table = pa.table( { @@ -5574,7 +5581,9 @@ def test_data_overlay_sparse_per_field(tmp_path: Path): assert result.column("val").to_pylist()[2] == 20 -def test_data_overlay_round_trips_through_fragment_metadata(tmp_path: Path): +def test_data_overlay_round_trips_through_fragment_metadata( + tmp_path: Path, enable_unstable_data_overlay_files +): import json base_dir = tmp_path / "test" @@ -5627,7 +5636,9 @@ def test_data_overlay_round_trips_through_fragment_metadata(tmp_path: Path): assert result.column("id").to_pylist() == list(range(10)) -def test_data_overlay_rejects_invalid_offsets(tmp_path: Path): +def test_data_overlay_rejects_invalid_offsets( + tmp_path: Path, enable_unstable_data_overlay_files +): base_dir = tmp_path / "test" table = pa.table({"val": pa.array([0, 1, 2], pa.int32())}) dataset = lance.write_dataset(table, base_dir) @@ -5669,7 +5680,9 @@ def test_data_overlay_rejects_invalid_offsets(tmp_path: Path): [[1, 1]], # sparse, duplicate ], ) -def test_data_overlay_rejects_unsorted_offsets(tmp_path: Path, offsets): +def test_data_overlay_rejects_unsorted_offsets( + tmp_path: Path, offsets, enable_unstable_data_overlay_files +): # Offsets map positionally to value rows in data_file. A RoaringBitmap would # silently reorder/dedup them, so a non-ascending list must be rejected up # front rather than corrupting the row mapping. diff --git a/rust/lance-namespace-impls/src/dir/manifest.rs b/rust/lance-namespace-impls/src/dir/manifest.rs index c7fb3699d07..4c0d915dfd1 100644 --- a/rust/lance-namespace-impls/src/dir/manifest.rs +++ b/rust/lance-namespace-impls/src/dir/manifest.rs @@ -55,7 +55,7 @@ use lance_namespace::models::{ TableExistsRequest, }; use lance_namespace::schema::arrow_schema_to_json; -use lance_table::feature_flags::apply_feature_flags; +use lance_table::feature_flags::{apply_feature_flags, ensure_can_write_manifest}; use lance_table::format::{Fragment, IndexMetadata, Manifest}; use lance_table::io::commit::{ CommitError, CommitHandler, commit_handler_from_url, write_manifest_file_to_path, @@ -1840,6 +1840,7 @@ impl ManifestNamespace { indices: Option>, transaction: Transaction, ) -> std::result::Result<(), CommitError> { + ensure_can_write_manifest(manifest).map_err(CommitError::from)?; apply_feature_flags(manifest, false, false).map_err(CommitError::from)?; let timestamp_nanos = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -1932,6 +1933,7 @@ impl ManifestNamespace { /// concurrent upgrade in between is still caught. async fn ensure_manifest_writable(&self) -> Result<()> { let dataset_guard = self.manifest_dataset.get().await?; + ensure_can_write_manifest(dataset_guard.manifest())?; ensure_writable(dataset_guard.metadata()) } @@ -1952,10 +1954,11 @@ impl ManifestNamespace { loop { let dataset_guard = self.manifest_dataset.get_refreshed().await?; + ensure_can_write_manifest(dataset_guard.manifest())?; let dataset = Arc::new(dataset_guard.clone()); drop(dataset_guard); - // Refuse to mutate a manifest written with a writer feature flag this - // build does not understand. + // The namespace format has its own capabilities in table metadata, + // separate from the Lance manifest capabilities checked above. ensure_writable(dataset.metadata())?; // Staged files, indices, the commit, and cleanup must all use the dataset's // own object store (see `commit_manifest_overwrite`). @@ -3858,6 +3861,7 @@ mod tests { CreateNamespaceRequest, CreateTableRequest, DescribeTableRequest, DropTableRequest, ListTablesRequest, TableExistsRequest, }; + use lance_table::feature_flags::FLAG_UNKNOWN; use lance_table::format::Fragment; use rstest::rstest; use std::collections::{HashMap, HashSet}; @@ -4390,6 +4394,64 @@ mod tests { ); } + #[tokio::test] + async fn test_manifest_rewrite_rejects_unknown_writer_flag_before_staging() { + let temp_dir = TempStdDir::default(); + let temp_path = temp_dir.to_str().unwrap(); + let manifest_ns = create_manifest_namespace(temp_path, false).await; + let data_paths_before = manifest_data_paths(&manifest_ns).await; + let original_version = { + let mut dataset = manifest_ns.manifest_dataset.get_mut().await.unwrap(); + let mut manifest = dataset.manifest().clone(); + manifest.writer_feature_flags |= FLAG_UNKNOWN << 1; + let version = manifest.version; + dataset.manifest = Arc::new(manifest); + version + }; + + let entries_before = dir_entry_names(temp_path); + let mut create_request = CreateTableRequest::new(); + create_request.id = Some(vec!["new_table".to_string()]); + let error = manifest_ns + .create_table(create_request, Bytes::from(create_test_ipc_data())) + .await + .unwrap_err(); + assert!( + error.to_string().to_lowercase().contains("upgrade"), + "expected an upgrade error, got: {error}" + ); + assert_eq!(dir_entry_names(temp_path), entries_before); + + let error = manifest_ns + .insert_into_manifest_with_metadata( + vec![ManifestEntry { + object_id: "table".to_string(), + object_type: ObjectType::Table, + location: Some("table.lance".to_string()), + metadata: None, + }], + None, + ) + .await + .unwrap_err(); + + assert!( + error.to_string().to_lowercase().contains("upgrade"), + "expected an upgrade error, got: {error}" + ); + assert_eq!( + manifest_ns + .manifest_dataset + .get() + .await + .unwrap() + .version() + .version, + original_version + ); + assert_eq!(manifest_data_paths(&manifest_ns).await, data_paths_before); + } + #[tokio::test] async fn test_manifest_noop_delete_uses_latest_snapshot() { let temp_dir = TempStdDir::default(); diff --git a/rust/lance-table/src/feature_flags.rs b/rust/lance-table/src/feature_flags.rs index 21568e3008b..bdeb439d479 100644 --- a/rust/lance-table/src/feature_flags.rs +++ b/rust/lance-table/src/feature_flags.rs @@ -50,16 +50,21 @@ pub const FLAG_UNSTABLE_DATA_OVERLAY_FILES: u64 = 64; /// that exposure comes with the reclamation and is inherited by whichever flag /// takes the bit. pub const FLAG_COVERED_INDEX_METADATA: u64 = 128; +/// Reserved for datasets that reference recognized V2 data files with +/// different exact versions. +pub const FLAG_MIXED_DATA_FILE_VERSIONS: u64 = 256; /// The first bit that is unknown as a feature flag -pub const FLAG_UNKNOWN: u64 = 256; +pub const FLAG_UNKNOWN: u64 = FLAG_MIXED_DATA_FILE_VERSIONS; -// The highest flag allocated must stay below the unknown boundary, or -// `supported_flags` would refuse a bit this code claims to understand. The next -// flag takes 256, so it has to move the boundary to 512 with it. +// Supported flags stay below the unknown boundary; the mixed-version bit is +// reserved at the boundary until its storage contract lands. const _: () = assert!(FLAG_COVERED_INDEX_METADATA < FLAG_UNKNOWN); // The fence needs a bit the current released build already refuses, which means // at or above the boundary that build shipped with (128). const _: () = assert!(FLAG_COVERED_INDEX_METADATA >= 128); +const _: () = assert!(FLAG_MIXED_DATA_FILE_VERSIONS == FLAG_UNKNOWN); + +pub(crate) const STICKY_PAIRED_FLAGS: u64 = FLAG_MIXED_DATA_FILE_VERSIONS; /// Environment variable that opts a release build into reading and writing data /// overlay files before the feature is generally released. @@ -78,6 +83,7 @@ pub fn apply_feature_flags( // immediately before the write. let covered_index_metadata = (manifest.reader_feature_flags | manifest.writer_feature_flags) & FLAG_COVERED_INDEX_METADATA; + let sticky_paired_flags = validated_sticky_paired_flags(manifest)?; // Reset flags manifest.reader_feature_flags = 0; @@ -139,10 +145,29 @@ pub fn apply_feature_flags( manifest.reader_feature_flags |= covered_index_metadata; manifest.writer_feature_flags |= covered_index_metadata; + manifest.reader_feature_flags |= sticky_paired_flags; + manifest.writer_feature_flags |= sticky_paired_flags; Ok(()) } +/// Carry sticky paired capabilities from the manifest a new one is derived +/// from. +/// +/// [`apply_feature_flags`] carries these bits across its own reset, but it only +/// ever sees one manifest. Constructors preserve these flags, and this helper +/// also validates that the source is not half-set before a derived manifest is +/// committed. +/// +/// A half-set state is refused rather than normalized: one bit set means a +/// legacy reader or a legacy writer is still permitted, which is neither mode. +pub fn inherit_sticky_feature_flags(destination: &mut Manifest, source: &Manifest) -> Result<()> { + let sticky_flags = validated_sticky_paired_flags(source)?; + destination.reader_feature_flags |= sticky_flags; + destination.writer_feature_flags |= sticky_flags; + Ok(()) +} + /// Whether this build understands data overlay files: always in debug builds, /// and in release builds only when [`ENABLE_UNSTABLE_DATA_OVERLAY_FILES_ENV`] is set. fn data_overlay_files_enabled() -> bool { @@ -183,10 +208,68 @@ pub fn can_write_dataset(writer_flags: u64) -> bool { writer_flags & !supported_flags() == 0 } +/// Refuse reads from manifests whose required reader features this build does +/// not support or whose paired capabilities are inconsistent. +pub fn ensure_can_read_manifest(manifest: &Manifest) -> Result<()> { + validate_paired_feature_flags(manifest)?; + if !can_read_dataset(manifest.reader_feature_flags) { + return Err(Error::not_supported_source( + format!( + "This dataset cannot be read by this version of Lance. Please upgrade \ + Lance to read this dataset. Flags: {}", + manifest.reader_feature_flags + ) + .into(), + )); + } + Ok(()) +} + +/// Refuse writes to manifests whose required writer features this build does +/// not support or whose paired capabilities are inconsistent. +pub fn ensure_can_write_manifest(manifest: &Manifest) -> Result<()> { + validate_paired_feature_flags(manifest)?; + if !can_write_dataset(manifest.writer_feature_flags) { + return Err(Error::not_supported_source( + format!( + "This dataset cannot be written by this version of Lance. Please upgrade \ + Lance to write this dataset. Flags: {}", + manifest.writer_feature_flags + ) + .into(), + )); + } + Ok(()) +} + pub fn has_deprecated_v2_feature_flag(writer_flags: u64) -> bool { writer_flags & FLAG_USE_V2_FORMAT_DEPRECATED != 0 } +/// Refuse a manifest whose paired reader and writer capability bits disagree. +/// +/// One word set and the other not is neither mode: it would let a legacy reader +/// or a legacy writer through on a table where the other half is enforcing. The +/// commit path refuses to *produce* this, so seeing it on read means the +/// manifest was written by something that did not. +pub fn validate_paired_feature_flags(manifest: &Manifest) -> Result<()> { + let reader = manifest.reader_feature_flags & FLAG_MIXED_DATA_FILE_VERSIONS != 0; + let writer = manifest.writer_feature_flags & FLAG_MIXED_DATA_FILE_VERSIONS != 0; + if reader != writer { + return Err(Error::corrupt_file_named( + "manifest", + "Manifest has only one of the mixed data-file-version reader and writer feature bits set, \ + so its semantics are undefined", + )); + } + Ok(()) +} + +fn validated_sticky_paired_flags(manifest: &Manifest) -> Result { + validate_paired_feature_flags(manifest)?; + Ok(manifest.reader_feature_flags & STICKY_PAIRED_FLAGS) +} + #[cfg(test)] mod tests { /// The covering fence only works if the bit is one the current released @@ -365,4 +448,107 @@ mod tests { 0 ); } + #[test] + fn inheriting_carries_sticky_paired_bits_from_the_source() { + let mut source = empty_manifest(); + source.reader_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + source.writer_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + // A fresh destination models any derived manifest before inheritance. + let mut destination = empty_manifest(); + + inherit_sticky_feature_flags(&mut destination, &source).unwrap(); + + assert_ne!( + destination.reader_feature_flags & FLAG_MIXED_DATA_FILE_VERSIONS, + 0 + ); + assert_ne!( + destination.writer_feature_flags & FLAG_MIXED_DATA_FILE_VERSIONS, + 0 + ); + } + + #[test] + fn inheriting_refuses_a_half_set_source() { + for (reader, writer) in [ + (FLAG_MIXED_DATA_FILE_VERSIONS, 0), + (0, FLAG_MIXED_DATA_FILE_VERSIONS), + ] { + let mut source = empty_manifest(); + source.reader_feature_flags = reader; + source.writer_feature_flags = writer; + let mut destination = empty_manifest(); + + let err = inherit_sticky_feature_flags(&mut destination, &source).unwrap_err(); + + assert!(err.to_string().contains("only one of"), "{err}"); + } + } + + #[test] + fn apply_feature_flags_carries_sticky_paired_bits_across_its_reset() { + let mut manifest = empty_manifest(); + manifest.reader_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + manifest.writer_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + + apply_feature_flags(&mut manifest, false, false).unwrap(); + + assert_ne!( + manifest.reader_feature_flags & FLAG_MIXED_DATA_FILE_VERSIONS, + 0 + ); + assert_ne!( + manifest.writer_feature_flags & FLAG_MIXED_DATA_FILE_VERSIONS, + 0 + ); + } + + #[test] + fn apply_feature_flags_rejects_half_set_sticky_bits() { + let mut manifest = empty_manifest(); + manifest.reader_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + + let err = apply_feature_flags(&mut manifest, false, false).unwrap_err(); + + assert!(matches!(err, Error::CorruptFile { .. })); + assert!(err.to_string().contains("only one of"), "{err}"); + } + + #[test] + fn writer_gate_rejects_reserved_mixed_capability() { + let mut manifest = empty_manifest(); + manifest.reader_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + manifest.writer_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS; + + let err = ensure_can_write_manifest(&manifest).unwrap_err(); + assert!(matches!(err, Error::NotSupported { .. })); + assert!(err.to_string().contains("cannot be written"), "{err}"); + } + + fn empty_manifest() -> Manifest { + use crate::format::DataStorageFormat; + use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema}; + use lance_core::datatypes::Schema; + use std::collections::HashMap; + use std::sync::Arc; + + let arrow_schema = ArrowSchema::new(vec![ArrowField::new("i", DataType::Int32, false)]); + Manifest::new( + Schema::try_from(&arrow_schema).unwrap(), + Arc::new(vec![]), + DataStorageFormat::default(), + HashMap::new(), + ) + } + + /// A build that does not know the bit must refuse the table rather than + /// continue with legacy semantics. + #[test] + fn mixed_capability_remains_at_the_unknown_boundary() { + assert!(can_read_dataset(FLAG_COVERED_INDEX_METADATA)); + assert!(can_write_dataset(FLAG_COVERED_INDEX_METADATA)); + assert!(!can_read_dataset(FLAG_MIXED_DATA_FILE_VERSIONS)); + assert!(!can_write_dataset(FLAG_MIXED_DATA_FILE_VERSIONS)); + assert_eq!(FLAG_MIXED_DATA_FILE_VERSIONS, FLAG_UNKNOWN); + } } diff --git a/rust/lance-table/src/format/manifest.rs b/rust/lance-table/src/format/manifest.rs index 0e968ab9e7e..628e313a9a7 100644 --- a/rust/lance-table/src/format/manifest.rs +++ b/rust/lance-table/src/format/manifest.rs @@ -18,7 +18,7 @@ use std::ops::Range; use std::sync::Arc; use super::Fragment; -use crate::feature_flags::FLAG_COVERED_INDEX_METADATA; +use crate::feature_flags::{FLAG_COVERED_INDEX_METADATA, STICKY_PAIRED_FLAGS}; use crate::feature_flags::{FLAG_STABLE_ROW_IDS, has_deprecated_v2_feature_flag}; use crate::format::fragment::DataFileFieldInterner; use crate::format::pb; @@ -219,8 +219,8 @@ impl Manifest { index_section: None, // Caller should update index if they want to keep them. timestamp_nanos: 0, // This will be set on commit tag: None, - reader_feature_flags: 0, // These will be set on commit - writer_feature_flags: 0, // These will be set on commit + reader_feature_flags: previous.reader_feature_flags & STICKY_PAIRED_FLAGS, + writer_feature_flags: previous.writer_feature_flags & STICKY_PAIRED_FLAGS, max_fragment_id: previous.max_fragment_id, transaction_file: None, transaction_section: None, @@ -283,8 +283,12 @@ impl Manifest { // covering could then open it and read carried columns as keyed ones. // Kept unconditionally rather than derived from the cloned indexes: // over-fencing a clone is harmless, under-fencing one is not. - reader_feature_flags: self.reader_feature_flags & FLAG_COVERED_INDEX_METADATA, - writer_feature_flags: self.writer_feature_flags & FLAG_COVERED_INDEX_METADATA, + // Sticky capabilities are also retained because the clone keeps the + // source file identities that require them. + reader_feature_flags: self.reader_feature_flags + & (FLAG_COVERED_INDEX_METADATA | STICKY_PAIRED_FLAGS), + writer_feature_flags: self.writer_feature_flags + & (FLAG_COVERED_INDEX_METADATA | STICKY_PAIRED_FLAGS), max_fragment_id: self.max_fragment_id, transaction_file: Some(transaction_file), transaction_section: None, diff --git a/rust/lance-table/src/transaction/manifest_build.rs b/rust/lance-table/src/transaction/manifest_build.rs index 484df49c95f..9857d15aa9d 100644 --- a/rust/lance-table/src/transaction/manifest_build.rs +++ b/rust/lance-table/src/transaction/manifest_build.rs @@ -10,7 +10,10 @@ //! operation vocabulary it matches on, the index rules it applies, the row version //! metadata it stamps, the validation that runs before it. -use crate::feature_flags::{FLAG_COVERED_INDEX_METADATA, FLAG_STABLE_ROW_IDS, apply_feature_flags}; +use crate::feature_flags::{ + FLAG_COVERED_INDEX_METADATA, FLAG_STABLE_ROW_IDS, apply_feature_flags, + ensure_can_read_manifest, ensure_can_write_manifest, inherit_sticky_feature_flags, +}; use crate::format::overlay::TOMBSTONE_FIELD_ID; use crate::format::{ DataFile, DataStorageFormat, Fragment, IndexMetadata, Manifest, ManifestBuildConfig, @@ -100,6 +103,11 @@ impl Transaction { .resolve_version_location(base_path, version, &object_store.inner) .await?; let mut manifest = read_manifest(object_store, &location.path, location.size).await?; + // This read bypasses Dataset's feature gates. Refuse unsupported target + // manifests before apply_feature_flags can clear their unknown bits and + // republish the referenced files as legacy-compatible. + ensure_can_read_manifest(&manifest)?; + ensure_can_write_manifest(&manifest)?; manifest.set_timestamp(config.timestamp_nanos); manifest.transaction_file = Some(tx_path.to_string()); let indices = read_manifest_indexes(object_store, &location, &manifest).await?; @@ -117,6 +125,7 @@ impl Transaction { collide with ids this table has already used" ))); } + inherit_sticky_feature_flags(&mut manifest, current_manifest)?; Ok((manifest, indices)) } @@ -1301,11 +1310,10 @@ impl Transaction { // derived there. // // Derived fresh from `final_indices` on every commit, never inherited. - // Every manifest this reaches starts with both words zeroed -- `Manifest::new` - // and `new_from_previous` alike -- so there is no stale bit to clear, and - // dropping the last covering index lifts the fence by simply not setting - // it again. Inheriting it from the previous manifest instead would make - // the fence permanent. + // Every manifest this reaches starts without the covering bit, so there + // is no stale bit to clear. Dropping the last covering index lifts the + // fence by simply not setting it again. Inheriting it from the previous + // manifest instead would make the fence permanent. // // Both words: a reader that selects a vector index by membership of // `fields` would answer a query on a merely-carried column with an index @@ -1319,6 +1327,10 @@ impl Transaction { manifest.writer_feature_flags |= FLAG_COVERED_INDEX_METADATA; } + if let Some(current_manifest) = current_manifest { + inherit_sticky_feature_flags(&mut manifest, current_manifest)?; + } + manifest.set_timestamp(config.timestamp_nanos); manifest.update_max_fragment_id(); diff --git a/rust/lance/src/dataset.rs b/rust/lance/src/dataset.rs index 391881afd4b..0011b8fed8c 100644 --- a/rust/lance/src/dataset.rs +++ b/rust/lance/src/dataset.rs @@ -148,7 +148,10 @@ pub use lance_core::ROW_ID; use lance_core::box_error; use lance_index::scalar::lance_format::LanceIndexStore; use lance_namespace::models::{DeclareTableRequest, DescribeTableRequest}; -use lance_table::feature_flags::{apply_feature_flags, can_read_dataset}; +use lance_table::feature_flags::{ + apply_feature_flags, ensure_can_read_manifest, ensure_can_write_manifest, + validate_paired_feature_flags, +}; use lance_table::io::deletion::{DELETIONS_DIR, relative_deletion_file_path}; use lance_table::rowids::{RowIdSequence, write_row_ids}; pub use schema_evolution::{ @@ -764,14 +767,7 @@ impl Dataset { read_struct(object_reader.as_ref(), offset).await }?; - if !can_read_dataset(manifest.reader_feature_flags) { - let message = format!( - "This dataset cannot be read by this version of Lance. \ - Please upgrade Lance to read this dataset.\n Flags: {}", - manifest.reader_feature_flags - ); - return Err(Error::not_supported_source(message.into())); - } + ensure_can_read_manifest(&manifest)?; // If indices were also in the last block, we can take the opportunity to // decode them now and cache them. @@ -844,6 +840,7 @@ impl Dataset { e_tag: manifest_location.e_tag.as_deref(), }; if let Some(cached) = metadata_cache.get_with_key(&manifest_key).await { + ensure_can_read_manifest(&cached)?; return Ok(cached); } let loaded = @@ -1203,35 +1200,13 @@ impl Dataset { .resolve_latest_location(&self.base, &self.object_store) .await?; - // Check if manifest is in cache before reading from storage - let manifest_key = ManifestKey { - version: location.version, - e_tag: location.e_tag.as_deref(), - }; - let cached_manifest = self.metadata_cache.get_with_key(&manifest_key).await; - if let Some(cached_manifest) = cached_manifest { - return Ok((cached_manifest, location)); - } - if self.already_checked_out(&location, self.manifest.branch.as_deref()) { + ensure_can_read_manifest(&self.manifest)?; return Ok((self.manifest.clone(), self.manifest_location.clone())); } - let mut manifest = read_manifest(&self.object_store, &location.path, location.size).await?; - if manifest.schema.has_dictionary_types() { - let reader = if let Some(size) = location.size { - self.object_store - .open_with_size(&location.path, size as usize) - .await? - } else { - self.object_store.open(&location.path).await? - }; - populate_manifest_schema_dictionaries(&mut manifest, reader.as_ref()).await?; - } - let manifest_arc = Arc::new(manifest); - self.metadata_cache - .insert_with_key(&manifest_key, manifest_arc.clone()) - .await; - Ok((manifest_arc, location)) + let manifest = + Self::get_manifest(&self.object_store, &location, &self.uri, &self.session).await?; + Ok((manifest, location)) } /// Read the transaction file for this version of the dataset. @@ -3303,6 +3278,7 @@ impl Dataset { // Resolve source dataset and its manifest using checkout_version let src_ds = self.checkout_version(version).await?; + ensure_can_write_manifest(&src_ds.manifest)?; let src_paths = src_ds.collect_paths().await?; // Prepare target object store and base path @@ -4106,6 +4082,7 @@ pub(crate) async fn write_manifest_file( naming_scheme: ManifestNamingScheme, transaction: Option, ) -> std::result::Result { + validate_paired_feature_flags(manifest)?; if config.auto_set_feature_flags { // build_manifest may have already set FLAG_STABLE_ROW_IDS on the manifest. // Preserve it here so this second apply_feature_flags call does not clear it diff --git a/rust/lance/src/dataset/tests/dataset_io.rs b/rust/lance/src/dataset/tests/dataset_io.rs index 14c87dc17f2..2987c01cef1 100644 --- a/rust/lance/src/dataset/tests/dataset_io.rs +++ b/rust/lance/src/dataset/tests/dataset_io.rs @@ -13,11 +13,12 @@ use super::dataset_common::{create_file, require_send}; use crate::dataset::WriteDestination; use crate::dataset::WriteMode::Overwrite; use crate::dataset::builder::DatasetBuilder; +use crate::dataset::transaction::Operation; use crate::dataset::{ManifestWriteConfig, validate_dataset_root_for_drop, write_manifest_file}; use crate::session::Session; use crate::session::caches::ManifestKey; use crate::{Dataset, Error, Result}; -use lance_table::format::{DataStorageFormat, Fragment}; +use lance_table::format::DataStorageFormat; use crate::dataset::write::{CommitBuilder, InsertBuilder, WriteMode, WriteParams}; use arrow::array::as_struct_array; @@ -43,7 +44,7 @@ use lance_file::{ }; use lance_io::assert_io_eq; use lance_table::feature_flags; -use lance_table::format::BasePath; +use lance_table::format::{BasePath, Fragment}; use object_store::ObjectStoreExt; use crate::index::DatasetIndexExt; @@ -1352,6 +1353,110 @@ async fn test_write_manifest( assert!(matches!(write_result, Err(Error::NotSupported { .. }))); } +#[tokio::test] +async fn test_restore_rejects_unknown_target_flags() { + let test_uri = TempStrDir::default(); + let data = gen_batch() + .col("i", array::step::()) + .into_reader_rows(RowCount::from(1), BatchCount::from(1)); + let dataset = Dataset::write(data, &test_uri, None).await.unwrap(); + + let write_config = ManifestWriteConfig { + auto_set_feature_flags: false, + ..Default::default() + }; + let mut unknown_manifest = dataset.manifest.as_ref().clone(); + unknown_manifest.version = 2; + unknown_manifest.reader_feature_flags |= feature_flags::FLAG_UNKNOWN; + unknown_manifest.writer_feature_flags |= feature_flags::FLAG_UNKNOWN; + write_manifest_file( + dataset.object_store.as_ref(), + dataset.commit_handler.as_ref(), + &dataset.base, + &mut unknown_manifest, + None, + &write_config, + dataset.manifest_location.naming_scheme, + None, + ) + .await + .unwrap(); + + let mut supported_manifest = dataset.manifest.as_ref().clone(); + supported_manifest.version = 3; + write_manifest_file( + dataset.object_store.as_ref(), + dataset.commit_handler.as_ref(), + &dataset.base, + &mut supported_manifest, + None, + &write_config, + dataset.manifest_location.naming_scheme, + None, + ) + .await + .unwrap(); + + let error = Dataset::commit( + &test_uri, + Operation::Restore { version: 2 }, + Some(3), + None, + None, + Default::default(), + false, + ) + .await + .unwrap_err(); + + assert!(matches!(error, Error::NotSupported { .. }), "{error}"); +} + +#[tokio::test] +async fn test_checkout_latest_rejects_unsupported_reader_before_caching() { + let test_uri = TempStrDir::default(); + let data = gen_batch() + .col("i", array::step::()) + .into_reader_rows(RowCount::from(1), BatchCount::from(1)); + let mut dataset = Dataset::write(data, &test_uri, None).await.unwrap(); + let original_version = dataset.version().version; + + let mut unsupported_manifest = dataset.manifest.as_ref().clone(); + unsupported_manifest.version += 1; + unsupported_manifest.reader_feature_flags |= feature_flags::FLAG_UNKNOWN; + unsupported_manifest.writer_feature_flags |= feature_flags::FLAG_UNKNOWN; + let location = write_manifest_file( + dataset.object_store.as_ref(), + dataset.commit_handler.as_ref(), + &dataset.base, + &mut unsupported_manifest, + None, + &ManifestWriteConfig { + auto_set_feature_flags: false, + ..Default::default() + }, + dataset.manifest_location.naming_scheme, + None, + ) + .await + .unwrap(); + + let error = dataset.checkout_latest().await.unwrap_err(); + assert!(matches!(error, Error::NotSupported { .. }), "{error}"); + assert_eq!(dataset.version().version, original_version); + assert!( + dataset + .metadata_cache + .get_with_key(&ManifestKey { + version: location.version, + e_tag: location.e_tag.as_deref(), + }) + .await + .is_none(), + "unsupported manifest must not be cached" + ); +} + #[tokio::test] async fn test_rle_v2_v23_write_and_append() { let test_uri = TempStrDir::default(); @@ -1732,6 +1837,52 @@ async fn test_deep_clone( assert_eq!(count_files(store, &dst_root, "_deletions").await, 0); } +#[tokio::test] +async fn test_deep_clone_rejects_unsupported_writer_before_copying() { + let test_dir = TempStdDir::default(); + let source_dir = test_dir.join("source"); + let target_dir = test_dir.join("target"); + let mut source = Dataset::write( + gen_batch() + .col("id", array::step::()) + .into_reader_rows(RowCount::from(32), BatchCount::from(1)), + source_dir.to_str().unwrap(), + None, + ) + .await + .unwrap(); + + let mut unsupported_manifest = source.manifest.as_ref().clone(); + unsupported_manifest.version += 1; + unsupported_manifest.writer_feature_flags |= feature_flags::FLAG_UNKNOWN << 1; + write_manifest_file( + source.object_store.as_ref(), + source.commit_handler.as_ref(), + &source.base, + &mut unsupported_manifest, + None, + &ManifestWriteConfig { + auto_set_feature_flags: false, + ..Default::default() + }, + source.manifest_location.naming_scheme, + None, + ) + .await + .unwrap(); + + let error = source + .deep_clone( + target_dir.to_str().unwrap(), + unsupported_manifest.version, + None, + ) + .await + .unwrap_err(); + assert!(matches!(error, Error::NotSupported { .. })); + assert!(!target_dir.exists()); +} + #[tokio::test] async fn test_deep_clone_recognizes_ambiguous_commit_as_own() { use crate::utils::test::{AmbiguousCommitHandler, AmbiguousFailure}; diff --git a/rust/lance/src/io/commit.rs b/rust/lance/src/io/commit.rs index 47a81a647b9..e854ca007fb 100644 --- a/rust/lance/src/io/commit.rs +++ b/rust/lance/src/io/commit.rs @@ -34,6 +34,7 @@ use lance_file::version::LanceFileVersion; use lance_index::metrics::NoOpMetricsCollector; use lance_io::utils::CachedFileSize; use lance_select::RowAddrTreeMap; +use lance_table::feature_flags::ensure_can_write_manifest; use lance_table::format::{ DETACHED_VERSION_MASK, DeletionFile, Fragment, IndexMetadata, Manifest, WriterVersion, is_detached_version, list_index_files_with_sizes, pb, @@ -389,6 +390,7 @@ async fn do_commit_new_dataset( &Session::default(), ) .await?; + ensure_can_write_manifest(&source_manifest)?; if *is_shallow { let new_base_id = source_manifest @@ -1041,6 +1043,7 @@ pub(crate) async fn do_commit_detached_transaction( commit_config: &CommitConfig, retry_timeout: Duration, ) -> Result<(Manifest, ManifestLocation)> { + ensure_can_write_manifest(&dataset.manifest)?; let pb_transaction = pb::Transaction::from(transaction); let inline_transaction = pb_transaction.encoded_len() <= MAX_INLINE_TRANSACTION_BYTES; @@ -1373,6 +1376,8 @@ pub(crate) async fn commit_transaction( if !strict_overwrite { (dataset, other_transactions) = load_and_sort_new_transactions(&dataset).await?; + ensure_can_write_manifest(&dataset.manifest)?; + // See if we can retry the commit. Try to account for all // transactions that have been committed since the read_version. // Use small amount of backoff to handle transactions that all @@ -1386,6 +1391,8 @@ pub(crate) async fn commit_transaction( } transaction = rebase.finish(&dataset).await?; + } else { + ensure_can_write_manifest(&dataset.manifest)?; } // Recomputed every attempt: the rebase above may have rewritten the