buoyant_kernel 0.21.103

#[internal_api]
pub(crate) use column_mapping::get_any_level_column_physical_name;
#[deprecated = "Enable internal-api and use TableConfiguration instead"]
pub use column_mapping::validate_schema_column_mapping;
pub use column_mapping::ColumnMappingMode;
pub(crate) use column_mapping::{
    assign_column_mapping_metadata, column_mapping_mode, get_column_mapping_mode_from_properties,
    get_field_column_mapping_info, physical_to_logical_column_name,
};
use delta_kernel_derive::internal_api;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use strum::{AsRefStr, Display as StrumDisplay, EnumCount, EnumIter, EnumString};
pub(crate) use timestamp_ntz::{
    schema_contains_timestamp_ntz, validate_timestamp_ntz_feature_support,
};

use crate::actions::Protocol;
use crate::expressions::Scalar;
use crate::schema::derive_macro_utils::ToDataType;
use crate::schema::DataType;
use crate::table_properties::TableProperties;
use crate::{DeltaResult, Error};
mod column_mapping;
mod timestamp_ntz;

/// Minimum reader/writer protocol version that the kernel can handle.
pub const MIN_VALID_RW_VERSION: i32 = 1;

/// Maximum reader protocol version that the kernel can handle.
pub const MAX_VALID_READER_VERSION: i32 = 3;

/// Maximum writer protocol version that the kernel can handle.
pub const MAX_VALID_WRITER_VERSION: i32 = 7;

/// Minimum reader version for tables that use table features.
/// When set to 3, the protocol requires an explicit `readerFeatures` array.
pub const TABLE_FEATURES_MIN_READER_VERSION: i32 = 3;

/// Minimum writer version for tables that use table features.
/// When set to 7, the protocol requires an explicit `writerFeatures` array.
pub const TABLE_FEATURES_MIN_WRITER_VERSION: i32 = 7;

/// Prefix for table feature override properties.
/// Properties with this prefix (e.g., `delta.feature.deletionVectors`) are used to
/// explicitly turn on support for the feature in the protocol.
pub const SET_TABLE_FEATURE_SUPPORTED_PREFIX: &str = "delta.feature.";

/// Value to add support for a table feature when used with [`SET_TABLE_FEATURE_SUPPORTED_PREFIX`].
/// Example: `"delta.feature.deletionVectors" -> "supported"`
pub const SET_TABLE_FEATURE_SUPPORTED_VALUE: &str = "supported";

/// Table features represent protocol capabilities required to correctly read or write a given
/// table.
/// - Readers must implement all features required for correct table reads.
/// - Writers must implement all features required for correct table writes.
///
/// Each variant corresponds to one such feature. A feature is either:
/// - **ReaderWriter** (must be supported by both readers and writers), or
/// - **WriterOnly** (applies only to writers).
/// There are no ReaderOnly features. See `TableFeature::feature_type` for the category of each.
///
/// The kernel currently supports all reader features except `V2Checkpoint`.
#[derive(
    Serialize,
    Deserialize,
    Debug,
    Clone,
    Eq,
    PartialEq,
    EnumString,
    StrumDisplay,
    AsRefStr,
    EnumCount,
    Hash,
)]
#[strum(
    serialize_all = "camelCase",
    parse_err_fn = xxx__not_needed__default_variant_means_parsing_is_infallible__xxx,
    parse_err_ty = Infallible // ignored, sadly: https://github.com/Peternator7/strum/issues/430
)]
#[serde(rename_all = "camelCase")]
#[internal_api]
#[derive(EnumIter)]
// ^^ We must derive EnumIter only after internal_api adjusts visibility. Otherwise, internal-api
// builds will fail because the now-public `TableFeature::iter()` returns a pub(crate) type.
pub(crate) enum TableFeature {
    //////////////////////////
    // Writer-only features //
    //////////////////////////
    /// Append Only Tables
    AppendOnly,
    /// Table invariants
    Invariants,
    /// Check constraints on columns
    CheckConstraints,
    /// CDF on a table
    ChangeDataFeed,
    /// Columns with generated values
    GeneratedColumns,
    /// ID Columns
    IdentityColumns,
    /// Monotonically increasing timestamps in the CommitInfo
    InCommitTimestamp,
    /// Row tracking on tables
    RowTracking,
    /// domain specific metadata
    DomainMetadata,
    /// Iceberg compatibility support
    IcebergCompatV1,
    /// Iceberg compatibility support
    IcebergCompatV2,
    /// The Clustered Table feature facilitates the physical clustering of rows
    /// that share similar values on a predefined set of clustering columns.
    #[strum(serialize = "clustering")]
    #[serde(rename = "clustering")]
    ClusteredTable,
    /// Materialize partition columns in parquet data files.
    MaterializePartitionColumns,

    ///////////////////////////
    // ReaderWriter features //
    ///////////////////////////
    /// CatalogManaged tables:
    /// <https://github.com/delta-io/delta/blob/master/protocol_rfcs/catalog-managed.md>
    CatalogManaged,
    #[strum(serialize = "catalogOwned-preview")]
    #[serde(rename = "catalogOwned-preview")]
    CatalogOwnedPreview,
    /// Mapping of one column to another
    ColumnMapping,
    /// Deletion vectors for merge, update, delete
    DeletionVectors,
    /// timestamps without timezone support
    #[strum(serialize = "timestampNtz")]
    #[serde(rename = "timestampNtz")]
    TimestampWithoutTimezone,
    // Allow columns to change type
    TypeWidening,
    #[strum(serialize = "typeWidening-preview")]
    #[serde(rename = "typeWidening-preview")]
    TypeWideningPreview,
    /// version 2 of checkpointing
    V2Checkpoint,
    /// vacuumProtocolCheck ReaderWriter feature ensures consistent application of reader and
    /// writer protocol checks during VACUUM operations
    VacuumProtocolCheck,
    /// This feature enables support for the variant data type, which stores semi-structured data.
    VariantType,
    #[strum(serialize = "variantType-preview")]
    #[serde(rename = "variantType-preview")]
    VariantTypePreview,
    #[strum(serialize = "variantShredding-preview")]
    #[serde(rename = "variantShredding-preview")]
    VariantShreddingPreview,

    #[serde(untagged)]
    #[strum(default)]
    Unknown(String),
}

/// ReaderWriter features that can be supported by legacy readers (min_reader_version < 3).
/// Only ColumnMapping qualifies with min_reader_version = 2.
pub(crate) static LEGACY_READER_FEATURES: [TableFeature; 1] = [TableFeature::ColumnMapping];

/// Writer and ReaderWriter features that can be supported by legacy writers (min_writer_version <
/// 7). These are features with min_writer_version in range [1, 6].
pub(crate) static LEGACY_WRITER_FEATURES: [TableFeature; 7] = [
    // Writer-only features (min_writer < 7)
    TableFeature::AppendOnly,       // min_writer = 2
    TableFeature::Invariants,       // min_writer = 2
    TableFeature::CheckConstraints, // min_writer = 3
    TableFeature::ChangeDataFeed,   // min_writer = 4
    TableFeature::GeneratedColumns, // min_writer = 4
    TableFeature::IdentityColumns,  // min_writer = 6
    // ReaderWriter features (min_writer < 7)
    TableFeature::ColumnMapping, // min_writer = 5
];

/// Classifies table features by their type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum FeatureType {
    /// Feature only affects write operations
    WriterOnly,
    /// Feature affects both read and write operations (must appear in both feature lists)
    ReaderWriter,
    /// Unknown feature type (for forward compatibility)
    Unknown,
}

/// Defines how a feature's enablement is determined
#[derive(Debug, Clone, Copy)]
pub(crate) enum EnablementCheck {
    /// Feature is enabled if it's supported (appears in protocol feature lists)
    AlwaysIfSupported,
    /// Feature is enabled if supported AND the provided function returns true when checking table
    /// properties
    EnabledIf(fn(&TableProperties) -> bool),
}

/// Represents the type of operation being performed on a table
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[internal_api]
pub(crate) enum Operation {
    /// Read operations on regular table data
    Scan,
    /// Read operations on change data feed data
    Cdf,
    /// Write operations on regular table data
    Write,
}

/// Defines whether the Rust kernel has implementation support for a feature's operation
pub(crate) enum KernelSupport {
    /// Kernel has full support for any operation on this feature
    Supported,
    /// Kernel does not support this operation on this feature
    NotSupported,
    /// Custom logic to determine support based on operation type and table properties.
    /// For example: Column Mapping may support Scan but not CDF, or CDF writes may only
    /// be supported when AppendOnly is true.
    Custom(fn(&Protocol, &TableProperties, Operation) -> DeltaResult<()>),
}

/// Types of requirements for feature dependencies
#[derive(Debug)]
pub(crate) enum FeatureRequirement {
    /// Feature must be supported (in protocol)
    Supported(TableFeature),
    /// Feature must be enabled (supported + property set)
    Enabled(TableFeature),
    /// Feature must NOT be supported
    NotSupported(TableFeature),
    /// Feature must NOT be enabled (may be supported but property must not activate it)
    NotEnabled(TableFeature),
    /// Custom validation logic
    Custom(fn(&Protocol, &TableProperties) -> DeltaResult<()>),
}

/// Minimum protocol versions for legacy (pre-feature-list) inference.
pub(crate) struct MinReaderWriterVersion {
    pub reader: i32,
    pub writer: i32,
}

impl MinReaderWriterVersion {
    pub(crate) const fn new(reader: i32, writer: i32) -> Self {
        Self { reader, writer }
    }
}

/// Rich metadata about a table feature including version requirements, dependencies, and support
/// status
pub(crate) struct FeatureInfo {
    /// The type of feature (WriterOnly, ReaderWriter, or Unknown)
    pub feature_type: FeatureType,
    /// Minimum legacy protocol versions for version-based feature inference.
    /// `Some` for features that predate feature lists and can be inferred from protocol version.
    /// `None` for features that require explicit feature lists (reader v3+ / writer v7+).
    pub min_legacy_version: Option<MinReaderWriterVersion>,
    /// Requirements this feature has (features + custom validations)
    pub feature_requirements: &'static [FeatureRequirement],
    /// Rust kernel's support for this feature (may vary by Operation type)
    ///
    /// Note: `kernel_support` validation depends on `feature_type`:
    /// WriterOnly features: Only checked during `Operation::Write`
    /// ReaderWriter features: Checked during all operations (Scan/Write/CDF)
    /// Read operations (Scan/CDF) only validate reader features, so `kernel_support` for
    /// WriterOnly features are never invoked for Scan/CDF regardless of the custom check logic.
    pub kernel_support: KernelSupport,
    /// How to check if this feature is enabled in a table
    pub enablement_check: EnablementCheck,
}

// Static FeatureInfo instances for each table feature
static APPEND_ONLY_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: Some(MinReaderWriterVersion::new(1, 2)),
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::EnabledIf(|props| props.append_only == Some(true)),
};

// Although kernel marks invariants as "Supported", invariants must NOT actually be present in the
// table schema. Kernel will fail to write to any table that actually uses invariants (see check in
// TableConfiguration::ensure_write_supported). This is to allow legacy tables with the Invariants
// feature enabled but not in use.
static INVARIANTS_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: Some(MinReaderWriterVersion::new(1, 2)),
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static CHECK_CONSTRAINTS_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: Some(MinReaderWriterVersion::new(1, 3)),
    feature_requirements: &[],
    kernel_support: KernelSupport::NotSupported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static CHANGE_DATA_FEED_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: Some(MinReaderWriterVersion::new(1, 4)),
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.enable_change_data_feed == Some(true)
    }),
};

static GENERATED_COLUMNS_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: Some(MinReaderWriterVersion::new(1, 4)),
    feature_requirements: &[],
    kernel_support: KernelSupport::NotSupported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static IDENTITY_COLUMNS_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: Some(MinReaderWriterVersion::new(1, 6)),
    feature_requirements: &[],
    kernel_support: KernelSupport::NotSupported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static IN_COMMIT_TIMESTAMP_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Custom(|_protocol, _properties, operation| match operation {
        Operation::Scan | Operation::Write | Operation::Cdf => Ok(()),
    }),
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.enable_in_commit_timestamps == Some(true)
    }),
};

static ROW_TRACKING_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[FeatureRequirement::Supported(TableFeature::DomainMetadata)],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.enable_row_tracking == Some(true) && props.row_tracking_suspended != Some(true)
    }),
};

static DOMAIN_METADATA_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

// TODO(#1125): IcebergCompatV1 requires schema type validation to block Map, Array, and Void types.
// This validation is not yet implemented. The feature is marked as NotSupported for writes until
// proper validation is added.
//
// See Delta Spark: IcebergCompat.scala CheckNoListMapNullType (lines 422-433)
// See Java Kernel: IcebergWriterCompatMetadataValidatorAndUpdater.java
// UNSUPPORTED_TYPES_CHECK See https://github.com/delta-io/delta/blob/master/PROTOCOL.md#writer-requirements-for-icebergcompatv1 for more requirements to support
static ICEBERG_COMPAT_V1_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[
        FeatureRequirement::Enabled(TableFeature::ColumnMapping),
        FeatureRequirement::Custom(|_protocol, properties| {
            let mode = properties.column_mapping_mode;
            if !matches!(
                mode,
                Some(ColumnMappingMode::Name) | Some(ColumnMappingMode::Id)
            ) {
                return Err(Error::generic(
                    "IcebergCompatV1 requires Column Mapping in 'name' or 'id' mode",
                ));
            }
            Ok(())
        }),
        FeatureRequirement::NotSupported(TableFeature::DeletionVectors),
    ],
    kernel_support: KernelSupport::NotSupported,
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.enable_iceberg_compat_v1 == Some(true)
    }),
};

// TODO(#1125): IcebergCompatV2 requires schema type validation. Unlike V1, V2 allows Map and Array
// types but needs validation against an allowlist of supported types.
// This validation is not yet implemented. The feature is marked as NotSupported for writes until
// proper validation is added.

// See Delta Spark: IcebergCompat.scala CheckTypeInV2AllowList (lines 450-459)
// See Java Kernel: IcebergCompatMetadataValidatorAndUpdater.java V2_SUPPORTED_TYPES
// See https://github.com/delta-io/delta/blob/master/PROTOCOL.md#writer-requirements-for-icebergcompatv2 for more requirements to support.
static ICEBERG_COMPAT_V2_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[
        FeatureRequirement::Enabled(TableFeature::ColumnMapping),
        FeatureRequirement::Custom(|_protocol, properties| {
            let mode = properties.column_mapping_mode;
            if !matches!(
                mode,
                Some(ColumnMappingMode::Name) | Some(ColumnMappingMode::Id)
            ) {
                return Err(Error::generic(
                    "IcebergCompatV2 requires Column Mapping in 'name' or 'id' mode",
                ));
            }
            Ok(())
        }),
        FeatureRequirement::NotEnabled(TableFeature::IcebergCompatV1),
        FeatureRequirement::NotEnabled(TableFeature::DeletionVectors),
    ],
    kernel_support: KernelSupport::NotSupported,
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.enable_iceberg_compat_v2 == Some(true)
    }),
};

static CLUSTERED_TABLE_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[FeatureRequirement::Supported(TableFeature::DomainMetadata)],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static MATERIALIZE_PARTITION_COLUMNS_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::WriterOnly,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static CATALOG_MANAGED_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Custom(|_, _, op| match op {
        Operation::Scan | Operation::Write => Ok(()),
        Operation::Cdf => Err(Error::unsupported(
            "Feature 'catalogManaged' is not supported for CDF",
        )),
    }),
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static CATALOG_OWNED_PREVIEW_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Custom(|_, _, op| match op {
        Operation::Scan | Operation::Write => Ok(()),
        Operation::Cdf => Err(Error::unsupported(
            "Feature 'catalogOwned-preview' is not supported for CDF",
        )),
    }),
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static COLUMN_MAPPING_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: Some(MinReaderWriterVersion::new(2, 5)),
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.column_mapping_mode.is_some()
            && props.column_mapping_mode != Some(ColumnMappingMode::None)
    }),
};

static DELETION_VECTORS_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    // We support writing to tables with DeletionVectors enabled, but we never write DV files
    // ourselves (no DML). The kernel only performs append operations.
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::EnabledIf(|props| {
        props.enable_deletion_vectors == Some(true)
    }),
};

static TIMESTAMP_WITHOUT_TIMEZONE_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static TYPE_WIDENING_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Custom(|_, _, op| match op {
        Operation::Scan | Operation::Cdf => Ok(()),
        Operation::Write => Err(Error::unsupported(
            "Feature 'typeWidening' is not supported for writes",
        )),
    }),
    enablement_check: EnablementCheck::EnabledIf(|props| props.enable_type_widening == Some(true)),
};

static TYPE_WIDENING_PREVIEW_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Custom(|_, _, op| match op {
        Operation::Scan | Operation::Cdf => Ok(()),
        Operation::Write => Err(Error::unsupported(
            "Feature 'typeWidening-preview' is not supported for writes",
        )),
    }),
    enablement_check: EnablementCheck::EnabledIf(|props| props.enable_type_widening == Some(true)),
};

static V2_CHECKPOINT_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static VACUUM_PROTOCOL_CHECK_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static VARIANT_TYPE_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static VARIANT_TYPE_PREVIEW_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

static VARIANT_SHREDDING_PREVIEW_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::ReaderWriter,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::Supported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

/// By definition, kernel cannot know how to handle unknown features and must assume they're always
/// enabled if supported in protocol. However, the read path ignores all writer-only features,
/// including unknown ones. Unknown features are never inferred from legacy protocol versions.
static UNKNOWN_FEATURE_INFO: FeatureInfo = FeatureInfo {
    feature_type: FeatureType::Unknown,
    min_legacy_version: None,
    feature_requirements: &[],
    kernel_support: KernelSupport::NotSupported,
    enablement_check: EnablementCheck::AlwaysIfSupported,
};

impl TableFeature {
    #[cfg(test)]
    pub(crate) const NO_LIST: Option<Vec<TableFeature>> = None;
    #[cfg(test)]
    pub(crate) const EMPTY_LIST: Vec<TableFeature> = vec![];

    pub(crate) fn feature_type(&self) -> FeatureType {
        match self {
            TableFeature::CatalogManaged
            | TableFeature::CatalogOwnedPreview
            | TableFeature::ColumnMapping
            | TableFeature::DeletionVectors
            | TableFeature::TimestampWithoutTimezone
            | TableFeature::TypeWidening
            | TableFeature::TypeWideningPreview
            | TableFeature::V2Checkpoint
            | TableFeature::VacuumProtocolCheck
            | TableFeature::VariantType
            | TableFeature::VariantTypePreview
            | TableFeature::VariantShreddingPreview => FeatureType::ReaderWriter,
            TableFeature::AppendOnly
            | TableFeature::DomainMetadata
            | TableFeature::Invariants
            | TableFeature::RowTracking
            | TableFeature::CheckConstraints
            | TableFeature::ChangeDataFeed
            | TableFeature::GeneratedColumns
            | TableFeature::IdentityColumns
            | TableFeature::InCommitTimestamp
            | TableFeature::IcebergCompatV1
            | TableFeature::IcebergCompatV2
            | TableFeature::ClusteredTable
            | TableFeature::MaterializePartitionColumns => FeatureType::WriterOnly,
            TableFeature::Unknown(_) => FeatureType::Unknown,
        }
    }

    /// Returns true if this feature can be inferred from a legacy reader protocol version.
    /// Always returns false for modern features (use feature lists instead).
    pub(crate) fn is_valid_for_legacy_reader(&self, reader_version: i32) -> bool {
        matches!(&self.info().min_legacy_version, Some(v) if reader_version >= v.reader)
    }

    /// Returns true if this feature can be inferred from a legacy writer protocol version.
    /// Always returns false for modern features (use feature lists instead).
    pub(crate) fn is_valid_for_legacy_writer(&self, writer_version: i32) -> bool {
        matches!(&self.info().min_legacy_version, Some(v) if writer_version >= v.writer)
    }

    /// Returns rich metadata about this table feature including protocol version requirements,
    /// dependencies, and kernel support status.
    pub(crate) fn info(&self) -> &FeatureInfo {
        match self {
            // Writer-only features
            TableFeature::AppendOnly => &APPEND_ONLY_INFO,
            TableFeature::Invariants => &INVARIANTS_INFO,
            TableFeature::CheckConstraints => &CHECK_CONSTRAINTS_INFO,
            TableFeature::ChangeDataFeed => &CHANGE_DATA_FEED_INFO,
            TableFeature::GeneratedColumns => &GENERATED_COLUMNS_INFO,
            TableFeature::IdentityColumns => &IDENTITY_COLUMNS_INFO,
            TableFeature::InCommitTimestamp => &IN_COMMIT_TIMESTAMP_INFO,
            TableFeature::RowTracking => &ROW_TRACKING_INFO,
            TableFeature::DomainMetadata => &DOMAIN_METADATA_INFO,
            TableFeature::IcebergCompatV1 => &ICEBERG_COMPAT_V1_INFO,
            TableFeature::IcebergCompatV2 => &ICEBERG_COMPAT_V2_INFO,
            TableFeature::ClusteredTable => &CLUSTERED_TABLE_INFO,
            TableFeature::MaterializePartitionColumns => &MATERIALIZE_PARTITION_COLUMNS_INFO,

            // ReaderWriter features
            TableFeature::CatalogManaged => &CATALOG_MANAGED_INFO,
            TableFeature::CatalogOwnedPreview => &CATALOG_OWNED_PREVIEW_INFO,
            TableFeature::ColumnMapping => &COLUMN_MAPPING_INFO,
            TableFeature::DeletionVectors => &DELETION_VECTORS_INFO,
            TableFeature::TimestampWithoutTimezone => &TIMESTAMP_WITHOUT_TIMEZONE_INFO,
            TableFeature::TypeWidening => &TYPE_WIDENING_INFO,
            TableFeature::TypeWideningPreview => &TYPE_WIDENING_PREVIEW_INFO,
            TableFeature::V2Checkpoint => &V2_CHECKPOINT_INFO,
            TableFeature::VacuumProtocolCheck => &VACUUM_PROTOCOL_CHECK_INFO,
            TableFeature::VariantType => &VARIANT_TYPE_INFO,
            TableFeature::VariantTypePreview => &VARIANT_TYPE_PREVIEW_INFO,
            TableFeature::VariantShreddingPreview => &VARIANT_SHREDDING_PREVIEW_INFO,

            // Unknown features: not supported by kernel, no legacy version inference.
            TableFeature::Unknown(_) => &UNKNOWN_FEATURE_INFO,
        }
    }
}

impl ToDataType for TableFeature {
    fn to_data_type() -> DataType {
        DataType::STRING
    }
}

impl From<TableFeature> for Scalar {
    fn from(feature: TableFeature) -> Self {
        Scalar::String(feature.to_string())
    }
}

#[cfg(test)] // currently only used in tests
impl TableFeature {
    pub(crate) fn unknown(s: impl ToString) -> Self {
        TableFeature::Unknown(s.to_string())
    }
}

/// Like `Into<TableFeature>`, but avoids collisions between strum's derived `EnumString` and the
/// blanket impl `TryFrom<&str>` that `From<&str> for TableFeature` would trigger.
///
/// Parsing is infallible: the `Unknown` default variant catches any unrecognized feature name. If
/// https://github.com/Peternator7/strum/pull/432 merges, use impl From for TableFeature instead.
pub(crate) trait IntoTableFeature {
    fn into_table_feature(self) -> TableFeature;
}

impl IntoTableFeature for TableFeature {
    fn into_table_feature(self) -> TableFeature {
        self
    }
}

impl IntoTableFeature for &TableFeature {
    fn into_table_feature(self) -> TableFeature {
        self.clone()
    }
}

/// Parsing is infallible thanks to `TableFeature::Unknown` default variant
impl IntoTableFeature for &str {
    fn into_table_feature(self) -> TableFeature {
        #[allow(clippy::unwrap_used)] // infallible, see strum parse_err_fn
        self.parse().unwrap()
    }
}

impl IntoTableFeature for String {
    fn into_table_feature(self) -> TableFeature {
        self.as_str().into_table_feature()
    }
}

/// Formats a slice of table features using Delta's standard serialization (camelCase).
pub(crate) fn format_features(features: &[TableFeature]) -> String {
    let feature_strings: Vec<&str> = features.iter().map(|f| f.as_ref()).collect_vec();
    format!("[{}]", feature_strings.join(", "))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_unknown_features() {
        let mixed_reader = &[
            TableFeature::DeletionVectors,
            TableFeature::unknown("cool_feature"),
            TableFeature::ColumnMapping,
        ];
        let mixed_writer = &[
            TableFeature::DeletionVectors,
            TableFeature::unknown("cool_feature"),
            TableFeature::AppendOnly,
        ];

        let reader_string = serde_json::to_string(mixed_reader).unwrap();
        let writer_string = serde_json::to_string(mixed_writer).unwrap();

        assert_eq!(
            &reader_string,
            "[\"deletionVectors\",\"cool_feature\",\"columnMapping\"]"
        );
        assert_eq!(
            &writer_string,
            "[\"deletionVectors\",\"cool_feature\",\"appendOnly\"]"
        );

        let typed_reader: Vec<TableFeature> = serde_json::from_str(&reader_string).unwrap();
        let typed_writer: Vec<TableFeature> = serde_json::from_str(&writer_string).unwrap();

        assert_eq!(typed_reader.len(), 3);
        assert_eq!(&typed_reader, mixed_reader);
        assert_eq!(typed_writer.len(), 3);
        assert_eq!(&typed_writer, mixed_writer);
    }

    #[test]
    fn test_roundtrip_table_features() {
        use strum::IntoEnumIterator as _;

        for feature in TableFeature::iter() {
            let expected = match feature {
                TableFeature::AppendOnly => "appendOnly",
                TableFeature::Invariants => "invariants",
                TableFeature::CheckConstraints => "checkConstraints",
                TableFeature::ChangeDataFeed => "changeDataFeed",
                TableFeature::GeneratedColumns => "generatedColumns",
                TableFeature::IdentityColumns => "identityColumns",
                TableFeature::InCommitTimestamp => "inCommitTimestamp",
                TableFeature::RowTracking => "rowTracking",
                TableFeature::DomainMetadata => "domainMetadata",
                TableFeature::IcebergCompatV1 => "icebergCompatV1",
                TableFeature::IcebergCompatV2 => "icebergCompatV2",
                TableFeature::ClusteredTable => "clustering",
                TableFeature::MaterializePartitionColumns => "materializePartitionColumns",
                TableFeature::CatalogManaged => "catalogManaged",
                TableFeature::CatalogOwnedPreview => "catalogOwned-preview",
                TableFeature::ColumnMapping => "columnMapping",
                TableFeature::DeletionVectors => "deletionVectors",
                TableFeature::TimestampWithoutTimezone => "timestampNtz",
                TableFeature::TypeWidening => "typeWidening",
                TableFeature::TypeWideningPreview => "typeWidening-preview",
                TableFeature::V2Checkpoint => "v2Checkpoint",
                TableFeature::VacuumProtocolCheck => "vacuumProtocolCheck",
                TableFeature::VariantType => "variantType",
                TableFeature::VariantTypePreview => "variantType-preview",
                TableFeature::VariantShreddingPreview => "variantShredding-preview",
                TableFeature::Unknown(_) => continue, // tested in test_unknown_features
            };

            // strum
            assert_eq!(feature.to_string(), expected);
            assert_eq!(feature, expected.into_table_feature());

            // json
            let serialized = serde_json::to_string(&feature).unwrap();
            assert_eq!(serialized, format!("\"{expected}\""));

            let deserialized: TableFeature = serde_json::from_str(&serialized).unwrap();
            assert_eq!(deserialized, feature);
        }
    }
}