lance-table 14.0.1

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The Lance Authors

//! Feature flags

use crate::format::Manifest;
use lance_core::{Error, Result, datatypes::Schema};

/// Fragments may contain deletion files, which record the tombstones of
/// soft-deleted rows.
pub const FLAG_DELETION_FILES: u64 = 1 << 0;
/// Row ids are stable for both moves and updates. Fragments contain an index
/// mapping row ids to row addresses.
pub const FLAG_STABLE_ROW_IDS: u64 = 1 << 1;
/// Files are written with the new v2 format (this flag is no longer used)
pub const FLAG_USE_V2_FORMAT_DEPRECATED: u64 = 1 << 2;
/// Table config is present
pub const FLAG_TABLE_CONFIG: u64 = 1 << 3;
/// Dataset uses multiple base paths (for shallow clones or multi-base datasets)
pub const FLAG_BASE_PATHS: u64 = 1 << 4;
/// Disable writing transaction file under _transaction/, this flag is set when we only want to write inline transaction in manifest
pub const FLAG_DISABLE_TRANSACTION_FILE: u64 = 1 << 5;
/// Fragments contain data overlay files, which supply new values for a subset of
/// cells without rewriting base data files. A reader that does not understand
/// overlays must refuse the dataset, since ignoring an overlay would silently
/// return stale base values.
///
/// Data overlay files are not yet a released feature: in release builds this flag
/// is treated as unknown (so a release reader/writer refuses an overlay dataset)
/// unless [`ENABLE_UNSTABLE_DATA_OVERLAY_FILES_ENV`] is set, which lets benchmarks opt in.
/// Debug builds always understand it so tests exercise the path.
pub const FLAG_UNSTABLE_DATA_OVERLAY_FILES: u64 = 1 << 6;
/// Some index declares covering columns: `IndexMetadata.covering_fields` names
/// columns the index carries values for, whether or not it is also keyed on
/// them.
///
/// Without [`FLAG_INDEPENDENT_COVERING_FIELDS`], the carried columns form the
/// trailing suffix of `fields` in emission order. A reader without this bit still selects
/// a vector index by testing membership of `fields`, so it would answer a query
/// on a merely-carried column with an index keyed on a different column and
/// return wrong neighbours with no error. A writer without it would maintain the
/// index as though every entry of `fields` were keyed. Both must refuse the table.
///
/// This takes the bit reclaimed from the retired MemWAL index-catchup flag
/// (<https://github.com/lance-format/lance/pull/8680>), which is the boundary the
/// current released build treats as unknown -- so that build refuses a covering
/// dataset without needing a change of its own. Builds from the window where the
/// bit was allocated to index catch-up (v11.0.0-beta.4 through beta.17) still
/// count it as supported and will open a covering dataset rather than refuse it;
/// that exposure comes with the reclamation and is inherited by whichever flag
/// takes the bit.
pub const FLAG_COVERED_INDEX_METADATA: u64 = 1 << 7;
/// A dataset may reference recognized V2 data files with different exact
/// versions. Readers and writers must both understand the per-file version
/// contract before either can safely access the dataset.
pub const FLAG_MIXED_DATA_FILE_VERSIONS: u64 = 1 << 8;
/// The table uses stable row ids and carries a fragment reuse index.
///
/// Understood only by debug builds until support is released (see
/// [`frag_reuse_with_stable_row_ids_enabled`]).
pub const FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS: u64 = 1 << 9;
/// Tagged FRI requires a reader that interprets its mappings and a writer that
/// preserves them during maintenance. Legacy-only FRI does not set this bit.
/// Bit 9 is taken by the stable-row-id FRI compatibility flag.
pub const FLAG_FRAGMENT_REUSE_INDEX: u64 = 1 << 10;
/// Some fragment stores a row lineage sequence -- its row ids, or its
/// created-at or last-updated-at versions -- as a hidden column of a data file
/// rather than inline in the manifest (`RowIdMeta::Column`,
/// `RowDatasetVersionMeta::Column`).
///
/// A reader without this bit sees an unset `row_id_sequence` oneof and would
/// take the fragment to have no row ids at all, on a table whose manifest says
/// every fragment has them; a writer without it would carry the fragment
/// forward and drop the sequence. Both must refuse the table.
///
/// Supported in every build. Table configuration controls whether a writer
/// spills sequences; readers must understand existing spilled sequences even
/// when spilling is disabled for subsequent writes.
///
/// ```
/// use lance_table::feature_flags::{FLAG_SPILLED_ROW_LINEAGE, can_read_dataset};
///
/// assert!(can_read_dataset(FLAG_SPILLED_ROW_LINEAGE));
/// ```
pub const FLAG_SPILLED_ROW_LINEAGE: u64 = 1 << 11;
/// Deprecated name for [`FLAG_SPILLED_ROW_LINEAGE`].
#[deprecated(since = "14.0.0", note = "use FLAG_SPILLED_ROW_LINEAGE instead")]
pub const FLAG_UNSTABLE_SPILLED_ROW_LINEAGE: u64 = FLAG_SPILLED_ROW_LINEAGE;
/// Reserved for fragment trees. Readers and writers reject it until implemented.
///
/// Bit 11 is spilled row lineage, so this takes the next free bit.
pub const FLAG_FRAGMENT_TREE: u64 = 1 << 12;
/// Index key fields and covering fields are declared independently.
///
/// When this flag is set, `IndexMetadata.fields` contains only the columns the
/// index is keyed on and `IndexMetadata.covering_fields` separately contains
/// the columns whose values the index carries. A field may occur in both lists.
/// The flag requires [`FLAG_COVERED_INDEX_METADATA`] in both feature words, and
/// the two are retained together while the independent bit is set.
///
/// Reserved ahead of its implementation. This build masks the bit out in
/// `supported_flags_when`, so it cannot open a table and apply the legacy
/// suffix contract to independent declarations.
pub const FLAG_INDEPENDENT_COVERING_FIELDS: u64 = 1 << 13;
/// Blob v2 descriptors may independently address Lance-owned objects. Readers
/// must resolve their explicit bases and writers/GC must preserve those references.
/// This capability is sticky, including across restore, and requires both words.
pub const FLAG_MANAGED_BLOBS: u64 = 1 << 14;

/// Schema metadata that explicitly opts writes and compaction into Managed blobs.
/// Set `lance.blob.managed` to `true` to activate the sticky table capability.
/// This requires Managed-aware readers and maintenance clients thereafter.
pub const MANAGED_BLOBS_META_KEY: &str = "lance.blob.managed";

/// Field IDs are allocated from a persistent high-water mark and are never reused.
/// Writers must understand this allocation contract. It does not change how
/// readers interpret the schema or data files.
pub const FLAG_NON_REUSABLE_FIELD_IDS: u64 = 1 << 15;
/// Required for data files containing version 1 zone-level statistics seeds.
/// Their min/max exclude NaNs, which are counted separately by sign. Set in
/// both reader and writer feature flags and preserve across commits.
///
/// Reserved: support for these seeds is not implemented yet.
pub const FLAG_ZONE_WRITE_SEEDS: u64 = 1 << 16;
/// Writer-only protection for clustering definitions, ID allocation, references,
/// and generic fragment/index segment metadata maintenance.
/// Reserved and unsupported until writers implement its maintenance rules.
/// Ordinary readers can ignore this metadata; all covered features share this one bit.
pub const FLAG_OBJECT_METADATA: u64 = 1 << 17;
/// The first bit that is unknown as a feature flag
pub const FLAG_UNKNOWN: u64 = 1 << 18;

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 (bit 7).
const _: () = assert!(FLAG_COVERED_INDEX_METADATA >= 1 << 7);
const _: () = assert!(FLAG_MIXED_DATA_FILE_VERSIONS < FLAG_UNKNOWN);
const _: () = assert!(FLAG_NON_REUSABLE_FIELD_IDS < FLAG_UNKNOWN);
// Same fence for this bit: v12.0.0 refuses bit 9 and up.
const _: () = assert!(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS >= 1 << 9);
const _: () = assert!(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS < FLAG_UNKNOWN);
const _: () = assert!(FLAG_FRAGMENT_REUSE_INDEX < FLAG_UNKNOWN);
const _: () = assert!(FLAG_SPILLED_ROW_LINEAGE < FLAG_UNKNOWN);
const _: () = assert!(FLAG_FRAGMENT_TREE < FLAG_UNKNOWN);
const _: () = assert!(FLAG_INDEPENDENT_COVERING_FIELDS < FLAG_UNKNOWN);

const _: () = assert!(FLAG_MANAGED_BLOBS < FLAG_UNKNOWN);
const _: () = assert!(FLAG_ZONE_WRITE_SEEDS < FLAG_UNKNOWN);
const _: () = assert!(FLAG_OBJECT_METADATA < FLAG_UNKNOWN);

pub(crate) const STICKY_PAIRED_FLAGS: u64 = FLAG_MIXED_DATA_FILE_VERSIONS
    | FLAG_FRAGMENT_REUSE_INDEX
    | FLAG_MANAGED_BLOBS
    | FLAG_ZONE_WRITE_SEEDS;
pub(crate) const STICKY_READER_FLAGS: u64 = STICKY_PAIRED_FLAGS;
pub(crate) const STICKY_WRITER_FLAGS: u64 = STICKY_PAIRED_FLAGS | FLAG_NON_REUSABLE_FIELD_IDS;

/// Environment variable that opts a release build into reading and writing data
/// overlay files before the feature is generally released.
pub const ENABLE_UNSTABLE_DATA_OVERLAY_FILES_ENV: &str = "LANCE_ENABLE_UNSTABLE_DATA_OVERLAY_FILES";

/// Set the reader and writer feature flags in the manifest based on the contents of the manifest.
pub fn apply_feature_flags(
    manifest: &mut Manifest,
    enable_stable_row_id: bool,
    disable_transaction_file: bool,
) -> Result<()> {
    // Carried across the reset: a `Manifest` only points at its index section,
    // so whether any index declares covering columns is not visible here.
    // `build_manifest` decides it from the index list it is committing and sets
    // the bit after calling this; without the carry the second call, from
    // `write_manifest_file`, would clear that decision 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)?;
    let non_reusable_field_ids = manifest.max_allocated_field_id.is_some();
    if non_reusable_field_ids {
        manifest.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
    }
    validate_non_reusable_field_id_flags(manifest)?;
    // Reset flags
    manifest.reader_feature_flags = 0;
    manifest.writer_feature_flags = 0;

    let has_deletion_files = manifest
        .fragments
        .iter()
        .any(|frag| frag.deletion_file.is_some());
    if has_deletion_files {
        // Both readers and writers need to be able to read deletion files
        manifest.reader_feature_flags |= FLAG_DELETION_FILES;
        manifest.writer_feature_flags |= FLAG_DELETION_FILES;
    }

    // If any fragment has row ids, they must all have row ids.
    let has_row_ids = manifest
        .fragments
        .iter()
        .any(|frag| frag.row_id_meta.is_some());
    if has_row_ids || enable_stable_row_id {
        if !manifest
            .fragments
            .iter()
            .all(|frag| frag.row_id_meta.is_some())
        {
            return Err(Error::invalid_input("All fragments must have row ids"));
        }
        manifest.reader_feature_flags |= FLAG_STABLE_ROW_IDS;
        manifest.writer_feature_flags |= FLAG_STABLE_ROW_IDS;
    }

    // Test whether any table metadata has been set
    if !manifest.config.is_empty() {
        manifest.writer_feature_flags |= FLAG_TABLE_CONFIG;
    }

    // Check if this dataset uses multiple base paths (for shallow clones or multi-base datasets)
    if !manifest.base_paths.is_empty() {
        manifest.reader_feature_flags |= FLAG_BASE_PATHS;
        manifest.writer_feature_flags |= FLAG_BASE_PATHS;
    }

    // Overlay files change cell values on read, so a reader that ignores them
    // would return stale base values. Both readers and writers must understand
    // them.
    let has_overlays = manifest
        .fragments
        .iter()
        .any(|frag| !frag.overlays.is_empty());
    if has_overlays {
        manifest.reader_feature_flags |= FLAG_UNSTABLE_DATA_OVERLAY_FILES;
        manifest.writer_feature_flags |= FLAG_UNSTABLE_DATA_OVERLAY_FILES;
    }

    let has_spilled_row_lineage = manifest
        .fragments
        .iter()
        .any(|frag| frag.has_spilled_row_lineage());
    if has_spilled_row_lineage {
        manifest.reader_feature_flags |= FLAG_SPILLED_ROW_LINEAGE;
        manifest.writer_feature_flags |= FLAG_SPILLED_ROW_LINEAGE;
    }

    if disable_transaction_file {
        manifest.writer_feature_flags |= FLAG_DISABLE_TRANSACTION_FILE;
    }

    if non_reusable_field_ids {
        manifest.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
    }

    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 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.
///
/// Non-reusable field IDs are activated explicitly and only require writer support.
pub fn inherit_sticky_feature_flags(destination: &mut Manifest, source: &Manifest) -> Result<()> {
    let sticky_flags = validated_sticky_paired_flags(source)?;
    validate_non_reusable_field_id_flags(source)?;
    destination.reader_feature_flags |= sticky_flags;
    destination.writer_feature_flags |=
        sticky_flags | (source.writer_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS);
    Ok(())
}

/// Whether a schema explicitly requests Managed blob storage. Absent or `false`
/// keeps the existing Inline/Packed/Dedicated storage layout, in all build profiles.
///
/// ```
/// use lance_core::datatypes::Schema;
/// use lance_table::feature_flags::{MANAGED_BLOBS_META_KEY, managed_blobs_enabled};
/// let mut schema = Schema::default();
/// assert!(!managed_blobs_enabled(&schema)?);
/// schema.metadata.insert(MANAGED_BLOBS_META_KEY.to_string(), "true".to_string());
/// assert!(managed_blobs_enabled(&schema)?);
/// # Ok::<(), lance_core::Error>(())
/// ```
pub fn managed_blobs_enabled(schema: &Schema) -> Result<bool> {
    match schema
        .metadata
        .get(MANAGED_BLOBS_META_KEY)
        .map(String::as_str)
    {
        None | Some("false") => Ok(false),
        Some("true") => Ok(true),
        Some(value) => Err(Error::invalid_input(format!(
            "Schema metadata {MANAGED_BLOBS_META_KEY} must be true or false, got {value:?}"
        ))),
    }
}

/// 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 {
    cfg!(debug_assertions) || std::env::var_os(ENABLE_UNSTABLE_DATA_OVERLAY_FILES_ENV).is_some()
}

/// Debug builds only, with no environment override, until continuous recording
/// and retention land. Hidden because it is removed then.
#[doc(hidden)]
pub fn frag_reuse_with_stable_row_ids_enabled() -> bool {
    cfg!(debug_assertions)
}

/// Clear `flag` from `flags` when its gating feature is not enabled in this
/// build; leave it set otherwise. One call per unstable flag, so support for
/// several unstable features chains cleanly.
fn mark_supported(flags: &mut u64, flag: u64, feature_enabled: bool) {
    if !feature_enabled {
        *flags &= !flag;
    }
}

/// The feature-flag bits this build understands, given which unstable features
/// are enabled. Split out from [`supported_flags`] so the policy is testable
/// without toggling the build profile or environment.
fn supported_flags_when(
    overlay_enabled: bool,
    frag_reuse_with_stable_row_ids_enabled: bool,
) -> u64 {
    let mut supported = FLAG_UNKNOWN - 1;
    mark_supported(
        &mut supported,
        FLAG_UNSTABLE_DATA_OVERLAY_FILES,
        overlay_enabled,
    );
    mark_supported(
        &mut supported,
        FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS,
        frag_reuse_with_stable_row_ids_enabled,
    );
    // Reserved, not implemented: see the flag's doc comment.
    mark_supported(&mut supported, FLAG_FRAGMENT_TREE, false);
    mark_supported(&mut supported, FLAG_INDEPENDENT_COVERING_FIELDS, false);
    mark_supported(&mut supported, FLAG_ZONE_WRITE_SEEDS, false);
    mark_supported(&mut supported, FLAG_OBJECT_METADATA, false);
    supported
}

fn supported_flags() -> u64 {
    supported_flags_when(
        data_overlay_files_enabled(),
        frag_reuse_with_stable_row_ids_enabled(),
    )
}

pub fn can_read_dataset(reader_flags: u64) -> bool {
    reader_flags & !supported_flags() == 0
}

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)?;
    validate_non_reusable_field_id_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)?;
    validate_non_reusable_field_id_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<()> {
    if (manifest.reader_feature_flags ^ manifest.writer_feature_flags) & FLAG_FRAGMENT_REUSE_INDEX
        != 0
    {
        return Err(Error::corrupt_file_named(
            "manifest",
            "FRI requires both reader and writer feature flags",
        ));
    }

    for (flag, name) in [
        (FLAG_MIXED_DATA_FILE_VERSIONS, "mixed data-file-version"),
        (FLAG_MANAGED_BLOBS, "Managed Blob"),
        (FLAG_ZONE_WRITE_SEEDS, "zone write seed"),
    ] {
        let reader = manifest.reader_feature_flags & flag != 0;
        let writer = manifest.writer_feature_flags & flag != 0;
        if reader != writer {
            return Err(Error::corrupt_file_named(
                "manifest",
                format!(
                    "Manifest has only one of the {name} reader and writer feature bits set, so its semantics are undefined"
                ),
            ));
        }
    }
    Ok(())
}

/// Refuse a manifest whose non-reusable-field-ID marker and required flags disagree.
///
/// The high-water mark is the activation marker and always requires the writer
/// bit. Non-reusable field IDs do not change read semantics, so the reader bit is not
/// a valid activation mode.
pub fn validate_non_reusable_field_id_flags(manifest: &Manifest) -> Result<()> {
    let activated = manifest.max_allocated_field_id.is_some();
    let reader = manifest.reader_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS != 0;
    let writer = manifest.writer_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS != 0;
    if activated != writer {
        return Err(Error::corrupt_file_named(
            "manifest",
            "Manifest non-reusable-field-ID high-water mark and writer feature flag disagree",
        ));
    }
    if reader {
        return Err(Error::corrupt_file_named(
            "manifest",
            "Manifest has a non-reusable-field-ID reader feature flag, but non-reusable field IDs only require writer support",
        ));
    }
    Ok(())
}

fn validated_sticky_paired_flags(manifest: &Manifest) -> Result<u64> {
    validate_paired_feature_flags(manifest)?;
    Ok(manifest.reader_feature_flags & STICKY_PAIRED_FLAGS)
}

#[cfg(test)]
mod tests {
    #[rstest::rstest]
    #[case::absent(None, false)]
    #[case::disabled(Some("false"), false)]
    #[case::enabled(Some("true"), true)]
    fn managed_blobs_require_explicit_opt_in(#[case] value: Option<&str>, #[case] enabled: bool) {
        let mut schema = empty_manifest().schema;
        if let Some(value) = value {
            schema
                .metadata
                .insert(MANAGED_BLOBS_META_KEY.to_string(), value.to_string());
        }
        assert_eq!(managed_blobs_enabled(&schema).unwrap(), enabled);
    }

    #[test]
    fn managed_blobs_reject_invalid_opt_in() {
        let mut schema = empty_manifest().schema;
        schema
            .metadata
            .insert(MANAGED_BLOBS_META_KEY.to_string(), "invalid".to_string());
        let error = managed_blobs_enabled(&schema).unwrap_err();
        assert!(matches!(error, Error::InvalidInput { .. }));
        assert!(
            error
                .to_string()
                .contains("lance.blob.managed must be true or false")
        );
    }

    #[test]
    fn tagged_fri_flag_is_supported_and_sticky() {
        let flag = super::FLAG_FRAGMENT_REUSE_INDEX;
        assert!(super::can_read_dataset(flag));
        assert!(super::can_write_dataset(flag));
        let mut manifest = empty_manifest();
        manifest.reader_feature_flags = flag;
        assert!(super::ensure_can_read_manifest(&manifest).is_err());
        manifest.writer_feature_flags = flag;
        super::apply_feature_flags(&mut manifest, false, false).unwrap();
        assert_eq!(manifest.reader_feature_flags & flag, flag);
        assert_eq!(manifest.writer_feature_flags & flag, flag);
    }

    /// The covering fence only works if the bit is one the current released
    /// build already rejects. That build's unknown boundary is 128, so the bit
    /// has to be 128 and this build has to have moved its own boundary past it
    /// -- otherwise either that build accepts a covering dataset, or we refuse
    /// our own.
    #[test]
    fn test_covered_index_metadata_fences_older_builds_only() {
        assert_eq!(
            FLAG_COVERED_INDEX_METADATA, 128,
            "the fence must sit on the boundary the released build shipped with"
        );
        assert!(
            can_read_dataset(FLAG_COVERED_INDEX_METADATA),
            "this build implements covering, so it must accept its own datasets"
        );
        assert!(can_write_dataset(FLAG_COVERED_INDEX_METADATA));
        // A build whose boundary is still 128 refuses the bit, which is the fence;
        // the module-level `const _` assertion keeps it at or above that boundary.
    }

    use super::*;
    use crate::format::BasePath;

    #[test]
    fn test_object_metadata_flag_fences_writes_without_blocking_reads() {
        let mut manifest = empty_manifest();
        manifest.writer_feature_flags = FLAG_OBJECT_METADATA;
        ensure_can_read_manifest(&manifest).unwrap();
        let error = ensure_can_write_manifest(&manifest).unwrap_err();
        assert!(matches!(error, Error::NotSupported { .. }));
        assert!(error.to_string().contains("cannot be written"));
        assert!(!can_read_dataset(FLAG_OBJECT_METADATA));
        assert!(!can_write_dataset(FLAG_OBJECT_METADATA));
        assert_eq!(FLAG_OBJECT_METADATA, 1 << 17);
        // Neither reserved bit is supported, and bits above them remain unknown.
        assert!(!can_read_dataset(FLAG_ZONE_WRITE_SEEDS));
        assert!(!can_write_dataset(FLAG_ZONE_WRITE_SEEDS));
        assert_eq!(FLAG_UNKNOWN, 1 << 18);
        assert!(!can_read_dataset(FLAG_UNKNOWN));
        assert!(!can_write_dataset(FLAG_UNKNOWN));
    }

    #[test]
    fn test_zone_write_seed_flag_is_reserved_and_sticky() {
        assert_eq!(FLAG_ZONE_WRITE_SEEDS, 1 << 16);
        assert!(!can_read_dataset(FLAG_ZONE_WRITE_SEEDS));
        assert!(!can_write_dataset(FLAG_ZONE_WRITE_SEEDS));

        let mut manifest = empty_manifest();
        manifest.reader_feature_flags = FLAG_ZONE_WRITE_SEEDS;
        manifest.writer_feature_flags = FLAG_ZONE_WRITE_SEEDS;
        apply_feature_flags(&mut manifest, false, false).unwrap();
        assert_eq!(manifest.reader_feature_flags, FLAG_ZONE_WRITE_SEEDS);
        assert_eq!(manifest.writer_feature_flags, FLAG_ZONE_WRITE_SEEDS);

        let read_error = ensure_can_read_manifest(&manifest).unwrap_err();
        assert!(matches!(read_error, Error::NotSupported { .. }));
        assert!(read_error.to_string().contains("Please upgrade"));
        let write_error = ensure_can_write_manifest(&manifest).unwrap_err();
        assert!(matches!(write_error, Error::NotSupported { .. }));
        assert!(write_error.to_string().contains("Please upgrade"));

        let mut destination = empty_manifest();
        inherit_sticky_feature_flags(&mut destination, &manifest).unwrap();
        assert_eq!(destination.reader_feature_flags, FLAG_ZONE_WRITE_SEEDS);
        assert_eq!(destination.writer_feature_flags, FLAG_ZONE_WRITE_SEEDS);
    }

    #[test]
    fn test_zone_write_seeds_require_both_feature_bits() {
        for (reader, writer) in [(true, false), (false, true)] {
            let mut manifest = empty_manifest();
            manifest.reader_feature_flags = if reader { FLAG_ZONE_WRITE_SEEDS } else { 0 };
            manifest.writer_feature_flags = if writer { FLAG_ZONE_WRITE_SEEDS } else { 0 };
            let error = validate_paired_feature_flags(&manifest).unwrap_err();
            assert!(matches!(error, Error::CorruptFile { .. }));
            assert!(error.to_string().contains("zone write seed"));
        }
    }

    #[test]
    fn test_fragment_tree_flag_is_reserved_not_supported() {
        assert_eq!(FLAG_FRAGMENT_TREE, 4096);
        assert_eq!(
            FLAG_FRAGMENT_TREE & (FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS | FLAG_FRAGMENT_REUSE_INDEX),
            0
        );
        assert!(!can_read_dataset(FLAG_FRAGMENT_TREE));
        assert!(!can_write_dataset(FLAG_FRAGMENT_TREE));
    }

    #[test]
    fn test_frag_reuse_with_stable_row_ids_flag_release_gating() {
        // v12.0.0's unknown boundary, so every released client refuses it.
        assert_eq!(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS, 512);
        let without_support = supported_flags_when(true, false);
        assert_ne!(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS & !without_support, 0);
        assert_eq!(FLAG_STABLE_ROW_IDS & !without_support, 0);
        let enabled = supported_flags_when(false, true);
        assert_eq!(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS & !enabled, 0);
        for unrelated in [
            FLAG_UNSTABLE_DATA_OVERLAY_FILES,
            FLAG_FRAGMENT_TREE,
            FLAG_UNKNOWN,
        ] {
            assert_ne!(unrelated & !enabled, 0);
        }
        assert_eq!(
            can_read_dataset(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS),
            frag_reuse_with_stable_row_ids_enabled()
        );
        assert_eq!(
            can_write_dataset(FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS),
            frag_reuse_with_stable_row_ids_enabled()
        );
    }

    #[test]
    fn test_independent_covering_fields_flag_is_reserved() {
        assert_eq!(FLAG_INDEPENDENT_COVERING_FIELDS, 8192);
        let flags = FLAG_COVERED_INDEX_METADATA | FLAG_INDEPENDENT_COVERING_FIELDS;
        assert!(!can_read_dataset(flags));
        assert!(!can_write_dataset(flags));
    }

    #[test]
    fn test_read_check() {
        assert!(can_read_dataset(0));
        assert!(can_read_dataset(super::FLAG_DELETION_FILES));
        assert!(can_read_dataset(super::FLAG_STABLE_ROW_IDS));
        assert!(can_read_dataset(super::FLAG_USE_V2_FORMAT_DEPRECATED));
        assert!(can_read_dataset(super::FLAG_TABLE_CONFIG));
        assert!(can_read_dataset(super::FLAG_BASE_PATHS));
        assert!(can_read_dataset(super::FLAG_DISABLE_TRANSACTION_FILE));
        assert!(can_read_dataset(super::FLAG_NON_REUSABLE_FIELD_IDS));
        assert!(can_read_dataset(super::FLAG_MIXED_DATA_FILE_VERSIONS));
        // Overlay support is gated on the build profile / env opt-in, so the
        // flag is readable exactly when overlays are enabled (see
        // test_data_overlay_flag_release_gating for the full policy).
        assert_eq!(
            can_read_dataset(super::FLAG_UNSTABLE_DATA_OVERLAY_FILES),
            data_overlay_files_enabled()
        );
        assert!(can_read_dataset(
            super::FLAG_DELETION_FILES
                | super::FLAG_STABLE_ROW_IDS
                | super::FLAG_USE_V2_FORMAT_DEPRECATED
        ));
        assert!(!can_read_dataset(super::FLAG_UNKNOWN));
    }

    #[test]
    fn test_data_overlay_flag_release_gating() {
        // Release default (overlays disabled): the overlay flag is treated as
        // unknown so the dataset is refused, while other known flags still pass.
        let supported = supported_flags_when(false, false);
        assert_eq!(supported & FLAG_UNSTABLE_DATA_OVERLAY_FILES, 0);
        assert_eq!(FLAG_DELETION_FILES & !supported, 0);
        assert_ne!(FLAG_UNSTABLE_DATA_OVERLAY_FILES & !supported, 0);
        // Enabled (debug or env opt-in): the overlay flag is understood.
        let supported = supported_flags_when(true, false);
        assert_eq!(FLAG_UNSTABLE_DATA_OVERLAY_FILES & !supported, 0);
    }

    #[test]
    fn test_spilled_row_lineage_flag_is_stable() {
        assert_eq!(FLAG_SPILLED_ROW_LINEAGE, 2048);
        // Support must not depend on enabling unrelated experimental formats.
        let supported = supported_flags_when(false, false);
        let flags = FLAG_STABLE_ROW_IDS | FLAG_SPILLED_ROW_LINEAGE;
        assert_eq!(flags & !supported, 0);
        assert!(can_read_dataset(flags));
        assert!(can_write_dataset(flags));
        for unsupported in [
            FLAG_UNSTABLE_DATA_OVERLAY_FILES,
            FLAG_FRAG_REUSE_WITH_STABLE_ROW_IDS,
            FLAG_FRAGMENT_TREE,
            FLAG_UNKNOWN,
        ] {
            assert_ne!(unsupported & !supported, 0);
        }
    }

    #[test]
    fn test_apply_feature_flags_sets_overlay_flag() {
        use crate::format::overlay::{DataOverlayFile, OverlayCoverage};
        use crate::format::{DataFile, DataStorageFormat, Fragment};
        use arrow_schema::{Field as ArrowField, Schema as ArrowSchema};
        use lance_core::datatypes::Schema;
        use roaring::RoaringBitmap;
        use std::collections::HashMap;
        use std::sync::Arc;

        let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
            "id",
            arrow_schema::DataType::Int64,
            false,
        )]);
        let schema = Schema::try_from(&arrow_schema).unwrap();
        let mut fragment = Fragment::new(0);
        fragment.overlays = vec![DataOverlayFile {
            data_file: DataFile::new_legacy_from_fields("o.lance", vec![0], None),
            coverage: OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
            committed_version: 1,
        }];
        let mut manifest = Manifest::new(
            schema,
            Arc::new(vec![fragment]),
            DataStorageFormat::default(),
            HashMap::new(),
        );
        apply_feature_flags(&mut manifest, false, false).unwrap();
        assert_ne!(
            manifest.reader_feature_flags & FLAG_UNSTABLE_DATA_OVERLAY_FILES,
            0
        );
        assert_ne!(
            manifest.writer_feature_flags & FLAG_UNSTABLE_DATA_OVERLAY_FILES,
            0
        );
    }

    #[test]
    fn test_apply_feature_flags_sets_spilled_row_lineage_flag() {
        use crate::format::{DataFile, DataStorageFormat, Fragment, ROW_ID_FIELD_ID, RowIdMeta};
        use crate::rowids::version::RowDatasetVersionMeta;
        use arrow_schema::{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(
            "id",
            arrow_schema::DataType::Int64,
            false,
        )]);
        let schema = Schema::try_from(&arrow_schema).unwrap();
        let spilled =
            DataFile::new_legacy_from_fields("lineage.lance", vec![ROW_ID_FIELD_ID], None);
        // Each of the three sequences on its own is enough to require the flag.
        let mut spilled_fragments = Vec::new();
        for kind in 0..3 {
            let mut fragment = Fragment::new(kind);
            fragment.files.push(spilled.clone());
            // Every fragment of a stable-row-id table carries row ids; the
            // version-only cases keep theirs inline.
            fragment.row_id_meta = Some(RowIdMeta::Inline(vec![].into()));
            match kind {
                0 => fragment.row_id_meta = Some(RowIdMeta::Column),
                1 => fragment.created_at_version_meta = Some(RowDatasetVersionMeta::Column),
                _ => fragment.last_updated_at_version_meta = Some(RowDatasetVersionMeta::Column),
            }
            spilled_fragments.push(fragment);
        }
        for fragment in spilled_fragments {
            let mut manifest = Manifest::new(
                schema.clone(),
                Arc::new(vec![fragment]),
                DataStorageFormat::default(),
                HashMap::new(),
            );
            apply_feature_flags(&mut manifest, true, false).unwrap();
            assert_ne!(manifest.reader_feature_flags & FLAG_SPILLED_ROW_LINEAGE, 0);
            assert_ne!(manifest.writer_feature_flags & FLAG_SPILLED_ROW_LINEAGE, 0);
        }

        // An inline sequence does not, so a table whose sequences all fit in
        // the manifest stays readable by builds without the flag.
        let mut fragment = Fragment::new(0);
        fragment.row_id_meta = Some(RowIdMeta::Inline(vec![].into()));
        let mut manifest = Manifest::new(
            schema,
            Arc::new(vec![fragment]),
            DataStorageFormat::default(),
            HashMap::new(),
        );
        apply_feature_flags(&mut manifest, true, false).unwrap();
        assert_eq!(manifest.reader_feature_flags & FLAG_SPILLED_ROW_LINEAGE, 0);
    }

    #[test]
    fn test_write_check() {
        assert!(can_write_dataset(0));
        assert!(can_write_dataset(super::FLAG_DELETION_FILES));
        assert!(can_write_dataset(super::FLAG_STABLE_ROW_IDS));
        assert!(can_write_dataset(super::FLAG_USE_V2_FORMAT_DEPRECATED));
        assert!(can_write_dataset(super::FLAG_TABLE_CONFIG));
        assert!(can_write_dataset(super::FLAG_BASE_PATHS));
        assert!(can_write_dataset(super::FLAG_DISABLE_TRANSACTION_FILE));
        assert!(can_write_dataset(super::FLAG_NON_REUSABLE_FIELD_IDS));
        assert!(can_write_dataset(super::FLAG_MIXED_DATA_FILE_VERSIONS));
        // Overlay support is gated on the build profile / env opt-in, so the
        // flag is writable exactly when overlays are enabled (see
        // test_data_overlay_flag_release_gating for the full policy).
        assert_eq!(
            can_write_dataset(super::FLAG_UNSTABLE_DATA_OVERLAY_FILES),
            data_overlay_files_enabled()
        );
        assert!(can_write_dataset(
            super::FLAG_DELETION_FILES
                | super::FLAG_STABLE_ROW_IDS
                | super::FLAG_USE_V2_FORMAT_DEPRECATED
                | super::FLAG_TABLE_CONFIG
                | super::FLAG_BASE_PATHS
        ));
        assert!(!can_write_dataset(super::FLAG_UNKNOWN));
    }

    #[test]
    fn test_base_paths_feature_flags() {
        use crate::format::{DataStorageFormat, Manifest};
        use arrow_schema::{Field as ArrowField, Schema as ArrowSchema};
        use lance_core::datatypes::Schema;
        use std::collections::HashMap;
        use std::sync::Arc;
        // Create a basic schema for testing
        let arrow_schema = ArrowSchema::new(vec![ArrowField::new(
            "test_field",
            arrow_schema::DataType::Int64,
            false,
        )]);
        let schema = Schema::try_from(&arrow_schema).unwrap();
        // Test 1: Normal dataset (no base_paths) should not have FLAG_BASE_PATHS
        let mut normal_manifest = Manifest::new(
            schema.clone(),
            Arc::new(vec![]),
            DataStorageFormat::default(),
            HashMap::new(), // Empty base_paths
        );
        apply_feature_flags(&mut normal_manifest, false, false).unwrap();
        assert_eq!(normal_manifest.reader_feature_flags & FLAG_BASE_PATHS, 0);
        assert_eq!(normal_manifest.writer_feature_flags & FLAG_BASE_PATHS, 0);
        // Test 2: Dataset with base_paths (shallow clone or multi-base) should have FLAG_BASE_PATHS
        let mut base_paths: HashMap<u32, BasePath> = HashMap::new();
        base_paths.insert(
            1,
            BasePath::new(
                1,
                "file:///path/to/original".to_string(),
                Some("test_ref".to_string()),
                true,
            ),
        );
        let mut multi_base_manifest = Manifest::new(
            schema,
            Arc::new(vec![]),
            DataStorageFormat::default(),
            base_paths,
        );
        apply_feature_flags(&mut multi_base_manifest, false, false).unwrap();
        assert_ne!(
            multi_base_manifest.reader_feature_flags & FLAG_BASE_PATHS,
            0
        );
        assert_ne!(
            multi_base_manifest.writer_feature_flags & FLAG_BASE_PATHS,
            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_preserves_non_reusable_field_id_writer_gate() {
        let mut source = empty_manifest();
        source.activate_non_reusable_field_ids();
        source.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
        let mut destination = empty_manifest();

        inherit_sticky_feature_flags(&mut destination, &source).unwrap();

        assert_eq!(
            destination.reader_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS,
            0
        );
        assert_ne!(
            destination.writer_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS,
            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 paired_validation_rejects_half_set_mixed_version_capability() {
        for (reader, writer) in [
            (FLAG_MIXED_DATA_FILE_VERSIONS, 0),
            (0, FLAG_MIXED_DATA_FILE_VERSIONS),
        ] {
            let mut manifest = empty_manifest();
            manifest.reader_feature_flags = reader;
            manifest.writer_feature_flags = writer;

            let err = validate_paired_feature_flags(&manifest).unwrap_err();

            assert!(err.to_string().contains("mixed data-file-version"), "{err}");
        }
    }

    #[test]
    fn writer_gate_accepts_mixed_capability_and_rejects_unknown_flags() {
        let mut manifest = empty_manifest();
        manifest.reader_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS;
        manifest.writer_feature_flags = FLAG_MIXED_DATA_FILE_VERSIONS;
        ensure_can_write_manifest(&manifest).unwrap();

        manifest.writer_feature_flags |= FLAG_UNKNOWN;
        let err = ensure_can_write_manifest(&manifest).unwrap_err();
        assert!(matches!(err, Error::NotSupported { .. }));
        assert!(err.to_string().contains("cannot be written"), "{err}");
    }

    #[rstest::rstest]
    fn apply_feature_flags_sets_writer_gate_for_explicit_non_reusable_field_id_activation(
        #[values(false, true)] managed_blobs: bool,
    ) {
        let mut manifest = empty_manifest();
        let managed_blob_flags = if managed_blobs { FLAG_MANAGED_BLOBS } else { 0 };
        manifest.reader_feature_flags = managed_blob_flags;
        manifest.writer_feature_flags = managed_blob_flags;
        manifest.activate_non_reusable_field_ids();

        apply_feature_flags(&mut manifest, false, false).unwrap();

        assert_eq!(
            manifest.reader_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS,
            0
        );
        assert_ne!(
            manifest.writer_feature_flags & FLAG_NON_REUSABLE_FIELD_IDS,
            0
        );
        assert_eq!(
            manifest.reader_feature_flags & FLAG_MANAGED_BLOBS,
            managed_blob_flags
        );
        assert_eq!(
            manifest.writer_feature_flags & FLAG_MANAGED_BLOBS,
            managed_blob_flags
        );
    }

    #[test]
    fn apply_feature_flags_rejects_non_reusable_field_id_reader_flag() {
        let mut manifest = empty_manifest();
        manifest.activate_non_reusable_field_ids();
        manifest.reader_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
        manifest.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;

        let err = apply_feature_flags(&mut manifest, false, false).unwrap_err();

        assert!(
            err.to_string().contains("only require writer support"),
            "{err}"
        );
    }

    #[test]
    fn non_reusable_field_id_marker_and_writer_gate_must_agree() {
        let mut activated_without_gate = empty_manifest();
        activated_without_gate.activate_non_reusable_field_ids();
        assert!(validate_non_reusable_field_id_flags(&activated_without_gate).is_err());

        let mut gate_without_marker = empty_manifest();
        gate_without_marker.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
        assert!(validate_non_reusable_field_id_flags(&gate_without_marker).is_err());

        let mut writer_only = empty_manifest();
        writer_only.activate_non_reusable_field_ids();
        writer_only.writer_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
        validate_non_reusable_field_id_flags(&writer_only).unwrap();

        let mut paired = writer_only.clone();
        paired.reader_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
        assert!(validate_non_reusable_field_id_flags(&paired).is_err());

        let mut reader_without_activation = empty_manifest();
        reader_without_activation.reader_feature_flags |= FLAG_NON_REUSABLE_FIELD_IDS;
        assert!(validate_non_reusable_field_id_flags(&reader_without_activation).is_err());
    }

    #[rstest::rstest]
    #[case::reader_only(true, false)]
    #[case::writer_only(false, true)]
    #[case::paired(true, true)]
    fn managed_capability_is_paired_and_sticky(#[case] reader: bool, #[case] writer: bool) {
        let mut source = empty_manifest();
        source.reader_feature_flags = if reader { FLAG_MANAGED_BLOBS } else { 0 };
        source.writer_feature_flags = if writer { FLAG_MANAGED_BLOBS } else { 0 };
        if reader != writer {
            for error in [
                ensure_can_read_manifest(&source).unwrap_err(),
                ensure_can_write_manifest(&source).unwrap_err(),
            ] {
                assert!(matches!(error, Error::CorruptFile { .. }));
                assert!(error.to_string().contains("Managed Blob"));
            }
            return;
        }
        ensure_can_read_manifest(&source).unwrap();
        ensure_can_write_manifest(&source).unwrap();
        let mut destination = empty_manifest();
        inherit_sticky_feature_flags(&mut destination, &source).unwrap();
        apply_feature_flags(&mut destination, false, false).unwrap();
        assert!(destination.has_managed_blobs());
        assert_eq!(
            destination.writer_feature_flags & FLAG_MANAGED_BLOBS,
            FLAG_MANAGED_BLOBS
        );
        // The released v11.0.0 client only accepts bits below 128.
        assert_ne!(destination.reader_feature_flags & !(128 - 1), 0);
    }

    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(),
        )
    }

    #[test]
    fn paired_capabilities_are_below_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!(can_read_dataset(FLAG_MANAGED_BLOBS));
        assert!(can_write_dataset(FLAG_MANAGED_BLOBS));
        assert!(!can_read_dataset(FLAG_UNKNOWN));
        assert!(!can_write_dataset(FLAG_UNKNOWN));
    }
}