delta_kernel 0.30.0

//! Provides parsing and manipulation of the various actions defined in the [Delta
//! specification](https://github.com/delta-io/delta/blob/master/PROTOCOL.md)

use std::collections::HashMap;
use std::fmt;
use std::sync::LazyLock;

use delta_kernel_derive::{internal_api, IntoStructData, ToSchema, TryFromStructData};
use derive_more::Constructor;
use serde::de::{MapAccess, Visitor};
use serde::{Deserialize, Deserializer, Serialize};
use tracing::warn;
use url::Url;
use visitors::{MetadataVisitor, ProtocolVisitor};

use self::deletion_vector::DeletionVectorDescriptor;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::amt_path_util::resolve_amt_location;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::expressions::Scalar;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::expressions::{ArrayData, StructData};
use crate::schema::{
    is_unsupported_delta_type_error, lazy_schema_ref, schema_ref, SchemaRef, StructField,
    StructType, ToSchema as _,
};
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::schema::{schema, ArrayType, DataType};
use crate::table_features::{
    FeatureType, TableFeature, LEGACY_READER_FEATURES, MIN_VALID_RW_VERSION,
    TABLE_FEATURES_MIN_READER_VERSION, TABLE_FEATURES_MIN_WRITER_VERSION,
};
use crate::table_properties::TableProperties;
use crate::utils::require;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::{create_row, Engine};
use crate::{EngineData, FileMeta, FileSize, KernelError, KernelResult, RowVisitor as _};

const KERNEL_VERSION: &str = env!("CARGO_PKG_VERSION");
const SERDE_JSON_RECURSION_LIMIT_ERROR_PREFIX: &str = "recursion limit exceeded";
const UNKNOWN_OPERATION: &str = "UNKNOWN";
pub(crate) const ROW_TRACKING_PRESERVED_TAG: &str = "delta.rowTracking.preserved";

pub mod deletion_vector;
pub mod deletion_vector_writer;
pub mod set_transaction;

// see comment in ../lib.rs for the path module for why we include this way
#[cfg(feature = "internal-api")]
pub mod visitors;
#[cfg(not(feature = "internal-api"))]
pub(crate) mod visitors;

#[internal_api]
pub(crate) const ADD_NAME: &str = "add";
#[internal_api]
pub(crate) const REMOVE_NAME: &str = "remove";
#[internal_api]
pub(crate) const METADATA_NAME: &str = "metaData";
#[internal_api]
pub(crate) const PROTOCOL_NAME: &str = "protocol";
#[internal_api]
pub(crate) const SET_TRANSACTION_NAME: &str = "txn";
#[internal_api]
pub(crate) const COMMIT_INFO_NAME: &str = "commitInfo";
#[internal_api]
pub(crate) const CDC_NAME: &str = "cdc";
#[internal_api]
pub(crate) const SIDECAR_NAME: &str = "sidecar";
#[internal_api]
pub(crate) const CHECKPOINT_METADATA_NAME: &str = "checkpointMetadata";
/// Optional `checkpointMetadata.tags` key whose value is the JSON-encoded `StructType` of the
/// checkpoint's sidecar files
pub(crate) const SIDECAR_FILE_SCHEMA_TAG: &str = "sidecarFileSchema";
#[internal_api]
pub(crate) const DOMAIN_METADATA_NAME: &str = "domainMetadata";
#[cfg(feature = "adaptive-metadata-in-dev")]
#[internal_api]
pub(crate) const CHECKPOINT_ACTION_NAME: &str = "checkpoint";
#[cfg(feature = "adaptive-metadata-in-dev")]
#[internal_api]
pub(crate) const CONTENT_ROOT_NAME: &str = "contentRoot";

pub(crate) const INTERNAL_DOMAIN_PREFIX: &str = "delta.";

/// Returns the required leaf used to identify rows containing `action_name`.
///
/// Returns `None` when `action_name` is unknown or has no required identifying leaf.
pub(crate) fn action_presence_leaf(action_name: &str) -> Option<&'static str> {
    match action_name {
        ADD_NAME | REMOVE_NAME | CDC_NAME | SIDECAR_NAME => Some("path"),
        METADATA_NAME => Some("id"),
        PROTOCOL_NAME => Some("minReaderVersion"),
        SET_TRANSACTION_NAME => Some("appId"),
        DOMAIN_METADATA_NAME => Some("domain"),
        CHECKPOINT_METADATA_NAME => Some("version"),
        _ => None,
    }
}

// === Sub-fields of an AddFile's `stats` struct ===
// See the Delta protocol spec, "Per-file Statistics", and `expected_stats_schema` in
// `scan/data_skipping/stats_schema/mod.rs` for the full semantics.
/// Logical (post-DV) row count, stored as a `long`.
#[internal_api]
pub(crate) const NUM_RECORDS: &str = "numRecords";
/// Per-column null counts, as a nested struct mirroring the table schema.
#[internal_api]
pub(crate) const NULL_COUNT: &str = "nullCount";
/// Per-column lower bounds, as a nested struct mirroring the table schema.
#[internal_api]
pub(crate) const MIN_VALUES: &str = "minValues";
/// Per-column upper bounds, as a nested struct mirroring the table schema.
#[internal_api]
pub(crate) const MAX_VALUES: &str = "maxValues";
/// Whether the min/max/nullCount stats are tight or wide. Defaults to `true` when absent.
#[internal_api]
pub(crate) const TIGHT_BOUNDS: &str = "tightBounds";

/// Struct-encoded per-file statistics column (checkpoints with `writeStatsAsStruct=true`).
#[internal_api]
pub(crate) const STATS_PARSED: &str = "stats_parsed";

pub(crate) static ADD_SCHEMA: LazyLock<StructType> = LazyLock::new(Add::to_schema);

pub(crate) static ADD_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(ADD_NAME, ADD_SCHEMA.clone()));
pub(crate) static REMOVE_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(REMOVE_NAME, Remove::to_schema()));
pub(crate) static METADATA_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(METADATA_NAME, Metadata::to_schema()));
pub(crate) static PROTOCOL_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(PROTOCOL_NAME, Protocol::to_schema()));
pub(crate) static SET_TRANSACTION_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(SET_TRANSACTION_NAME, SetTransaction::to_schema()));
pub(crate) static COMMIT_INFO_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(COMMIT_INFO_NAME, CommitInfo::to_schema()));
pub(crate) static CDC_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(CDC_NAME, Cdc::to_schema()));
pub(crate) static DOMAIN_METADATA_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(DOMAIN_METADATA_NAME, DomainMetadata::to_schema()));
pub(crate) static CHECKPOINT_METADATA_FIELD: LazyLock<StructField> = LazyLock::new(|| {
    StructField::nullable(CHECKPOINT_METADATA_NAME, CheckpointMetadata::to_schema())
});
pub(crate) static SIDECAR_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(SIDECAR_NAME, Sidecar::to_schema()));

#[cfg(feature = "adaptive-metadata-in-dev")]
pub(crate) static CONTENT_ROOT_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable(CONTENT_ROOT_NAME, ContentRoot::to_schema()));

/// A `sidecar` element inside a `checkpoint` action array. Unlike the V2-checkpoint [`Sidecar`]
/// (which references spilled file actions), this references spilled user `txn` / `domainMetadata`
/// entries, discriminated by its `type` field (`"txn"` or `"domainMetadata"`). Its shape is a
/// [`Sidecar`] prefixed with that `type` column, so the schema is composed from
/// [`Sidecar::to_schema`] here rather than duplicated: `type` cannot be produced by [`ToSchema`]
/// (which stringifies the Rust identifier, and `r#type` stringifies to `"r#type"`).
#[cfg(feature = "adaptive-metadata-in-dev")]
static CONTENT_SIDECAR_FIELD: LazyLock<StructField> = LazyLock::new(|| {
    StructField::nullable(
        SIDECAR_NAME,
        schema! {
            not_null "type": STRING,
            ..(Sidecar::to_schema().into_fields()),
        },
    )
});

/// The `checkpoint` action serializes as an array whose elements are each one of the metadata
/// actions embedded in an adaptiveMetadata manifest commit. This schema is the union of every
/// element type that may appear in that array:
/// `checkpointMetadata`, `contentRoot`, `protocol`, `metaData`, `domainMetadata`, `txn`, `sidecar`.
#[cfg(feature = "adaptive-metadata-in-dev")]
static CHECKPOINT_ACTION_ELEMENT_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    (&CHECKPOINT_METADATA_FIELD),
    (&CONTENT_ROOT_FIELD),
    (&PROTOCOL_FIELD),
    (&METADATA_FIELD),
    (&DOMAIN_METADATA_FIELD),
    (&SET_TRANSACTION_FIELD),
    (&CONTENT_SIDECAR_FIELD),
};

#[cfg(feature = "adaptive-metadata-in-dev")]
pub(crate) static CHECKPOINT_ACTION_FIELD: LazyLock<StructField> = LazyLock::new(|| {
    StructField::nullable(
        CHECKPOINT_ACTION_NAME,
        ArrayType::new(CHECKPOINT_ACTION_ELEMENT_SCHEMA.clone(), false),
    )
});

/// The `lastManifestCommit` field as nested in commitInfo actions and CRC files.
#[cfg(feature = "adaptive-metadata-in-dev")]
pub(crate) static LAST_MANIFEST_COMMIT_FIELD: LazyLock<StructField> =
    LazyLock::new(|| StructField::nullable("lastManifestCommit", LastManifestCommit::to_schema()));

/// The `checkpoint` action field, present only under the `adaptive-metadata-in-dev` feature;
/// otherwise an empty iterator.
fn checkpoint_action_field() -> impl IntoIterator<Item = &'static StructField> {
    #[cfg(feature = "adaptive-metadata-in-dev")]
    {
        Some(&*CHECKPOINT_ACTION_FIELD)
    }
    #[cfg(not(feature = "adaptive-metadata-in-dev"))]
    {
        None::<&'static StructField>
    }
}

#[cfg(any(test, feature = "internal-api"))]
static COMMIT_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    (&ADD_FIELD),
    (&REMOVE_FIELD),
    (&METADATA_FIELD),
    (&PROTOCOL_FIELD),
    (&SET_TRANSACTION_FIELD),
    (&COMMIT_INFO_FIELD),
    (&CDC_FIELD),
    (&DOMAIN_METADATA_FIELD),
    ..(checkpoint_action_field()),
};

static ALL_ACTIONS_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    (&ADD_FIELD),
    (&REMOVE_FIELD),
    (&METADATA_FIELD),
    (&PROTOCOL_FIELD),
    (&SET_TRANSACTION_FIELD),
    (&COMMIT_INFO_FIELD),
    (&CDC_FIELD),
    (&DOMAIN_METADATA_FIELD),
    ..(checkpoint_action_field()),
    (&CHECKPOINT_METADATA_FIELD),
    (&SIDECAR_FIELD),
};

/// Schema for Add actions in the Delta log.
/// Wraps the Add action schema in a top-level struct with "add" field name.
#[internal_api]
pub(crate) static LOG_ADD_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! { (&ADD_FIELD) };

/// Schema for Remove actions in the Delta log.
/// Wraps the Remove action schema in a top-level struct with "remove" field name.
#[internal_api]
pub(crate) static LOG_REMOVE_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! { (&REMOVE_FIELD) };

#[internal_api]
pub(crate) static LOG_METADATA_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! { (&METADATA_FIELD) };

#[cfg(feature = "adaptive-metadata-in-dev")]
#[internal_api]
pub(crate) static LOG_CHECKPOINT_SCHEMA: LazyLock<SchemaRef> =
    lazy_schema_ref! { (&CHECKPOINT_ACTION_FIELD) };

#[internal_api]
pub(crate) static LOG_PROTOCOL_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! { (&PROTOCOL_FIELD) };

/// Schema for CommitInfo actions in the Delta log.
/// Wraps the CommitInfo schema in a top-level struct with "commitInfo" field name.
#[internal_api]
pub(crate) static LOG_COMMIT_INFO_SCHEMA: LazyLock<SchemaRef> =
    lazy_schema_ref! { (&COMMIT_INFO_FIELD) };

/// Schema for transaction (txn) actions in the Delta log.
/// Wraps the SetTransaction schema in a top-level struct with "txn" field name.
#[internal_api]
pub(crate) static LOG_TXN_SCHEMA: LazyLock<SchemaRef> =
    lazy_schema_ref! { (&SET_TRANSACTION_FIELD) };

#[internal_api]
pub(crate) static LOG_DOMAIN_METADATA_SCHEMA: LazyLock<SchemaRef> =
    lazy_schema_ref! { (&DOMAIN_METADATA_FIELD) };

#[cfg(any(test, feature = "internal-api"))]
#[internal_api]
/// Gets the schema for all actions that can appear in commits
/// logs.  This excludes actions that can only appear in checkpoints.
pub(crate) fn get_commit_schema() -> &'static SchemaRef {
    &COMMIT_SCHEMA
}

#[internal_api]
#[allow(dead_code)]
/// Gets a schema for all actions defined by the delta spec.
pub(crate) fn get_all_actions_schema() -> &'static SchemaRef {
    &ALL_ACTIONS_SCHEMA
}

/// Returns true if the schema contains file actions (add or remove)
/// columns.
#[internal_api]
pub(crate) fn schema_contains_file_actions(schema: &SchemaRef) -> bool {
    schema.contains(ADD_NAME) || schema.contains(REMOVE_NAME)
}

/// Nest an existing add action schema in an additional [`ADD_NAME`] struct.
///
/// This is useful for JSON conversion, as it allows us to wrap a dynamically maintained add action
/// schema in a top-level "add" struct.
pub(crate) fn as_log_add_schema(add_schema: SchemaRef) -> SchemaRef {
    schema_ref! { nullable ADD_NAME: (add_schema) }
}

// Serde derives are needed for CRC file deserialization (see `crc::reader`).
#[derive(
    Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, IntoStructData, TryFromStructData,
)]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct Format {
    /// Name of the encoding for files in this table
    pub(crate) provider: String,
    /// A map containing configuration options for the format
    pub(crate) options: HashMap<String, String>,
}

impl Default for Format {
    fn default() -> Self {
        Self {
            provider: String::from("parquet"),
            options: HashMap::new(),
        }
    }
}

// Serde derives are needed for CRC file deserialization (see `crc::reader`).
//
// TODO(#2446): `Metadata` stores the schema only as a JSON string. Callers that already hold
// a parsed `SchemaRef` (e.g. CREATE TABLE) serialize into `schema_string` and then re-parse
// downstream in `TableConfiguration::try_new` via `parse_schema()`. Caching the parsed schema
// on `Metadata` would eliminate the round-trip.
#[derive(
    Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, IntoStructData,
)]
#[serde(rename_all = "camelCase")]
pub struct Metadata {
    /// Unique identifier for this table
    id: String,
    /// User-provided identifier for this table
    name: Option<String>,
    /// User-provided description for this table
    description: Option<String>,
    /// Specification of the encoding for the files stored in the table
    format: Format,
    /// Schema of the table
    schema_string: String,
    /// Column names by which the data should be partitioned
    partition_columns: Vec<String>,
    /// The time when this metadata action is created, in milliseconds since the Unix epoch
    created_time: Option<i64>,
    /// Configuration options for the metadata action. These are parsed into [`TableProperties`].
    configuration: HashMap<String, String>,
}

impl Metadata {
    /// Reconstructs metadata from its serialized action fields.
    ///
    /// This constructor does not validate the schema, partition columns, format, or table
    /// configuration. Callers must validate the result before using it as table state.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn from_parts(
        id: String,
        name: Option<String>,
        description: Option<String>,
        format_provider: String,
        format_options: HashMap<String, String>,
        schema_string: String,
        partition_columns: Vec<String>,
        created_time: Option<i64>,
        configuration: HashMap<String, String>,
    ) -> Self {
        Self {
            id,
            name,
            description,
            format: Format {
                provider: format_provider,
                options: format_options,
            },
            schema_string,
            partition_columns,
            created_time,
            configuration,
        }
    }

    /// Create a new [`Metadata`] instances.
    ///
    /// # Errors
    ///
    /// Returns an error if there are any metadata columns in the schema.
    #[internal_api]
    pub(crate) fn try_new(
        name: Option<String>,
        description: Option<String>,
        schema: SchemaRef,
        partition_columns: Vec<String>,
        created_time: i64,
        configuration: HashMap<String, String>,
    ) -> KernelResult<Self> {
        // Validate that the schema does not contain metadata columns
        // Note: We don't have to look for nested metadata columns because that is already validated
        // when creating a StructType.
        if let Some(metadata_field) = schema.fields().find(|field| field.is_metadata_column()) {
            return Err(KernelError::Schema(format!(
                "Table schema must not contain metadata columns. Found metadata column: '{}'",
                metadata_field.name
            )));
        }

        Ok(Self {
            id: uuid::Uuid::new_v4().to_string(),
            name,
            description,
            // As of Delta Lake 0.3.0, user-facing APIs only allow the creation of tables where
            // format = 'parquet' and options = {}. Support for reading other formats is present
            // both for legacy reasons and to enable possible support for other formats in the
            // future (See delta-io/delta#87).
            format: Format::default(),
            schema_string: serde_json::to_string(&schema)?,
            partition_columns,
            created_time: Some(created_time),
            configuration,
        })
    }

    #[internal_api]
    pub(crate) fn try_new_from_data(data: &dyn EngineData) -> KernelResult<Option<Metadata>> {
        let mut visitor = MetadataVisitor::default();
        visitor.visit_rows_of(data)?;
        Ok(visitor.metadata)
    }

    // TODO(#1068/1069): make these just pub directly or make better internal_api macro for fields
    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn id(&self) -> &str {
        &self.id
    }

    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn name(&self) -> Option<&str> {
        self.name.as_deref()
    }

    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn description(&self) -> Option<&str> {
        self.description.as_deref()
    }

    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn created_time(&self) -> Option<i64> {
        self.created_time
    }

    #[internal_api]
    pub(crate) fn configuration(&self) -> &HashMap<String, String> {
        &self.configuration
    }

    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn format_provider(&self) -> &str {
        &self.format.provider
    }

    /// Returns the arbitrary format-specific options stored in this metadata action.
    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn format_options(&self) -> &HashMap<String, String> {
        &self.format.options
    }

    #[internal_api]
    pub(crate) fn schema_string(&self) -> &String {
        &self.schema_string
    }

    /// Parses the table schema from its JSON representation.
    ///
    /// # Errors
    ///
    /// Returns [`KernelError::Schema`] when the schema exceeds the supported decoding depth or
    /// declares a type the kernel doesn't support, or [`KernelError::MalformedJson`] for other
    /// JSON decoding failures.
    #[internal_api]
    pub(crate) fn parse_schema(&self) -> KernelResult<StructType> {
        // TODO(#1896): Increase the supported nesting depth or use non-recursive schema decoding.
        serde_json::from_str(&self.schema_string).map_err(|error| {
            // serde_json keeps ErrorCode::RecursionLimitExceeded private, so we use string
            // matching.
            if error.is_syntax()
                && error
                    .to_string()
                    .starts_with(SERDE_JSON_RECURSION_LIMIT_ERROR_PREFIX)
            {
                KernelError::schema(format!(
                    "Table schema is too deeply nested: decoding metaData.schemaString exceeded \
                     serde_json's recursion limit: {error}"
                ))
                .with_backtrace()
            } else if is_unsupported_delta_type_error(&error) {
                KernelError::schema(error.to_string()).with_backtrace()
            } else {
                error.into()
            }
        })
    }

    #[internal_api]
    pub(crate) fn partition_columns(&self) -> &[String] {
        &self.partition_columns
    }

    /// Parse the metadata configuration HashMap<String, String> into a TableProperties struct.
    /// Note that parsing is infallible -- any items that fail to parse are simply propagated
    /// through to the `TableProperties.unknown_properties` field.
    #[internal_api]
    pub(crate) fn parse_table_properties(&self) -> TableProperties {
        TableProperties::from(self.configuration.iter())
    }

    /// Returns a new Metadata with the schema replaced, preserving all other fields.
    ///
    /// # Errors
    ///
    /// Returns an error if schema serialization fails.
    pub(crate) fn with_schema(self, schema: SchemaRef) -> KernelResult<Self> {
        Ok(Self {
            schema_string: serde_json::to_string(&schema)?,
            ..self
        })
    }

    /// Returns a new Metadata with a single configuration entry inserted (or replaced),
    /// preserving all other configuration entries and metadata fields.
    pub(crate) fn with_configuration_entry(
        mut self,
        key: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.configuration.insert(key.into(), value.into());
        self
    }

    #[cfg(test)]
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new_unchecked(
        id: impl Into<String>,
        name: Option<String>,
        description: Option<String>,
        format: Format,
        schema_string: impl Into<String>,
        partition_columns: Vec<String>,
        created_time: Option<i64>,
        configuration: HashMap<String, String>,
    ) -> Self {
        Self {
            id: id.into(),
            name,
            description,
            format,
            schema_string: schema_string.into(),
            partition_columns,
            created_time,
            configuration,
        }
    }
}

#[derive(
    Default, Debug, Clone, PartialEq, Eq, ToSchema, IntoStructData, Serialize, Deserialize,
)]
// Deserialization goes through `ProtocolRaw` so every serde entry point (e.g. CRC files) is
// validated by `try_new`, like the JSON-replay path. Otherwise a CRC file could load a malformed
// feature shape that log replay would reject.
#[serde(rename_all = "camelCase", try_from = "ProtocolRaw")]
// TODO move to another module so that we disallow constructing this struct without using the
// try_new function.
pub struct Protocol {
    /// The minimum version of the Delta read protocol that a client must implement
    /// in order to correctly read this table
    min_reader_version: i32,
    /// The minimum version of the Delta write protocol that a client must implement
    /// in order to correctly write this table
    min_writer_version: i32,
    /// A collection of features that a client must implement in order to correctly
    /// read this table (exist only when minReaderVersion is set to 3)
    #[serde(skip_serializing_if = "Option::is_none")]
    reader_features: Option<Vec<TableFeature>>,
    /// A collection of features that a client must implement in order to correctly
    /// write this table (exist only when minWriterVersion is set to 7)
    #[serde(skip_serializing_if = "Option::is_none")]
    writer_features: Option<Vec<TableFeature>>,
}

/// Raw, unvalidated form of [`Protocol`] that serde reads before validation. Deserialize-only
/// (never serialized): `Protocol`'s `#[serde(try_from)]` converts it via [`Protocol::try_new`],
/// so every deserialization is validated.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct ProtocolRaw {
    min_reader_version: i32,
    min_writer_version: i32,
    reader_features: Option<Vec<TableFeature>>,
    writer_features: Option<Vec<TableFeature>>,
}

impl TryFrom<ProtocolRaw> for Protocol {
    type Error = KernelError;

    fn try_from(protocol: ProtocolRaw) -> KernelResult<Self> {
        Protocol::try_new(
            protocol.min_reader_version,
            protocol.min_writer_version,
            protocol.reader_features,
            protocol.writer_features,
        )
    }
}

/// Parse a list of feature identifiers into TableFeatures. Returns `None` for `None` input;
/// otherwise infallible (unrecognized names become `TableFeature::Unknown`).
fn parse_features(
    features: Option<impl IntoIterator<Item = impl Into<TableFeature>>>,
) -> Option<Vec<TableFeature>> {
    let features = features?.into_iter().map(Into::into);
    Some(features.collect())
}

impl Protocol {
    /// Try to create a new modern Protocol instance with the given table feature lists
    pub(crate) fn try_new_modern(
        reader_features: impl IntoIterator<Item = impl Into<TableFeature>>,
        writer_features: impl IntoIterator<Item = impl Into<TableFeature>>,
    ) -> KernelResult<Self> {
        Self::try_new(
            TABLE_FEATURES_MIN_READER_VERSION,
            TABLE_FEATURES_MIN_WRITER_VERSION,
            Some(reader_features),
            Some(writer_features),
        )
    }

    /// Try to create a new legacy Protocol instance with the given reader/writer versions
    #[cfg(test)]
    pub(crate) fn try_new_legacy(
        min_reader_version: i32,
        min_writer_version: i32,
    ) -> KernelResult<Self> {
        Self::try_new(
            min_reader_version,
            min_writer_version,
            TableFeature::NO_LIST,
            TableFeature::NO_LIST,
        )
    }

    /// Try to create a new Protocol instance from reader/writer versions and table features.
    #[internal_api]
    pub(crate) fn try_new(
        min_reader_version: i32,
        min_writer_version: i32,
        reader_features: Option<impl IntoIterator<Item = impl Into<TableFeature>>>,
        writer_features: Option<impl IntoIterator<Item = impl Into<TableFeature>>>,
    ) -> KernelResult<Self> {
        require!(
            min_reader_version >= MIN_VALID_RW_VERSION,
            KernelError::InvalidProtocol(format!(
                "min_reader_version must be >= {MIN_VALID_RW_VERSION}, got {min_reader_version}"
            ))
        );
        require!(
            min_writer_version >= MIN_VALID_RW_VERSION,
            KernelError::InvalidProtocol(format!(
                "min_writer_version must be >= {MIN_VALID_RW_VERSION}, got {min_writer_version}"
            ))
        );

        let reader_features = parse_features(reader_features);
        let writer_features = parse_features(writer_features);

        // The protocol states that Reader features may be present if and only if the
        // min_reader_version is 3
        if min_reader_version == TABLE_FEATURES_MIN_READER_VERSION {
            require!(
                reader_features.is_some(),
                KernelError::invalid_protocol(
                    "Reader features must be present when minimum reader version = 3"
                )
            );
        } else {
            require!(
                reader_features.is_none(),
                KernelError::invalid_protocol(
                    "Reader features must not be present when minimum reader version != 3"
                )
            );
        }

        // The protocol states that Writer features may be present if and only if the
        // min_writer_version is 7
        if min_writer_version == TABLE_FEATURES_MIN_WRITER_VERSION {
            require!(
                writer_features.is_some(),
                KernelError::invalid_protocol(
                    "Writer features must be present when minimum writer version = 7"
                )
            );
        } else {
            require!(
                writer_features.is_none(),
                KernelError::invalid_protocol(
                    "Writer features must not be present when minimum writer version != 7"
                )
            );
        }

        // Self- and cross-validate the reader and writer feature lists.
        match (&reader_features, &writer_features) {
            (Some(reader_features), Some(writer_features)) => {
                // Check all reader features are ReaderWriter and present in writer features.
                // Unknown features are treated as potentially ReaderWriter for forward
                // compatibility.
                if let Some(offending) = reader_features.iter().find(|feature| {
                    !matches!(
                        feature.feature_type(),
                        FeatureType::ReaderWriter | FeatureType::Unknown
                    ) || !writer_features.contains(*feature)
                }) {
                    return Err(KernelError::invalid_protocol(format!(
                        "Reader features must contain only ReaderWriter features that are also \
                         listed in writer features, but {offending:?} is not \
                         (readerFeatures={reader_features:?}, writerFeatures={writer_features:?}, \
                         minReaderVersion={min_reader_version}, minWriterVersion={min_writer_version})"
                    )));
                }

                // Every ReaderWriter feature in writerFeatures must also appear in readerFeatures.
                // Unknown features are treated as potentially Writer-only for forward
                // compatibility.
                //
                // Accept the legacy writer-list-only shape for delta-spark compatibility: a
                // past delta-spark bug produced (3, 7) tables with ColumnMapping in writerFeatures
                // only and an empty readerFeatures. Such tables still read correctly because the
                // mode comes from writerFeatures, and rejecting them would break existing
                // production tables. See #3110 to tighten this once such tables are migrated.
                //
                // Validate the whole writer list before warning: a non-legacy orphan rejects the
                // protocol outright, so we must not emit an acceptance warning for a legacy orphan
                // seen earlier in the list only to fail on a later one.
                let mut legacy_orphans = Vec::new();
                for feature in writer_features.iter() {
                    let orphaned_reader_writer_feature = feature.feature_type()
                        == FeatureType::ReaderWriter
                        && !reader_features.contains(feature);
                    if !orphaned_reader_writer_feature {
                        continue;
                    }
                    if LEGACY_READER_FEATURES.contains(feature) {
                        legacy_orphans.push(feature);
                    } else {
                        return Err(KernelError::invalid_protocol(format!(
                            "Writer features must be Writer-only or also listed in reader features, \
                             but ReaderWriter feature {feature:?} is listed in writerFeatures and \
                             missing from readerFeatures \
                             (readerFeatures={reader_features:?}, \
                             writerFeatures={writer_features:?}, \
                             minReaderVersion={min_reader_version}, \
                             minWriterVersion={min_writer_version})"
                        )));
                    }
                }
                // Reached only once the whole writer list is known valid.
                for feature in legacy_orphans {
                    warn!(
                        "ReaderWriter feature {feature:?} is listed in writerFeatures but \
                         missing from readerFeatures at minReaderVersion={min_reader_version}; \
                         treating it as reader-enabled (malformed protocol)"
                    );
                }
                Ok(())
            }
            (None, None) => Ok(()),
            (None, Some(writer_features)) => {
                // Special case: reader version 2 implies ColumnMapping support.
                // All other ReaderWriter features require explicit reader_features list (reader
                // version 3). Unknown features are treated as potentially
                // Writer-only for forward compatibility.
                if let Some(offending) = writer_features.iter().find(|feature| {
                    match feature.feature_type() {
                        FeatureType::WriterOnly | FeatureType::Unknown => false,
                        FeatureType::ReaderWriter => {
                            // ColumnMapping is allowed when reader version is 2 (implied support)
                            !(min_reader_version == 2 && *feature == &TableFeature::ColumnMapping)
                        }
                    }
                }) {
                    return Err(KernelError::invalid_protocol(format!(
                        "Writer features must be Writer-only or also listed in reader features, \
                         but ReaderWriter feature {offending:?} is listed in writerFeatures with \
                         no reader features present \
                         (writerFeatures={writer_features:?}, minReaderVersion={min_reader_version}, \
                         minWriterVersion={min_writer_version})"
                    )));
                }
                Ok(())
            }
            (Some(_), None) => Err(KernelError::invalid_protocol(
                "Reader features should be present in writer features",
            )),
        }?;

        Ok(Protocol {
            min_reader_version,
            min_writer_version,
            reader_features,
            writer_features,
        })
    }

    /// Create a new Protocol by visiting the EngineData and extracting the first protocol row into
    /// a Protocol instance. If no protocol row is found, returns Ok(None).
    pub(crate) fn try_new_from_data(data: &dyn EngineData) -> KernelResult<Option<Protocol>> {
        let mut visitor = ProtocolVisitor::default();
        visitor.visit_rows_of(data)?;
        Ok(visitor.protocol)
    }

    /// This protocol's minimum reader version
    #[internal_api]
    pub(crate) fn min_reader_version(&self) -> i32 {
        self.min_reader_version
    }

    /// This protocol's minimum writer version
    #[internal_api]
    pub(crate) fn min_writer_version(&self) -> i32 {
        self.min_writer_version
    }

    /// Get the reader features for the protocol
    #[internal_api]
    pub(crate) fn reader_features(&self) -> Option<&[TableFeature]> {
        self.reader_features.as_deref()
    }

    /// Get the writer features for the protocol
    #[internal_api]
    pub(crate) fn writer_features(&self) -> Option<&[TableFeature]> {
        self.writer_features.as_deref()
    }

    /// True if this protocol has the requested feature
    pub(crate) fn has_table_feature(&self, feature: &TableFeature) -> bool {
        // Since each reader features is a subset of writer features, we only check writer feature
        self.writer_features()
            .is_some_and(|features| features.contains(feature))
    }

    #[cfg(test)]
    pub(crate) fn new_unchecked(
        min_reader_version: i32,
        min_writer_version: i32,
        reader_features: Option<Vec<TableFeature>>,
        writer_features: Option<Vec<TableFeature>>,
    ) -> Self {
        Self {
            min_reader_version,
            min_writer_version,
            reader_features,
            writer_features,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, ToSchema, IntoStructData)]
#[internal_api]
#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
pub(crate) struct CommitInfo {
    /// The time this logical file was created, as milliseconds since the epoch.
    /// Read: optional, write: required (that is, kernel always writes).
    pub(crate) timestamp: Option<i64>,
    /// The time this logical file was created, as milliseconds since the epoch. Unlike
    /// `timestamp`, this field is guaranteed to be monotonically increase with each commit.
    /// Note: If in-commit timestamps are enabled, both the following must be true:
    /// - The `inCommitTimestamp` field must always be present in CommitInfo.
    /// - The CommitInfo action must always be the first one in a commit.
    pub(crate) in_commit_timestamp: Option<i64>,
    /// An arbitrary string that identifies the operation associated with this commit. This is
    /// specified by the engine. Read: optional, write: required (that is, kernel alwarys writes).
    pub(crate) operation: Option<String>,
    /// Map of arbitrary string key-value pairs that provide additional information about the
    /// operation. This is specified by the connector and may be populated on write.
    pub(crate) operation_parameters: Option<HashMap<String, Option<String>>>,
    /// Map of arbitrary string key-value pairs that provide operation metrics.
    /// This is specified by the engine.
    pub(crate) operation_metrics: Option<HashMap<String, Option<String>>>,
    /// The version of the delta_kernel crate used to write this commit. The kernel will always
    /// write this field, but it is optional since many tables will not have this field (i.e. any
    /// tables not written by kernel).
    pub(crate) kernel_version: Option<String>,
    /// Whether this commit is a blind append.
    pub(crate) is_blind_append: Option<bool>,
    /// A place for the engine to store additional metadata associated with this commit
    pub(crate) engine_info: Option<String>,
    /// A unique transaction identifier for this commit.
    pub(crate) txn_id: Option<String>,
    /// Map of tags associated with this commit.
    pub(crate) tags: Option<HashMap<String, Option<String>>>,
    /// Whether this commit changes the table's logical records. Required on every commit when
    /// adaptiveMetadata is enabled; readers then treat it as the source of truth over the
    /// per-file-action `dataChange` flags.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) data_change: Option<bool>,
    /// Identifies the latest manifest commit up to this version. Absent until the table's first
    /// manifest commit (adaptiveMetadata).
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) last_manifest_commit: Option<LastManifestCommit>,
}

impl CommitInfo {
    pub(crate) fn new(
        timestamp: i64,
        in_commit_timestamp: Option<i64>,
        operation: Option<String>,
        engine_info: Option<String>,
        is_blind_append: bool,
    ) -> Self {
        Self {
            timestamp: Some(timestamp),
            in_commit_timestamp,
            operation: Some(operation.unwrap_or_else(|| UNKNOWN_OPERATION.to_string())),
            operation_parameters: Some(HashMap::new()),
            operation_metrics: None,
            kernel_version: Some(format!("v{KERNEL_VERSION}")),
            is_blind_append: is_blind_append.then_some(true),
            engine_info,
            txn_id: Some(uuid::Uuid::new_v4().to_string()),
            tags: None,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            data_change: None,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            last_manifest_commit: None,
        }
    }

    pub(crate) fn set_row_tracking_preserved(&mut self) {
        self.tags.get_or_insert_default().insert(
            ROW_TRACKING_PRESERVED_TAG.to_string(),
            Some("true".to_string()),
        );
    }

    pub(crate) fn set_operation_parameters(
        &mut self,
        operation_parameters: HashMap<String, Option<String>>,
    ) {
        self.operation_parameters = Some(operation_parameters);
    }

    pub(crate) fn set_operation_metrics(
        &mut self,
        operation_metrics: HashMap<String, Option<String>>,
    ) {
        self.operation_metrics = Some(operation_metrics);
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    pub(crate) fn set_data_change(&mut self, data_change: bool) {
        self.data_change = Some(data_change);
    }

    /// Merges the supplied tags into this CommitInfo's tags.
    ///
    /// Existing values take precedence when both maps contain the same key.
    pub(crate) fn merge_tags(&mut self, tags: Option<HashMap<String, Option<String>>>) {
        let Some(tags) = tags else {
            return;
        };
        let current_tags = self.tags.get_or_insert_default();
        for (key, value) in tags {
            current_tags.entry(key).or_insert(value);
        }
    }
}

/// Identifies the location of a file's existing entry within the adaptive metadata tree, pointing
/// at a specific position in a leaf manifest.
///
/// A back reference lets a writer locate an existing tree entry without scanning entire leaf
/// manifests. It is meaningful only relative to a specific tree version. See the
/// [Iceberg V4 metadata RFC].
///
/// [Iceberg V4 metadata RFC]: https://github.com/delta-io/delta/blob/master/protocol_rfcs/iceberg-v4-metadata.md#backreferences
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, PartialEq, Eq, ToSchema, IntoStructData, Deserialize)]
#[cfg_attr(test, derive(Serialize))]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct BackReference {
    /// Path to the leaf manifest containing this file, relative to the table root
    /// (e.g. `metadata/leaf-m1.parquet`). Resolved by joining the table location and this path
    /// with a `/` separator, so it must not start with `/`.
    pub(crate) manifest: String,
    /// Row position (0-indexed) of the file entry within the manifest.
    pub(crate) pos: i32,
}

#[derive(Debug, Clone, PartialEq, Eq, ToSchema, IntoStructData, Deserialize)]
#[cfg_attr(test, derive(Serialize, Default))]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct Add {
    /// A relative path to a data file from the root of the table or an absolute path to a file
    /// that should be added to the table. The path is a URI as specified by
    /// [RFC 2396 URI Generic Syntax], which needs to be decoded to get the data file path.
    ///
    /// [RFC 2396 URI Generic Syntax]: https://www.ietf.org/rfc/rfc2396.txt
    pub(crate) path: String,

    /// A map from partition column to value for this logical file. This map can contain null in
    /// the values meaning a partition is null. We drop those values from this map, due to the
    /// `allow_null_container_values` annotation allowing them and because [`materialize`] drops
    /// null values. This means an engine can assume that if a partition is found in
    /// [`Metadata::partition_columns`] but not in this map, its value is null.
    ///
    /// [`materialize`]: crate::engine_data::MapItem::materialize
    #[allow_null_container_values]
    #[serde(deserialize_with = "deserialize_partition_values")]
    pub(crate) partition_values: HashMap<String, String>,

    /// The size of this data file in bytes
    pub(crate) size: i64,

    /// The time this logical file was created, as milliseconds since the epoch.
    pub(crate) modification_time: i64,

    /// When `false` the logical file must already be present in the table or the records
    /// in the added file must be contained in one or more remove actions in the same version.
    pub(crate) data_change: bool,

    /// Contains [statistics] (e.g., count, min/max values for columns) about the data in this
    /// logical file encoded as a JSON string.
    ///
    /// [statistics]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Per-file-Statistics
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub stats: Option<String>,

    /// Map containing metadata about this logical file.
    /// Note: map values can be null.
    /// We don't use `#[allow_null_container_values]` here because [`MapItem::materialize`]
    /// drops null values when that attribute is present.
    ///
    /// [`MapItem::materialize`]: crate::engine_data::MapItem::materialize
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub tags: Option<HashMap<String, Option<String>>>,

    /// Information about deletion vector (DV) associated with this add action
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub deletion_vector: Option<DeletionVectorDescriptor>,

    /// Default generated Row ID of the first row in the file. The default generated Row IDs
    /// of the other rows in the file can be reconstructed by adding the physical index of the
    /// row within the file to the base Row ID.
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub base_row_id: Option<i64>,

    /// First commit version in which an add action with the same path was committed to the table.
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub default_row_commit_version: Option<i64>,

    /// The name of the clustering implementation
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub clustering_provider: Option<String>,

    /// Back reference into the adaptive metadata tree. Present only when this `add` re-adds a file
    /// that has no paired `remove` (e.g. stats backfilling); otherwise absent.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) back_reference: Option<BackReference>,
}

fn deserialize_partition_values<'de, D>(
    deserializer: D,
) -> KernelResult<HashMap<String, String>, D::Error>
where
    D: Deserializer<'de>,
{
    struct PartitionValuesVisitor;

    impl<'de> Visitor<'de> for PartitionValuesVisitor {
        type Value = HashMap<String, String>;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            formatter.write_str("a map of nullable partition values")
        }

        fn visit_map<A>(self, mut map: A) -> KernelResult<Self::Value, A::Error>
        where
            A: MapAccess<'de>,
        {
            let mut values = HashMap::new();
            while let Some((key, value)) = map.next_entry::<String, Option<String>>()? {
                match value {
                    Some(value) => {
                        values.insert(key, value);
                    }
                    None => {
                        values.remove(&key);
                    }
                }
            }
            Ok(values)
        }
    }

    deserializer.deserialize_map(PartitionValuesVisitor)
}

impl Add {
    /// Returns the URI-encoded data-file path.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn path(&self) -> &str {
        &self.path
    }

    /// Returns non-null partition values. Missing partition columns represent null values.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn partition_values(&self) -> &HashMap<String, String> {
        &self.partition_values
    }

    /// Returns the data-file size in bytes.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn size(&self) -> i64 {
        self.size
    }

    /// Returns the file modification time in milliseconds since the Unix epoch.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn modification_time(&self) -> i64 {
        self.modification_time
    }

    /// Returns whether the action changes table data.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn data_change(&self) -> bool {
        self.data_change
    }

    /// Returns this action's adaptive-metadata back reference, if present.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn back_reference(&self) -> Option<&BackReference> {
        self.back_reference.as_ref()
    }

    /// Reconstructs an Add action from its serialized fields.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn from_parts(
        path: String,
        partition_values: HashMap<String, String>,
        size: i64,
        modification_time: i64,
        data_change: bool,
        stats: Option<String>,
        tags: Option<HashMap<String, Option<String>>>,
        deletion_vector: Option<DeletionVectorDescriptor>,
        base_row_id: Option<i64>,
        default_row_commit_version: Option<i64>,
        clustering_provider: Option<String>,
    ) -> Self {
        Self {
            path,
            partition_values,
            size,
            modification_time,
            data_change,
            stats,
            tags,
            deletion_vector,
            base_row_id,
            default_row_commit_version,
            clustering_provider,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            back_reference: None,
        }
    }

    #[internal_api]
    #[allow(dead_code)]
    pub(crate) fn dv_unique_id(
        &self,
        dv_normalization_root: Option<&Url>,
    ) -> KernelResult<Option<String>> {
        self.deletion_vector
            .as_ref()
            .map(|dv| {
                DeletionVectorDescriptor::unique_id_with_root(
                    &dv.storage_type.to_string(),
                    &dv.path_or_inline_dv,
                    dv.offset,
                    dv_normalization_root,
                )
            })
            .transpose()
    }
}

#[derive(Debug, Clone, PartialEq, Eq, ToSchema)]
#[internal_api]
#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
pub(crate) struct Remove {
    /// A relative path to a data file from the root of the table or an absolute path to a file
    /// that should be added to the table. The path is a URI as specified by
    /// [RFC 2396 URI Generic Syntax], which needs to be decoded to get the data file path.
    ///
    /// [RFC 2396 URI Generic Syntax]: https://www.ietf.org/rfc/rfc2396.txt
    pub(crate) path: String,

    /// The time this logical file was created, as milliseconds since the epoch.
    ///
    /// Must be null when adaptiveMetadata is enabled on the table since metadata cleanup
    /// uses tree reachability instead of timestamp-based expiration.
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) deletion_timestamp: Option<i64>,

    /// When `false` the logical file must already be present in the table or the records
    /// in the added file must be contained in one or more remove actions in the same version.
    pub(crate) data_change: bool,

    /// When true, the fields `partition_values` and `size` are present
    ///
    /// Must be true when adaptiveMetadata is enabled on the table.
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) extended_file_metadata: Option<bool>,

    /// A map from partition column to value for this logical file. This map can contain null in
    /// the values meaning a partition is null. We drop those values from this map, due to the
    /// `allow_null_container_values` annotation allowing them and because [`materialize`] drops
    /// null values. This means an engine can assume that if a partition is found in
    /// [`Metadata::partition_columns`] but not in this map, its value is null.
    ///
    /// [`materialize`]: crate::engine_data::EngineMap::materialize
    #[allow_null_container_values]
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) partition_values: Option<HashMap<String, String>>,

    /// The size of this data file in bytes
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) size: Option<i64>,

    /// Contains [statistics] (e.g., count, min/max values for columns) about the data in this
    /// logical file encoded as a JSON string.
    ///
    /// Must be set when adaptiveMetadata is enabled on the table.
    ///
    /// [statistics]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Per-file-Statistics
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub stats: Option<String>,

    /// Map containing metadata about this logical file. Values can be null.
    #[allow_null_container_values]
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) tags: Option<HashMap<String, String>>,

    /// Information about deletion vector (DV) associated with this add action
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) deletion_vector: Option<DeletionVectorDescriptor>,

    /// Default generated Row ID of the first row in the file. The default generated Row IDs
    /// of the other rows in the file can be reconstructed by adding the physical index of the
    /// row within the file to the base Row ID
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) base_row_id: Option<i64>,

    /// First commit version in which an add action with the same path was committed to the table.
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) default_row_commit_version: Option<i64>,

    /// Back reference into the adaptive metadata tree. Required when the file's entry lives in a
    /// leaf manifest; absent when the file has no leaf-manifest entry (it has no entry in the
    /// tree, or its entry is inline in the root manifest).
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
    pub(crate) back_reference: Option<BackReference>,
}

#[derive(Debug, Clone, PartialEq, Eq, ToSchema)]
#[internal_api]
#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
pub(crate) struct Cdc {
    /// A relative path to a change data file from the root of the table or an absolute path to a
    /// change data file that should be added to the table. The path is a URI as specified by
    /// [RFC 2396 URI Generic Syntax], which needs to be decoded to get the file path.
    ///
    /// [RFC 2396 URI Generic Syntax]: https://www.ietf.org/rfc/rfc2396.txt
    pub path: String,

    /// A map from partition column to value for this logical file. This map can contain null in
    /// the values meaning a partition is null. We drop those values from this map, due to the
    /// `allow_null_container_values` annotation allowing them and because [`materialize`] drops
    /// null values. This means an engine can assume that if a partition is found in
    /// [`Metadata::partition_columns`] but not in this map, its value is null.
    ///
    /// [`materialize`]: crate::engine_data::MapItem::materialize
    #[allow_null_container_values]
    pub partition_values: HashMap<String, String>,

    /// The size of this cdc file in bytes
    pub size: i64,

    /// When `false` the logical file must already be present in the table or the records
    /// in the added file must be contained in one or more remove actions in the same version.
    ///
    /// Should always be set to false for `cdc` actions because they *do not* change the underlying
    /// data of the table
    pub data_change: bool,

    /// Map containing metadata about this logical file. Values can be null.
    #[allow_null_container_values]
    pub tags: Option<HashMap<String, String>>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[internal_api]
#[derive(Constructor, IntoStructData, ToSchema)]
pub(crate) struct SetTransaction {
    /// A unique identifier for the application performing the transaction.
    pub(crate) app_id: String,

    /// An application-specific numeric identifier for this transaction.
    pub(crate) version: i64,

    /// The time when this transaction action was created in milliseconds since the Unix epoch.
    pub(crate) last_updated: Option<i64>,
}

impl SetTransaction {
    /// Returns the application identifier.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn app_id(&self) -> &str {
        &self.app_id
    }

    /// Returns the application-specific transaction version.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn version(&self) -> i64 {
        self.version
    }

    /// Returns the last-updated timestamp, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn last_updated(&self) -> Option<i64> {
        self.last_updated
    }

    /// Whether this transaction is expired: `last_updated <= expiration_timestamp` with both
    /// present. A `None` `last_updated` (no timestamp recorded) or a `None` `expiration_timestamp`
    /// (no retention duration configured) never expires.
    pub(crate) fn is_expired(&self, expiration_timestamp: Option<i64>) -> bool {
        matches!(
            (expiration_timestamp, self.last_updated),
            (Some(exp_ts), Some(lu)) if lu <= exp_ts
        )
    }

    /// This transaction's `version`, unless it is expired under `expiration_timestamp`.
    pub(crate) fn non_expired_version(&self, expiration_timestamp: Option<i64>) -> Option<i64> {
        (!self.is_expired(expiration_timestamp)).then_some(self.version)
    }
}

/// Reference to a root of an adaptive metadata tree.
///
/// Contains the path, size, and version of the root manifest file.
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, PartialEq, Eq, ToSchema, IntoStructData, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[internal_api]
#[cfg_attr(test, derive(Default))]
pub(crate) struct ContentRoot {
    /// Path to the root manifest file. It is absolute if it begins with an [RFC 3986] URI scheme
    /// (e.g. `s3://bucket/...`); otherwise it is relative and resolved against the table root by
    /// concatenation with a `/` separator, matching the [Iceberg V4 relative paths specification].
    /// Unlike [`Add`]/[`Remove`] paths, this is not RFC 2396 percent-encoded.
    ///
    /// [RFC 3986]: https://datatracker.ietf.org/doc/html/rfc3986#section-3.1
    /// [Iceberg V4 relative paths specification]: https://iceberg.apache.org/spec/#paths-in-metadata
    pub(crate) path: String,
    /// Size of the root manifest file in bytes. Not exposed directly -- use
    /// [`CheckpointAction::root_filemeta`] to get a validated [`FileMeta`].
    size_in_bytes: i64,
    /// The table version the root manifest reflects. This is
    /// `<= checkpointMetadata.version`: equal in a manifest commit, and strictly less in a
    /// standalone checkpoint (where inline file actions cover the gap up to the checkpoint
    /// version). Distinct from [`CheckpointAction::version`], which is
    /// `checkpointMetadata.version`.
    version: i64,
}

/// Identifies the latest manifest commit up to a given table version.
///
/// Recorded on the `commitInfo` action and in the version checksum (`.crc`) file so readers can
/// locate the most recent `checkpoint` action without scanning the log. See the
/// [adaptiveMetadata RFC].
///
/// [adaptiveMetadata RFC]: https://github.com/delta-io/delta/pull/6978
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, PartialEq, Eq, ToSchema, IntoStructData, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct LastManifestCommit {
    /// Version of the manifest commit that emitted the latest [`CheckpointAction`].
    pub(crate) version: i64,
    /// The [`ContentRoot::version`] of that checkpoint action. Never newer than [`Self::version`].
    pub(crate) content_root_version: i64,
}

#[cfg(feature = "adaptive-metadata-in-dev")]
impl LastManifestCommit {
    /// Builds a reference to the manifest commit at `version` whose checkpoint action's content
    /// root reflects `content_root_version`.
    ///
    /// Enforces the adaptiveMetadata invariants that both versions are non-negative and the
    /// referenced content root version never exceeds the manifest commit version, so an invalid
    /// pair can never be constructed. Mirrors the validation on `CheckpointAction`.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn new(version: i64, content_root_version: i64) -> KernelResult<Self> {
        let last_manifest_commit = LastManifestCommit {
            version,
            content_root_version,
        };
        last_manifest_commit.validate()?;
        Ok(last_manifest_commit)
    }

    /// Enforce the adaptiveMetadata invariants that `contentRootVersion` is non-negative and never
    /// exceeds the manifest commit `version`. Because [`LastManifestCommit`] derives
    /// [`Deserialize`], values parsed from JSON bypass [`Self::new`], so callers that deserialize
    /// must invoke this explicitly.
    pub(crate) fn validate(&self) -> KernelResult<()> {
        require!(
            self.content_root_version >= 0,
            KernelError::generic(format!(
                "lastManifestCommit contentRootVersion {} must be non-negative",
                self.content_root_version
            ))
        );
        require!(
            self.content_root_version <= self.version,
            KernelError::generic(format!(
                "lastManifestCommit contentRootVersion {} exceeds version {}",
                self.content_root_version, self.version
            ))
        );
        Ok(())
    }
}

/// The checkpoint action embeds metadata tree state in a Delta log entry.
///
/// When a manifest commit occurs, the Delta log entry contains a `checkpoint` action that
/// references a root manifest file. The `version` field indicates the table version up to
/// which the checkpoint is complete. For manifest commits, the checkpoint action also contains
/// the table protocol and metadata, making the commit self-contained with respect to P+M.
///
/// Example manifest-commit JSON:
/// ```json
/// { "checkpoint": [
///     { "checkpointMetadata": { "version": 42 } },
///     { "contentRoot": { "path": "...", "sizeInBytes": 1024, "version": 42 } },
///     { "protocol": { ... } },
///     { "metaData": { ... } },
///     { "txn": { ... } },
///     { "domainMetadata": { ... } },
///     { "sidecar": { "type": "txn", "path": "...", "sizeInBytes": 1024, "modificationTime": 0 } }
///   ]
/// }
/// ```
// Serde is hand-written (see below), not derived: the wire form is a JSON array of tagged element
// objects (`[{"checkpointMetadata":..}, {"contentRoot":..}, ..]`), not a struct. This is the same
// shape the EngineData path uses, so the `_last_checkpoint` hint (which serdes this action) and log
// replay share the wire form and the enumerated invariants (required singletons, no duplicates,
// known sidecar `type`, `contentRoot.version <= checkpointMetadata.version`). They intentionally
// differ on unknown elements: log replay skips a future element kind for forward compatibility (see
// `visitors::CheckpointElementVisitor::visit`), whereas the serde path fails closed on it (an
// externally-tagged enum with no catch-all), so a hint carrying one is dropped and the reader falls
// back to log replay. Used by the hint's `AmtCheckpoint.checkpoint` field.
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, PartialEq, Eq)]
#[internal_api]
pub(crate) struct CheckpointAction {
    /// The table version up to which the checkpoint is complete, sourced from the wire
    /// `checkpointMetadata.version`. May be less than or equal to the commit version containing
    /// this checkpoint action, and is `>= content_root.version` (see [`ContentRoot::version`]).
    pub(crate) version: i64,
    /// Reference to the root manifest file.
    pub(crate) content_root: ContentRoot,
    /// The table protocol at the checkpoint version.
    pub(crate) protocol: Protocol,
    /// The table metadata at the checkpoint version.
    pub(crate) metadata: Metadata,
    /// Inline `txn` ([`SetTransaction`]) entries carried in the checkpoint array.
    pub(crate) transactions: Vec<SetTransaction>,
    /// Inline `domainMetadata` ([`DomainMetadata`]) entries carried in the checkpoint array.
    pub(crate) domain_metadata: Vec<DomainMetadata>,
    /// `sidecar` entries of type `txn`, referencing spilled [`SetTransaction`] actions.
    pub(crate) txn_sidecars: Vec<Sidecar>,
    /// `sidecar` entries of type `domainMetadata`, referencing spilled [`DomainMetadata`] actions.
    pub(crate) domain_metadata_sidecars: Vec<Sidecar>,
}

// === CheckpointAction <-> JSON (array of tagged elements) ===

/// One element of a [`CheckpointAction`]'s serialized array (the adaptiveMetadata "Checkpoint
/// Action" wire form). A checkpoint action serializes as a JSON array of single-key tagged objects,
/// so this is an externally-tagged enum keyed by the action name, reusing kernel's action structs
/// to yield
/// the same types as log replay. Having no catch-all variant, it fails the parse on an unrecognized
/// action key -- deliberately fail-closed, unlike the forward-compatible EngineData
/// `CheckpointElementVisitor`, which skips unknown elements. A hint carrying a future
/// element kind is thus dropped and the reader falls back to log replay.
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) enum CheckpointActionElement {
    CheckpointMetadata(CheckpointMetadata),
    ContentRoot(ContentRoot),
    Protocol(Protocol),
    #[serde(rename = "metaData")]
    Metadata(Metadata),
    DomainMetadata(DomainMetadata),
    Txn(SetTransaction),
    Sidecar(CheckpointSidecar),
}

/// A `sidecar` element inside a checkpoint action array. The wire form prefixes the [`Sidecar`]
/// fields with a `type` discriminator (`"txn"` or `"domainMetadata"`) identifying which action kind
/// the sidecar spills; [`Sidecar`] itself carries no type, so it is flattened in alongside it.
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct CheckpointSidecar {
    /// The action kind this sidecar spills: `"txn"` or `"domainMetadata"`.
    #[serde(rename = "type")]
    pub(crate) sidecar_type: String,
    #[serde(flatten)]
    pub(crate) sidecar: Sidecar,
}

#[cfg(feature = "adaptive-metadata-in-dev")]
impl Serialize for CheckpointAction {
    /// Emits the array of tagged elements in the same canonical order as
    /// `try_into_scalar` (`checkpointMetadata`, `contentRoot`, `protocol`, `metaData`,
    /// then `txn`, `domainMetadata`, and the `txn`/`domainMetadata` sidecars). Validates first,
    /// so an invalid action can never be written through serde either.
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> KernelResult<S::Ok, S::Error> {
        self.validate().map_err(serde::ser::Error::custom)?;
        let checkpoint_metadata = CheckpointMetadata {
            version: self.version,
            tags: None,
        };
        let mut elements = vec![
            CheckpointActionElement::CheckpointMetadata(checkpoint_metadata),
            CheckpointActionElement::ContentRoot(self.content_root.clone()),
            CheckpointActionElement::Protocol(self.protocol.clone()),
            CheckpointActionElement::Metadata(self.metadata.clone()),
        ];
        elements.extend(
            self.transactions
                .iter()
                .cloned()
                .map(CheckpointActionElement::Txn),
        );
        elements.extend(
            self.domain_metadata
                .iter()
                .cloned()
                .map(CheckpointActionElement::DomainMetadata),
        );
        let sidecar_element = |type_str: &str, sidecar: &Sidecar| {
            CheckpointActionElement::Sidecar(CheckpointSidecar {
                sidecar_type: type_str.to_string(),
                sidecar: sidecar.clone(),
            })
        };
        elements.extend(
            self.txn_sidecars
                .iter()
                .map(|s| sidecar_element(SET_TRANSACTION_NAME, s)),
        );
        elements.extend(
            self.domain_metadata_sidecars
                .iter()
                .map(|s| sidecar_element(DOMAIN_METADATA_NAME, s)),
        );
        elements.serialize(serializer)
    }
}

#[cfg(feature = "adaptive-metadata-in-dev")]
impl<'de> Deserialize<'de> for CheckpointAction {
    /// Folds the array of tagged elements into a typed action, applying the same enumerated
    /// checks as the EngineData [`visitors::CheckpointVisitor`]: required singletons, no
    /// duplicates, known sidecar `type`, and the `contentRoot.version <=
    /// checkpointMetadata.version` invariant. The one intended difference is unknown elements:
    /// this path fails closed on an unrecognized element key (see [`CheckpointActionElement`]),
    /// whereas the visitor skips it for forward compatibility.
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> KernelResult<Self, D::Error> {
        let elements = Vec::<CheckpointActionElement>::deserialize(deserializer)?;
        Self::from_elements(elements).map_err(serde::de::Error::custom)
    }
}

// === CheckpointAction -> EngineData ===

/// Build the `sidecar` element payload: a [`Sidecar`] scalar prefixed with a `type` discriminator
/// (`"txn"` or `"domainMetadata"`), matching [`CONTENT_SIDECAR_FIELD`].
#[cfg(feature = "adaptive-metadata-in-dev")]
fn content_sidecar_element(type_str: &str, sidecar: Sidecar) -> KernelResult<Scalar> {
    let sidecar: StructData = sidecar.into();
    let fields = std::iter::once(StructField::not_null("type", DataType::STRING))
        .chain(sidecar.fields().iter().cloned());
    let values = std::iter::once(Scalar::from(type_str))
        .chain(sidecar.values().iter().cloned())
        .collect();
    let fields = StructType::try_new(fields)?.into_fields().collect();
    Ok(Scalar::Struct(StructData::try_new(fields, values)?))
}

/// Wrap a single element `value` into a full union struct matching the checkpoint array's element
/// type: the field named `field_name` holds `value`, every other field is a typed null.
#[cfg(feature = "adaptive-metadata-in-dev")]
fn checkpoint_action_union_element(field_name: &str, value: Scalar) -> KernelResult<Scalar> {
    let fields: Vec<StructField> = CHECKPOINT_ACTION_ELEMENT_SCHEMA.fields().cloned().collect();
    require!(
        fields.iter().any(|f| f.name() == field_name),
        KernelError::generic(format!(
            "checkpoint union element field {field_name:?} not found in element schema"
        ))
    );
    let values = fields
        .iter()
        .map(|field| {
            if field.name() == field_name {
                value.clone()
            } else {
                Scalar::null(field.data_type().clone())
            }
        })
        .collect();
    Ok(Scalar::Struct(StructData::try_new(fields, values)?))
}

#[cfg(feature = "adaptive-metadata-in-dev")]
impl CheckpointAction {
    /// Encodes this action as its single `checkpoint` column [`Scalar`]: an array whose elements
    /// are a union struct (one field per action kind). Unlike the other actions, it has no derived
    /// struct-scalar conversion because that nested array-of-union shape can't be expressed by the
    /// derive, so we build the `Scalar::Array` by hand. This is also where the action is validated,
    /// hence a fallible method rather than an infallible `From`.
    fn try_into_scalar(self) -> KernelResult<Scalar> {
        self.validate()?;
        let checkpoint_metadata = CheckpointMetadata {
            version: self.version,
            tags: None,
        };
        let mut elements = vec![
            checkpoint_action_union_element(CHECKPOINT_METADATA_NAME, checkpoint_metadata.into())?,
            checkpoint_action_union_element(CONTENT_ROOT_NAME, self.content_root.into())?,
            checkpoint_action_union_element(PROTOCOL_NAME, self.protocol.into())?,
            checkpoint_action_union_element(METADATA_NAME, self.metadata.into())?,
        ];
        for txn in self.transactions {
            elements.push(checkpoint_action_union_element(
                SET_TRANSACTION_NAME,
                txn.into(),
            )?);
        }
        for dm in self.domain_metadata {
            elements.push(checkpoint_action_union_element(
                DOMAIN_METADATA_NAME,
                dm.into(),
            )?);
        }
        for sidecar in self.txn_sidecars {
            let element = content_sidecar_element(SET_TRANSACTION_NAME, sidecar)?;
            elements.push(checkpoint_action_union_element(SIDECAR_NAME, element)?);
        }
        for sidecar in self.domain_metadata_sidecars {
            let element = content_sidecar_element(DOMAIN_METADATA_NAME, sidecar)?;
            elements.push(checkpoint_action_union_element(SIDECAR_NAME, element)?);
        }

        let array_type = ArrayType::new(CHECKPOINT_ACTION_ELEMENT_SCHEMA.clone(), false);
        Ok(Scalar::Array(ArrayData::try_new(array_type, elements)?))
    }
}

#[cfg(feature = "adaptive-metadata-in-dev")]
impl ContentRoot {
    /// Builds a reference to a root manifest at `path`, `size_in_bytes`, reflecting `version`.
    pub(crate) fn new(path: String, size_in_bytes: i64, version: i64) -> Self {
        ContentRoot {
            path,
            size_in_bytes,
            version,
        }
    }
}

#[cfg(feature = "adaptive-metadata-in-dev")]
impl CheckpointAction {
    /// Builds a checkpoint action at `version` with all `transactions` and `domain_metadata`
    /// inlined and no sidecars.
    // TODO(#2866): spill transactions and domain metadata into sidecars once the adaptiveMetadata
    // sidecar format is defined.
    #[internal_api]
    pub(crate) fn new(
        version: i64,
        content_root: ContentRoot,
        protocol: Protocol,
        metadata: Metadata,
        transactions: Vec<SetTransaction>,
        domain_metadata: Vec<DomainMetadata>,
    ) -> Self {
        CheckpointAction {
            version,
            content_root,
            protocol,
            metadata,
            transactions,
            domain_metadata,
            txn_sidecars: vec![],
            domain_metadata_sidecars: vec![],
        }
    }

    /// Serialize this checkpoint action into a single-row `EngineData`.
    #[internal_api]
    pub(crate) fn into_engine_data(self, engine: &dyn Engine) -> KernelResult<Box<dyn EngineData>> {
        create_row(
            engine,
            LOG_CHECKPOINT_SCHEMA.clone(),
            self.try_into_scalar()?,
        )
    }

    /// Parse the first `checkpoint` action in `data`, ignoring any later ones. Rows without a
    /// `checkpoint` action are skipped, so `Ok(None)` means the batch had none at all.
    ///
    /// Returns an error if a `checkpoint` action is present but malformed: a required singleton
    /// element is missing or repeated, an element or sidecar `type` is unrecognized, or
    /// `contentRoot.version` exceeds `checkpointMetadata.version`.
    #[internal_api]
    pub(crate) fn try_new_from_data(
        data: &dyn EngineData,
    ) -> KernelResult<Option<CheckpointAction>> {
        let mut visitor = visitors::CheckpointVisitor::default();
        visitor.visit_rows_of(data)?;
        Ok(visitor.checkpoint)
    }

    /// Folds a deserialized array of tagged [`CheckpointActionElement`]s into a checkpoint
    /// action, mirroring [`visitors::CheckpointVisitor`]: the four required elements are singletons
    /// (missing or repeated is an error), `txn`/`domainMetadata` are collected, `sidecar` entries
    /// are split by their `type` (unknown types error), and the assembled action is validated.
    fn from_elements(elements: Vec<CheckpointActionElement>) -> KernelResult<Self> {
        fn set_once<T>(slot: &mut Option<T>, value: T, name: &str) -> KernelResult<()> {
            require!(
                slot.replace(value).is_none(),
                KernelError::generic(format!("duplicate `{name}` element in checkpoint action"))
            );
            Ok(())
        }

        let mut version = None;
        let mut content_root = None;
        let mut protocol = None;
        let mut metadata = None;
        let mut transactions = Vec::new();
        let mut domain_metadata = Vec::new();
        let mut txn_sidecars = Vec::new();
        let mut domain_metadata_sidecars = Vec::new();
        for element in elements {
            match element {
                CheckpointActionElement::CheckpointMetadata(cm) => {
                    set_once(&mut version, cm.version, CHECKPOINT_METADATA_NAME)?
                }
                CheckpointActionElement::ContentRoot(cr) => {
                    set_once(&mut content_root, cr, CONTENT_ROOT_NAME)?
                }
                CheckpointActionElement::Protocol(p) => set_once(&mut protocol, p, PROTOCOL_NAME)?,
                CheckpointActionElement::Metadata(m) => set_once(&mut metadata, m, METADATA_NAME)?,
                CheckpointActionElement::Txn(t) => transactions.push(t),
                CheckpointActionElement::DomainMetadata(dm) => domain_metadata.push(dm),
                CheckpointActionElement::Sidecar(cs) => match cs.sidecar_type.as_str() {
                    SET_TRANSACTION_NAME => txn_sidecars.push(cs.sidecar),
                    DOMAIN_METADATA_NAME => domain_metadata_sidecars.push(cs.sidecar),
                    other => {
                        return Err(KernelError::generic(format!(
                            "checkpoint sidecar has unsupported type `{other}`"
                        )))
                    }
                },
            }
        }

        let missing = |field: &str| {
            KernelError::generic(format!(
                "checkpoint action is missing required `{field}` element"
            ))
        };
        let action = CheckpointAction {
            version: version.ok_or_else(|| missing(CHECKPOINT_METADATA_NAME))?,
            content_root: content_root.ok_or_else(|| missing(CONTENT_ROOT_NAME))?,
            protocol: protocol.ok_or_else(|| missing(PROTOCOL_NAME))?,
            metadata: metadata.ok_or_else(|| missing(METADATA_NAME))?,
            transactions,
            domain_metadata,
            txn_sidecars,
            domain_metadata_sidecars,
        };
        action.validate()?;
        Ok(action)
    }

    /// Enforce the adaptiveMetadata invariant that `contentRoot.version` never exceeds the
    /// checkpoint version. Called on both the parse and serialize paths so a `CheckpointAction`
    /// can never be written in a shape the reader would reject.
    fn validate(&self) -> KernelResult<()> {
        require!(
            self.content_root.version <= self.version,
            KernelError::generic(format!(
                "checkpoint contentRoot.version {} exceeds checkpointMetadata.version {}",
                self.content_root.version, self.version
            ))
        );
        Ok(())
    }

    /// Path to the root manifest file (delegates to the nested [`ContentRoot`]).
    #[internal_api]
    pub(crate) fn path(&self) -> &str {
        &self.content_root.path
    }

    /// The table version the root manifest reflects (delegates to the nested [`ContentRoot`]).
    #[internal_api]
    pub(crate) fn content_root_version(&self) -> i64 {
        self.content_root.version
    }

    /// Get the checkpoint version.
    #[internal_api]
    pub(crate) fn version(&self) -> i64 {
        self.version
    }

    /// Convert the referenced root manifest into a [`FileMeta`] for engine I/O.
    ///
    /// The `contentRoot` path is absolute if it has a URI scheme, otherwise it is resolved relative
    /// to `table_root` by concatenation with a single `/` separator, matching Iceberg V4's
    /// [relative paths specification].
    ///
    /// Returns an error if the resolved location fails to parse as a [`Url`], or if the size does
    /// not fit a [`crate::FileSize`].
    ///
    /// [relative paths specification]: https://iceberg.apache.org/spec/#paths-in-metadata
    #[internal_api]
    pub(crate) fn root_filemeta(&self, table_root: &Url) -> KernelResult<FileMeta> {
        let content_root = &self.content_root;
        Ok(FileMeta {
            location: resolve_amt_location(&content_root.path, table_root)?,
            last_modified: i64::MAX,
            size: to_file_size(content_root.size_in_bytes, "checkpoint contentRoot")?,
        })
    }

    /// The table protocol embedded in this checkpoint action (at [`Self::version`]).
    #[internal_api]
    pub(crate) fn protocol(&self) -> &Protocol {
        &self.protocol
    }

    /// The table metadata embedded in this checkpoint action (at [`Self::version`]).
    #[internal_api]
    pub(crate) fn metadata(&self) -> &Metadata {
        &self.metadata
    }
}

/// The sidecar action references a sidecar file which provides some of the checkpoint's
/// file actions. This action is only allowed in checkpoints following the V2 spec.
///
/// [More info]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#sidecar-file-information
#[derive(ToSchema, IntoStructData, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct Sidecar {
    /// A path to a sidecar file that can be either:
    /// - A relative path (just the file name) within the `_delta_log/_sidecars` directory.
    /// - An absolute path
    /// The path is a URI as specified by [RFC 2396 URI Generic Syntax], which needs to be decoded
    /// to get the file path.
    ///
    /// [RFC 2396 URI Generic Syntax]: https://www.ietf.org/rfc/rfc2396.txt
    pub path: String,

    /// The size of the sidecar file in bytes.
    pub size_in_bytes: i64,

    /// The time this logical file was created, as milliseconds since the epoch.
    pub modification_time: i64,

    /// A map containing any additional metadata about the logical file. Values can be null.
    #[allow_null_container_values]
    pub tags: Option<HashMap<String, String>>,
}

/// Convert an `i64` byte count from a log action into a [`FileSize`], erroring with `context` (a
/// short action name, e.g. `"sidecar"`) and the offending value when it is negative.
fn to_file_size(bytes: i64, context: &str) -> KernelResult<FileSize> {
    bytes.try_into().map_err(|_| {
        KernelError::generic(format!(
            "Failed to convert {context} size {bytes} to FileSize"
        ))
    })
}

impl Sidecar {
    /// Creates a sidecar action.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn new(
        path: String,
        size_in_bytes: i64,
        modification_time: i64,
        tags: Option<HashMap<String, String>>,
    ) -> Self {
        Self {
            path,
            size_in_bytes,
            modification_time,
            tags,
        }
    }

    /// Convert a Sidecar record to a FileMeta.
    ///
    /// This helper first builds the URL by joining the provided log_root with
    /// the "_sidecars/" folder and the given sidecar path.
    pub(crate) fn to_filemeta(&self, log_root: &Url) -> KernelResult<FileMeta> {
        Ok(FileMeta {
            location: log_root.join("_sidecars/")?.join(&self.path)?,
            last_modified: self.modification_time,
            size: to_file_size(self.size_in_bytes, "sidecar")?,
        })
    }
}

/// The CheckpointMetadata action describes details about a checkpoint following the V2
/// specification.
///
/// [More info]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#checkpoint-metadata
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
#[internal_api]
#[derive(Constructor, IntoStructData, ToSchema)]
pub(crate) struct CheckpointMetadata {
    /// The version of the V2 spec checkpoint.
    ///
    /// Currently using `i64` for compatibility with other actions' representations.
    /// Future work will address converting numeric fields to unsigned types (e.g., `u64`) where
    /// semantically appropriate (e.g., for version, size, timestamps, etc.).
    /// See issue #786 for tracking progress.
    pub(crate) version: i64,

    /// Map containing any additional metadata about the V2 spec checkpoint. Values can be null.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[allow_null_container_values]
    pub(crate) tags: Option<HashMap<String, String>>,
}

impl CheckpointMetadata {
    /// Returns the checkpoint version.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn version(&self) -> i64 {
        self.version
    }

    /// Returns checkpoint tags, preserving absent versus empty maps.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn tags(&self) -> Option<&HashMap<String, String>> {
        self.tags.as_ref()
    }
}

/// The [DomainMetadata] action contains a configuration (string) for a named metadata domain. Two
/// overlapping transactions conflict if they both contain a domain metadata action for the same
/// metadata domain.
///
/// Note that the `delta.*` domain is reserved for internal use.
///
/// [DomainMetadata]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#domain-metadata
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema, IntoStructData)]
pub struct DomainMetadata {
    domain: String,
    configuration: String,
    removed: bool,
}

impl DomainMetadata {
    /// Create a new DomainMetadata action.
    #[internal_api]
    pub(crate) fn new(domain: String, configuration: String) -> Self {
        Self {
            domain,
            configuration,
            removed: false,
        }
    }

    /// Create a new DomainMetadata action to remove a domain.
    #[internal_api]
    pub(crate) fn remove(domain: String, configuration: String) -> Self {
        Self {
            domain,
            configuration,
            removed: true,
        }
    }

    // returns true if the domain metadata is an system-controlled domain (all domains that start
    // with "delta.")
    #[allow(unused)]
    #[internal_api]
    pub(crate) fn is_internal(&self) -> bool {
        self.domain.starts_with(INTERNAL_DOMAIN_PREFIX)
    }

    pub fn domain(&self) -> &str {
        &self.domain
    }

    pub fn configuration(&self) -> &str {
        &self.configuration
    }

    /// Returns `true` if this action is a tombstone (marking domain removal).
    pub fn is_removed(&self) -> bool {
        self.removed
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use rstest::rstest;
    use serde_json::json;

    use super::*;
    use crate::arrow::array::{
        Array, Int32Array, ListBuilder, RecordBatch, StringBuilder, StructArray,
    };
    use crate::arrow::datatypes::{DataType as ArrowDataType, Field, Schema};
    use crate::arrow::json::ReaderBuilder;
    use crate::engine::arrow_data::EngineDataArrowExt as _;
    use crate::engine::arrow_expression::ArrowEvaluationHandler;
    #[cfg(feature = "adaptive-metadata-in-dev")]
    use crate::engine::to_json_bytes;
    #[cfg(feature = "adaptive-metadata-in-dev")]
    use crate::engine_data::FilteredEngineData;
    use crate::expressions::{Scalar, StructData};
    use crate::schema::{schema, schema_ref, DataType, MapType, StructField};
    use crate::unit_test_utils::assert_result_error_with_message;
    use crate::{
        create_row, Engine, EvaluationHandler, JsonHandler, ParquetHandler, StorageHandler,
    };

    #[rstest]
    #[case::add(ADD_NAME, Some("path"))]
    #[case::remove(REMOVE_NAME, Some("path"))]
    #[case::metadata(METADATA_NAME, Some("id"))]
    #[case::protocol(PROTOCOL_NAME, Some("minReaderVersion"))]
    #[case::transaction(SET_TRANSACTION_NAME, Some("appId"))]
    #[case::cdc(CDC_NAME, Some("path"))]
    #[case::domain_metadata(DOMAIN_METADATA_NAME, Some("domain"))]
    #[case::checkpoint_metadata(CHECKPOINT_METADATA_NAME, Some("version"))]
    #[case::sidecar(SIDECAR_NAME, Some("path"))]
    #[case::witnessless_action(COMMIT_INFO_NAME, None)]
    #[case::unknown_action("futureAction", None)]
    fn test_action_presence_leaf(#[case] action_name: &str, #[case] expected_leaf: Option<&str>) {
        assert_eq!(action_presence_leaf(action_name), expected_leaf);
    }

    // duplicated
    struct ExprEngine(Arc<dyn EvaluationHandler>);

    impl ExprEngine {
        fn new() -> Self {
            ExprEngine(Arc::new(ArrowEvaluationHandler))
        }
    }

    impl Engine for ExprEngine {
        fn evaluation_handler(&self) -> Arc<dyn EvaluationHandler> {
            self.0.clone()
        }

        fn json_handler(&self) -> Arc<dyn JsonHandler> {
            unimplemented!()
        }

        fn parquet_handler(&self) -> Arc<dyn ParquetHandler> {
            unimplemented!()
        }

        fn storage_handler(&self) -> Arc<dyn StorageHandler> {
            unimplemented!()
        }
    }

    #[rstest]
    #[case::no_expiration_configured(None, Some(1000), false)]
    #[case::null_last_updated_never_expires(Some(5000), None, false)]
    #[case::both_none(None, None, false)]
    #[case::last_updated_before_expiration(Some(2000), Some(1000), true)]
    #[case::last_updated_at_expiration(Some(1000), Some(1000), true)]
    #[case::last_updated_after_expiration(Some(2000), Some(3000), false)]
    fn test_set_transaction_expiration(
        #[case] expiration_timestamp: Option<i64>,
        #[case] last_updated: Option<i64>,
        #[case] expired: bool,
    ) {
        let txn = SetTransaction::new("app".to_string(), 7, last_updated);
        assert_eq!(txn.is_expired(expiration_timestamp), expired);
        assert_eq!(
            txn.non_expired_version(expiration_timestamp),
            (!expired).then_some(7)
        );
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_last_manifest_commit_schema() {
        let expected = schema! {
            not_null "version": LONG,
            not_null "contentRootVersion": LONG,
        };
        assert_eq!(LastManifestCommit::to_schema(), expected);
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest]
    #[case::equal(5, 5, true)]
    #[case::content_root_older(5, 3, true)]
    #[case::content_root_newer(3, 5, false)]
    fn test_last_manifest_commit_new_validates(
        #[case] version: i64,
        #[case] content_root_version: i64,
        #[case] ok: bool,
    ) {
        let result = LastManifestCommit::new(version, content_root_version);
        assert_eq!(result.is_ok(), ok);
    }

    #[test]
    fn test_metadata_schema() {
        let schema = get_commit_schema()
            .project(&[METADATA_NAME])
            .expect("Couldn't get metaData field");

        let expected = schema_ref! {
            nullable "metaData": {
                not_null "id": STRING,
                nullable "name": STRING,
                nullable "description": STRING,
                not_null "format": {
                    not_null "provider": STRING,
                    not_null "options": { STRING => not_null STRING },
                },
                not_null "schemaString": STRING,
                not_null "partitionColumns": [ not_null STRING ],
                nullable "createdTime": LONG,
                not_null "configuration": { STRING => not_null STRING },
            },
        };
        assert_eq!(schema, expected);
    }

    #[rstest]
    #[case::supported(41, false)]
    #[case::exceeded(42, true)]
    fn parse_schema_nesting_boundary(#[case] depth: usize, #[case] exceeds_limit: bool) {
        let metadata = Metadata {
            schema_string: serde_json::to_string(&nested_schema(depth)).unwrap(),
            ..Default::default()
        };

        let result = metadata.parse_schema();
        if exceeds_limit {
            assert_result_error_with_message(
                result.as_ref(),
                concat!(
                    "Schema error: Table schema is too deeply nested: decoding ",
                    "metaData.schemaString exceeded serde_json's ",
                    "recursion limit: recursion limit exceeded"
                ),
            );
            let error = match result.unwrap_err() {
                KernelError::Backtraced { source, .. } => *source,
                error => error,
            };
            assert!(matches!(error, KernelError::Schema(_)));
        } else {
            result.unwrap();
        }
    }

    #[rstest]
    // Syntax error -> MalformedJson.
    #[case::malformed_syntax("{", "MalformedJson")]
    // Data error lacking the unsupported-type prefix (invalid decimal) -> MalformedJson, NOT
    // Schema: the reclassification must not fire for every is_data error.
    #[case::malformed_bad_decimal(
        r#"{"type":"struct","fields":[{"name":"t","type":"decimal(nope)","nullable":true,"metadata":{}}]}"#,
        "MalformedJson"
    )]
    // Regression guard: a well-formed schema whose wrong-typed field value echoes the prefix must
    // stay MalformedJson. `nullable` is a bool, so a string value is an is_data error whose message
    // *contains* the prefix; matching by `starts_with` keeps it out of the Schema arm.
    #[case::malformed_value_echoes_prefix(
        r#"{"type":"struct","fields":[{"name":"t","type":"string","nullable":"Unsupported Delta table type","metadata":{}}]}"#,
        "MalformedJson"
    )]
    // Unsupported primitive types -> Schema.
    #[case::unsupported_time(
        r#"{"type":"struct","fields":[{"name":"t","type":"time(6)","nullable":true,"metadata":{}}]}"#,
        "Schema"
    )]
    #[case::unsupported_interval(
        r#"{"type":"struct","fields":[{"name":"t","type":"interval week","nullable":true,"metadata":{}}]}"#,
        "Schema"
    )]
    // Unsupported primitive nested inside a struct field -> Schema (reclassification is
    // position-agnostic, not limited to top-level columns).
    #[case::unsupported_nested(
        r#"{"type":"struct","fields":[{"name":"t","type":{"type":"struct","fields":[{"name":"inner","type":"time(6)","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}}]}"#,
        "Schema"
    )]
    // Unknown complex type -> Schema.
    #[case::unsupported_complex(
        r#"{"type":"struct","fields":[{"name":"t","type":{"type":"matrix"},"nullable":true,"metadata":{}}]}"#,
        "Schema"
    )]
    fn parse_schema_error_classification(
        #[case] schema_string: &str,
        #[case] expected_error: &str,
    ) {
        let metadata = Metadata {
            schema_string: schema_string.to_string(),
            ..Default::default()
        };
        // Error conversion captures a backtrace only when enabled, so normalize both forms before
        // checking the underlying error.
        let error = match metadata.parse_schema().unwrap_err() {
            KernelError::Backtraced { source, .. } => *source,
            error => error,
        };
        match expected_error {
            "MalformedJson" => {
                assert!(
                    matches!(error, KernelError::MalformedJson(_)),
                    "got: {error:?}"
                )
            }
            "Schema" => {
                assert!(matches!(error, KernelError::Schema(_)), "got: {error:?}")
            }
            other => panic!("unknown expected_error discriminant: {other}"),
        }
    }

    fn nested_schema(depth: usize) -> StructType {
        (0..depth).fold(
            schema! { nullable "leaf": INTEGER },
            |nested, depth| schema! { nullable (format!("level_{depth}")): (nested) },
        )
    }

    #[test]
    fn test_add_schema() {
        let schema = get_commit_schema()
            .project(&[ADD_NAME])
            .expect("Couldn't get add field");

        #[cfg(feature = "adaptive-metadata-in-dev")]
        let expected = schema_ref! {
            nullable "add": {
                not_null "path": STRING,
                not_null "partitionValues": { STRING => nullable STRING },
                not_null "size": LONG,
                not_null "modificationTime": LONG,
                not_null "dataChange": BOOLEAN,
                nullable "stats": STRING,
                nullable "tags": { STRING => nullable STRING },
                (deletion_vector_field()),
                nullable "baseRowId": LONG,
                nullable "defaultRowCommitVersion": LONG,
                nullable "clusteringProvider": STRING,
                nullable "backReference": (BackReference::to_schema()),
            },
        };
        #[cfg(not(feature = "adaptive-metadata-in-dev"))]
        let expected = schema_ref! {
            nullable "add": {
                not_null "path": STRING,
                not_null "partitionValues": { STRING => nullable STRING },
                not_null "size": LONG,
                not_null "modificationTime": LONG,
                not_null "dataChange": BOOLEAN,
                nullable "stats": STRING,
                nullable "tags": { STRING => nullable STRING },
                (deletion_vector_field()),
                nullable "baseRowId": LONG,
                nullable "defaultRowCommitVersion": LONG,
                nullable "clusteringProvider": STRING,
            },
        };
        assert_eq!(schema, expected);
    }

    fn tags_field() -> StructField {
        StructField::nullable(
            "tags",
            MapType::new(DataType::STRING, DataType::STRING, true),
        )
    }

    fn partition_values_field() -> StructField {
        StructField::nullable(
            "partitionValues",
            MapType::new(DataType::STRING, DataType::STRING, true),
        )
    }

    fn deletion_vector_field() -> StructField {
        StructField::nullable(
            "deletionVector",
            schema! {
                not_null "storageType": STRING,
                not_null "pathOrInlineDv": STRING,
                nullable "offset": INTEGER,
                not_null "sizeInBytes": INTEGER,
                not_null "cardinality": LONG,
            },
        )
    }

    #[test]
    fn test_remove_schema() {
        let schema = get_commit_schema()
            .project(&[REMOVE_NAME])
            .expect("Couldn't get remove field");
        #[cfg(feature = "adaptive-metadata-in-dev")]
        let expected = schema_ref! {
            nullable "remove": {
                not_null "path": STRING,
                nullable "deletionTimestamp": LONG,
                not_null "dataChange": BOOLEAN,
                nullable "extendedFileMetadata": BOOLEAN,
                (partition_values_field()),
                nullable "size": LONG,
                nullable "stats": STRING,
                (tags_field()),
                (deletion_vector_field()),
                nullable "baseRowId": LONG,
                nullable "defaultRowCommitVersion": LONG,
                nullable "backReference": (BackReference::to_schema()),
            },
        };
        #[cfg(not(feature = "adaptive-metadata-in-dev"))]
        let expected = schema_ref! {
            nullable "remove": {
                not_null "path": STRING,
                nullable "deletionTimestamp": LONG,
                not_null "dataChange": BOOLEAN,
                nullable "extendedFileMetadata": BOOLEAN,
                (partition_values_field()),
                nullable "size": LONG,
                nullable "stats": STRING,
                (tags_field()),
                (deletion_vector_field()),
                nullable "baseRowId": LONG,
                nullable "defaultRowCommitVersion": LONG,
            },
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_cdc_schema() {
        let schema = get_commit_schema()
            .project(&[CDC_NAME])
            .expect("Couldn't get cdc field");
        let expected = schema_ref! {
            nullable "cdc": {
                not_null "path": STRING,
                not_null "partitionValues": { STRING => nullable STRING },
                not_null "size": LONG,
                not_null "dataChange": BOOLEAN,
                (tags_field()),
            },
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_sidecar_schema() {
        let schema = Sidecar::to_schema();
        let expected = schema! {
            not_null "path": STRING,
            not_null "sizeInBytes": LONG,
            not_null "modificationTime": LONG,
            (tags_field()),
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_checkpoint_metadata_schema() {
        let schema = get_all_actions_schema()
            .project(&[CHECKPOINT_METADATA_NAME])
            .expect("Couldn't get checkpointMetadata field");
        let expected = schema_ref! {
            nullable "checkpointMetadata": {
                not_null "version": LONG,
                (tags_field()),
            },
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_transaction_schema() {
        let schema = get_commit_schema()
            .project(&["txn"])
            .expect("Couldn't get transaction field");

        let expected = schema_ref! {
            nullable "txn": {
                not_null "appId": STRING,
                not_null "version": LONG,
                nullable "lastUpdated": LONG,
            },
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_commit_info_schema() {
        let schema = get_commit_schema()
            .project(&["commitInfo"])
            .expect("Couldn't get commitInfo field");

        #[cfg(feature = "adaptive-metadata-in-dev")]
        let expected = schema_ref! {
            nullable "commitInfo": {
                nullable "timestamp": LONG,
                nullable "inCommitTimestamp": LONG,
                nullable "operation": STRING,
                nullable "operationParameters": { STRING => nullable STRING },
                nullable "operationMetrics": { STRING => nullable STRING },
                nullable "kernelVersion": STRING,
                nullable "isBlindAppend": BOOLEAN,
                nullable "engineInfo": STRING,
                nullable "txnId": STRING,
                nullable "tags": { STRING => nullable STRING },
                nullable "dataChange": BOOLEAN,
                nullable "lastManifestCommit": {
                    not_null "version": LONG,
                    not_null "contentRootVersion": LONG,
                },
            },
        };
        #[cfg(not(feature = "adaptive-metadata-in-dev"))]
        let expected = schema_ref! {
            nullable "commitInfo": {
                nullable "timestamp": LONG,
                nullable "inCommitTimestamp": LONG,
                nullable "operation": STRING,
                nullable "operationParameters": { STRING => nullable STRING },
                nullable "operationMetrics": { STRING => nullable STRING },
                nullable "kernelVersion": STRING,
                nullable "isBlindAppend": BOOLEAN,
                nullable "engineInfo": STRING,
                nullable "txnId": STRING,
                nullable "tags": { STRING => nullable STRING },
            },
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_domain_metadata_schema() {
        let schema = get_commit_schema()
            .project(&[DOMAIN_METADATA_NAME])
            .expect("Couldn't get domainMetadata field");
        let expected = schema_ref! {
            nullable "domainMetadata": {
                not_null "domain": STRING,
                not_null "configuration": STRING,
                not_null "removed": BOOLEAN,
            },
        };
        assert_eq!(schema, expected);
    }

    #[test]
    fn test_validate_protocol() {
        let invalid_protocols = [
            Protocol {
                min_reader_version: 3,
                min_writer_version: 7,
                reader_features: None,
                writer_features: Some(vec![]),
            },
            Protocol {
                min_reader_version: 3,
                min_writer_version: 7,
                reader_features: Some(vec![]),
                writer_features: None,
            },
            Protocol {
                min_reader_version: 3,
                min_writer_version: 7,
                reader_features: None,
                writer_features: None,
            },
        ];
        for Protocol {
            min_reader_version,
            min_writer_version,
            reader_features,
            writer_features,
        } in invalid_protocols
        {
            assert!(matches!(
                Protocol::try_new(
                    min_reader_version,
                    min_writer_version,
                    reader_features,
                    writer_features
                ),
                Err(KernelError::InvalidProtocol(_)),
            ));
        }
    }

    #[rstest]
    #[case(0, 1)]
    #[case(1, 0)]
    #[case(-1, 2)]
    #[case(1, -1)]
    fn reject_protocol_version_below_minimum(#[case] rv: i32, #[case] wv: i32) {
        let expected = if rv < 1 {
            format!("Invalid protocol action in the delta log: min_reader_version must be >= 1, got {rv}")
        } else {
            format!("Invalid protocol action in the delta log: min_writer_version must be >= 1, got {wv}")
        };
        assert_result_error_with_message(
            Protocol::try_new(rv, wv, TableFeature::NO_LIST, TableFeature::NO_LIST),
            &expected,
        );
    }

    #[test]
    fn accept_min_versions() {
        let p = Protocol::try_new_legacy(1, 1).unwrap();
        assert_eq!(p.min_reader_version(), 1);
        assert_eq!(p.min_writer_version(), 1);
    }

    #[test]
    fn test_validate_table_features_invalid() {
        // (reader_feature, writer_feature)
        let invalid_features = [
            // ReaderWriter feature not present in writer features
            (
                vec![TableFeature::DeletionVectors],
                vec![TableFeature::AppendOnly],
                "Reader features must contain only ReaderWriter features that are also listed in writer features",
            ),
            (
                vec![TableFeature::DeletionVectors],
                vec![],
                "Reader features must contain only ReaderWriter features that are also listed in writer features",
            ),
            // ReaderWriter feature not present in reader features
            (
                vec![],
                vec![TableFeature::DeletionVectors],
                "Writer features must be Writer-only or also listed in reader features",
            ),
            (
                vec![TableFeature::VariantType],
                vec![
                    TableFeature::VariantType,
                    TableFeature::DeletionVectors,
                ],
                "Writer features must be Writer-only or also listed in reader features",
            ),
            // WriterOnly feature present in reader features
            (
                vec![TableFeature::AppendOnly],
                vec![TableFeature::AppendOnly],
                "Reader features must contain only ReaderWriter features that are also listed in writer features",
            ),
        ];

        for (reader_features, writer_features, error_msg) in invalid_features {
            let res = Protocol::try_new_modern(reader_features, writer_features);
            // The error message is enriched with the offending feature and the parsed
            // feature lists, so match on a prefix rather than the whole string.
            assert!(
                matches!(
                    &res,
                    Err(KernelError::InvalidProtocol(error)) if error.to_string().contains(error_msg)
                ),
                "Expected message containing:\t{error_msg}\nBut got:{res:?}\n"
            );
        }
    }

    #[test]
    fn test_validate_table_features_unknown() {
        // Unknown features are allowed during validation for forward compatibility,
        // but will be rejected when trying to use the protocol (ensure_operation_supported)

        // Test unknown features in reader - validation passes
        let protocol = Protocol::try_new_modern(
            vec![TableFeature::Unknown("unknown_reader".to_string())],
            vec![TableFeature::Unknown("unknown_reader".to_string())],
        );
        assert!(protocol.is_ok());

        // Test unknown features in writer - validation passes
        let protocol = Protocol::try_new_modern(
            TableFeature::EMPTY_LIST,
            vec![TableFeature::Unknown("unknown_writer".to_string())],
        );
        assert!(protocol.is_ok());
    }

    #[test]
    fn test_validate_table_features_valid() {
        // (reader_feature, writer_feature)
        let valid_features = [
            // ReaderWriter feature present in both reader/writer features,
            // WriterOnly feature present in writer feature
            (
                vec![TableFeature::DeletionVectors],
                vec![TableFeature::DeletionVectors],
            ),
            (vec![], vec![TableFeature::AppendOnly]),
            (
                vec![TableFeature::VariantType],
                vec![TableFeature::VariantType, TableFeature::AppendOnly],
            ),
            // Unknown feature may be ReaderWriter or WriterOnly (for forward compatibility)
            (
                vec![TableFeature::Unknown("rw".to_string())],
                vec![
                    TableFeature::Unknown("rw".to_string()),
                    TableFeature::Unknown("w".to_string()),
                ],
            ),
            // Empty feature set is valid
            (vec![], vec![]),
        ];

        for (reader_features, writer_features) in valid_features {
            assert!(Protocol::try_new_modern(reader_features, writer_features).is_ok());
        }
    }

    #[test]
    fn test_validate_legacy_column_mapping_valid() {
        // Valid: ColumnMapping with reader v2
        // Reader version 2 implies columnMapping support (no explicit reader_features)
        // Writer version 7 requires explicit writer_features list
        let protocol = Protocol::try_new(
            2,
            7,
            TableFeature::NO_LIST,
            Some(vec![TableFeature::ColumnMapping]),
        );
        assert!(protocol.is_ok());
    }

    #[test]
    fn test_validate_legacy_writer_only_features_valid() {
        // Valid: Writer-only features with reader v1
        let protocol = Protocol::try_new(
            1,
            7,
            TableFeature::NO_LIST,
            Some(vec![TableFeature::AppendOnly]),
        );
        assert!(protocol.is_ok());
    }

    #[test]
    fn test_validate_legacy_column_mapping_with_writer_features_valid() {
        // Valid: Mix of Writer-only and ColumnMapping with reader v2
        let protocol = Protocol::try_new(
            2,
            7,
            TableFeature::NO_LIST,
            Some(vec![TableFeature::AppendOnly, TableFeature::ColumnMapping]),
        );
        assert!(protocol.is_ok());
    }

    #[test]
    fn test_validate_column_mapping_reader_v1_invalid() {
        // Invalid: ColumnMapping with reader v1
        // Reader v1 doesn't imply any ReaderWriter features
        let protocol = Protocol::try_new(
            1,
            7,
            TableFeature::NO_LIST,
            Some(vec![TableFeature::ColumnMapping]),
        );
        assert!(protocol.is_err());
    }

    #[test]
    fn test_validate_multiple_readerwriter_features_reader_v2_invalid() {
        // Invalid: Multiple ReaderWriter features with reader v2
        // Only ColumnMapping alone is allowed with reader v2
        let protocol = Protocol::try_new(
            2,
            7,
            TableFeature::NO_LIST,
            Some(vec![
                TableFeature::ColumnMapping,
                TableFeature::DeletionVectors,
            ]),
        );
        assert!(protocol.is_err());
    }

    #[test]
    fn test_parse_table_feature_never_fails() {
        // weird strs
        let features = Some(["", "absurD_)(+13%^⚙️"]);
        let expected = Some(FromIterator::from_iter([
            TableFeature::unknown(""),
            TableFeature::unknown("absurD_)(+13%^⚙️"),
        ]));
        assert_eq!(parse_features(features), expected);
    }

    #[test]
    fn test_metadata_try_new() {
        let schema = schema_ref! { not_null "id": INTEGER };
        let config = HashMap::from([("key1".to_string(), "value1".to_string())]);

        let metadata = Metadata::try_new(
            Some("test_table".to_string()),
            Some("description".to_string()),
            schema.clone(),
            vec!["year".to_string()],
            1234567890,
            config.clone(),
        )
        .unwrap();

        assert!(!metadata.id.is_empty());
        assert_eq!(metadata.name, Some("test_table".to_string()));
        assert_eq!(
            metadata.schema_string,
            serde_json::to_string(&schema).unwrap()
        );
        assert_eq!(metadata.created_time, Some(1234567890));
        assert_eq!(metadata.configuration, config);
    }

    #[test]
    fn test_metadata_try_new_default() {
        let schema = schema_ref! { not_null "id": INTEGER };
        let metadata = Metadata::try_new(None, None, schema, vec![], 0, HashMap::new()).unwrap();

        assert!(!metadata.id.is_empty());
        assert_eq!(metadata.name, None);
        assert_eq!(metadata.description, None);
    }

    #[test]
    fn test_metadata_unique_ids() {
        let schema = schema_ref! { not_null "id": INTEGER };
        let m1 = Metadata::try_new(None, None, schema.clone(), vec![], 0, HashMap::new()).unwrap();
        let m2 = Metadata::try_new(None, None, schema, vec![], 0, HashMap::new()).unwrap();
        assert_ne!(m1.id, m2.id);
    }

    #[rstest]
    #[case::typical(HashMap::from([
        ("path".to_string(), "/delta/table".to_string()),
        ("compressionType".to_string(), "snappy".to_string()),
    ]))]
    #[case::empty(HashMap::new())]
    #[case::special_characters(HashMap::from([
        ("path".to_string(), "/path/with spaces".to_string()),
        ("unicode".to_string(), "测试🎉".to_string()),
        ("empty".to_string(), String::new()),
    ]))]
    fn test_format_scalar_round_trip(#[case] options: HashMap<String, String>) {
        let format = Format {
            provider: "parquet".to_string(),
            options: options.clone(),
        };
        let scalar = Scalar::from(format.clone());

        let Scalar::Struct(struct_data) = &scalar else {
            panic!("Expected struct scalar, got {scalar}");
        };
        let field_names: Vec<_> = struct_data.fields().iter().map(|f| f.name()).collect();
        assert_eq!(field_names, ["provider", "options"]);
        assert_eq!(struct_data.values()[0], Scalar::from("parquet"));

        let Scalar::Map(map_data) = &struct_data.values()[1] else {
            panic!("Expected map options");
        };
        assert_eq!(map_data.pairs().len(), options.len());

        assert_eq!(Format::try_from(scalar).unwrap(), format);
    }

    #[test]
    fn test_format_default() {
        let format = Format::default();
        let expected = Format {
            provider: "parquet".to_string(),
            options: HashMap::new(),
        };
        assert_eq!(format, expected);
    }

    #[test]
    fn test_metadata_with_log_schema() {
        let engine = ExprEngine::new();
        let schema = schema_ref! { not_null "id": INTEGER };

        let metadata = Metadata::try_new(
            Some("table".to_string()),
            None, // test that omitting description will omit entire field
            schema,
            vec![],
            456,
            HashMap::new(),
        )
        .unwrap();

        let metadata_id = metadata.id.clone();

        // test with the full log schema that wraps metadata in a "metaData" field
        let commit_schema = LOG_METADATA_SCHEMA.clone();
        let actual = create_row(&engine, commit_schema, metadata)
            .unwrap()
            .try_into_record_batch()
            .unwrap();

        let expected_json = json!({
            "metaData": {
                "id": metadata_id,
                "name": "table",
                "format": {
                    "provider": "parquet",
                    "options": {}
                },
                "schemaString": "{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":false,\"metadata\":{}}]}",
                "partitionColumns": [],
                "createdTime": 456,
                "configuration": {}
            }
        }).to_string();
        let expected = ReaderBuilder::new(actual.schema())
            .build(expected_json.as_bytes())
            .unwrap()
            .next()
            .unwrap()
            .unwrap();

        assert_eq!(actual, expected);
    }

    #[test]
    fn test_protocol_creates_log_row() {
        let engine = ExprEngine::new();
        let protocol = Protocol::try_new_modern(
            [TableFeature::DeletionVectors, TableFeature::ColumnMapping],
            [TableFeature::DeletionVectors, TableFeature::ColumnMapping],
        )
        .unwrap();

        let list_field = Arc::new(Field::new("element", ArrowDataType::Utf8, false));
        let protocol_fields = vec![
            Field::new("minReaderVersion", ArrowDataType::Int32, false),
            Field::new("minWriterVersion", ArrowDataType::Int32, false),
            Field::new(
                "readerFeatures",
                ArrowDataType::List(list_field.clone()),
                true, // nullable
            ),
            Field::new(
                "writerFeatures",
                ArrowDataType::List(list_field.clone()),
                true, // nullable
            ),
        ];

        let string_builder = StringBuilder::new();
        let mut list_builder = ListBuilder::new(string_builder).with_field(list_field.clone());
        list_builder.values().append_value("deletionVectors");
        list_builder.values().append_value("columnMapping");
        list_builder.append(true);
        let reader_features_array = list_builder.finish();

        let string_builder = StringBuilder::new();
        let mut list_builder = ListBuilder::new(string_builder).with_field(list_field.clone());
        list_builder.values().append_value("deletionVectors");
        list_builder.values().append_value("columnMapping");
        list_builder.append(true);
        let writer_features_array = list_builder.finish();

        let commit_schema = LOG_PROTOCOL_SCHEMA.clone();
        let engine_data = create_row(&engine, commit_schema, protocol);

        let schema = Arc::new(Schema::new(vec![Field::new(
            "protocol",
            ArrowDataType::Struct(protocol_fields.into()),
            true,
        )]));

        let expected = RecordBatch::try_new(
            schema,
            vec![Arc::new(StructArray::from(vec![
                (
                    Arc::new(Field::new("minReaderVersion", ArrowDataType::Int32, false)),
                    Arc::new(Int32Array::from(vec![3])) as Arc<dyn Array>,
                ),
                (
                    Arc::new(Field::new("minWriterVersion", ArrowDataType::Int32, false)),
                    Arc::new(Int32Array::from(vec![7])) as Arc<dyn Array>,
                ),
                (
                    Arc::new(Field::new(
                        "readerFeatures",
                        ArrowDataType::List(list_field.clone()),
                        true,
                    )),
                    Arc::new(reader_features_array) as Arc<dyn Array>,
                ),
                (
                    Arc::new(Field::new(
                        "writerFeatures",
                        ArrowDataType::List(list_field),
                        true,
                    )),
                    Arc::new(writer_features_array) as Arc<dyn Array>,
                ),
            ]))],
        )
        .unwrap();

        let record_batch = engine_data.try_into_record_batch().unwrap();

        assert_eq!(record_batch, expected);
    }

    #[test]
    fn test_schema_contains_file_actions_with_add() {
        let schema = get_commit_schema()
            .project(&[ADD_NAME, PROTOCOL_NAME])
            .unwrap();
        assert!(schema_contains_file_actions(&schema));
        assert!(schema_contains_file_actions(
            &schema.project(&[ADD_NAME]).unwrap()
        ));
    }

    #[test]
    fn test_schema_contains_file_actions_with_remove() {
        let schema = get_commit_schema()
            .project(&[REMOVE_NAME, METADATA_NAME])
            .unwrap();
        assert!(schema_contains_file_actions(&schema));
        assert!(schema_contains_file_actions(
            &schema.project(&[REMOVE_NAME]).unwrap()
        ));
    }

    #[test]
    fn test_schema_contains_file_actions_with_both() {
        let schema = get_commit_schema()
            .project(&[ADD_NAME, REMOVE_NAME])
            .unwrap();
        assert!(schema_contains_file_actions(&schema));
    }

    #[test]
    fn test_schema_contains_file_actions_with_neither() {
        let schema = get_commit_schema()
            .project(&[PROTOCOL_NAME, METADATA_NAME])
            .unwrap();
        assert!(!schema_contains_file_actions(&schema));
    }

    #[test]
    fn test_schema_contains_file_actions_empty_schema() {
        let schema = schema_ref! {};
        assert!(!schema_contains_file_actions(&schema));
    }

    #[test]
    fn test_add_tags_deserialization_null_case() {
        let json1 = r#"{"path":"file1.parquet","partitionValues":{},"size":100,"modificationTime":1234567890,"dataChange":true,"tags":null}"#;
        let add1: Add = serde_json::from_str(json1).unwrap();
        assert_eq!(add1.tags, None);
    }

    #[test]
    fn test_add_tags_deserialization_nullable_values_case() {
        let json2 = r#"{"path":"file2.parquet","partitionValues":{},"size":200,"modificationTime":1234567890,"dataChange":true,"tags":{"INSERTION_TIME":"1677811178336000","NULLABLE_TAG":null}}"#;
        let add2: Add = serde_json::from_str(json2).unwrap();
        assert!(add2.tags.is_some());
        let tags = add2.tags.unwrap();
        assert_eq!(tags.len(), 2);
        assert_eq!(
            tags.get("INSERTION_TIME"),
            Some(&Some("1677811178336000".to_string()))
        );
        assert_eq!(tags.get("NULLABLE_TAG"), Some(&None));
    }

    #[test]
    fn test_add_tags_deserialization_non_null_values_case() {
        let json3 = r#"{"path":"file3.parquet","partitionValues":{},"size":300,"modificationTime":1234567890,"dataChange":true,"tags":{"INSERTION_TIME":"1677811178336000","MIN_INSERTION_TIME":"1677811178336000"}}"#;
        let add3: Add = serde_json::from_str(json3).unwrap();
        assert!(add3.tags.is_some());
        let tags = add3.tags.unwrap();
        assert_eq!(tags.len(), 2);
        assert_eq!(
            tags.get("INSERTION_TIME"),
            Some(&Some("1677811178336000".to_string()))
        );
        assert_eq!(
            tags.get("MIN_INSERTION_TIME"),
            Some(&Some("1677811178336000".to_string()))
        );
    }

    #[test]
    fn test_add_deserializes_complete_wire_shape() {
        let json = r#"{
            "path":"file.parquet",
            "partitionValues":{"present":"value","null_partition":null},
            "size":300,
            "modificationTime":1234567890,
            "dataChange":false,
            "stats":"{\"numRecords\":1}",
            "tags":{"tag":"value","nullable":null},
            "deletionVector":{
                "storageType":"i",
                "pathOrInlineDv":"",
                "sizeInBytes":0,
                "cardinality":0
            },
            "baseRowId":10,
            "defaultRowCommitVersion":20,
            "clusteringProvider":"liquid",
            "backReference":{"manifest":"manifest.parquet","pos":3}
        }"#;

        let add: Add = serde_json::from_str(json).unwrap();
        let data: StructData = add.clone().into();
        StructData::try_new(data.fields().to_vec(), data.values().to_vec())
            .expect("complete add must match its derived schema");
        assert_eq!(
            add.partition_values,
            HashMap::from([("present".to_string(), "value".to_string())])
        );
        assert_eq!(add.stats.as_deref(), Some(r#"{"numRecords":1}"#));
        assert_eq!(
            add.tags,
            Some(HashMap::from([
                ("tag".to_string(), Some("value".to_string())),
                ("nullable".to_string(), None),
            ]))
        );
        assert_eq!(
            add.deletion_vector.as_ref().unwrap().storage_type,
            deletion_vector::DeletionVectorStorageType::Inline
        );
        assert_eq!(add.base_row_id, Some(10));
        assert_eq!(add.default_row_commit_version, Some(20));
        assert_eq!(add.clustering_provider.as_deref(), Some("liquid"));
        #[cfg(feature = "adaptive-metadata-in-dev")]
        assert_eq!(
            add.back_reference(),
            Some(&BackReference {
                manifest: "manifest.parquet".to_string(),
                pos: 3,
            })
        );
    }

    #[test]
    fn test_add_partition_values_duplicate_key_uses_last_value() {
        let json = r#"{
            "path":"file.parquet",
            "partitionValues":{"part":"value","part":null},
            "size":1,
            "modificationTime":0,
            "dataChange":false
        }"#;

        let add: Add = serde_json::from_str(json).unwrap();
        assert!(add.partition_values.is_empty());
        #[cfg(feature = "adaptive-metadata-in-dev")]
        assert!(add.back_reference().is_none());
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_schema() {
        let schema = get_commit_schema()
            .project(&[CHECKPOINT_ACTION_NAME])
            .unwrap();

        // The `checkpoint` action serializes as an array whose elements are a union of the
        // embedded metadata actions.
        let checkpoint_field = schema.field(CHECKPOINT_ACTION_NAME).unwrap();
        assert!(checkpoint_field.is_nullable());
        let array = match checkpoint_field.data_type() {
            DataType::Array(array) => array,
            other => panic!("Expected array, got {other:?}"),
        };
        assert!(!array.contains_null());
        let element = match array.element_type() {
            DataType::Struct(s) => s,
            other => panic!("Expected struct element, got {other:?}"),
        };
        let field_names: Vec<&str> = element.fields().map(|f| f.name.as_str()).collect();
        assert_eq!(
            field_names,
            vec![
                CHECKPOINT_METADATA_NAME,
                CONTENT_ROOT_NAME,
                PROTOCOL_NAME,
                METADATA_NAME,
                DOMAIN_METADATA_NAME,
                SET_TRANSACTION_NAME,
                SIDECAR_NAME,
            ]
        );
        // `commitInfo` must NOT be a checkpoint-array element; adaptiveMetadata routes it to the
        // top-level Delta log.
        assert!(!field_names.contains(&COMMIT_INFO_NAME));

        // Every element type is an optional (union member) struct.
        for field in element.fields() {
            assert!(field.is_nullable(), "{} should be nullable", field.name);
            assert!(matches!(field.data_type(), DataType::Struct(_)));
        }
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_content_sidecar_is_type_prefixed_sidecar() {
        // The content-sidecar schema is composed from `Sidecar::to_schema()` with a `type`
        // discriminator prepended, so it must equal exactly `type` followed by `Sidecar`'s fields.
        let DataType::Struct(content_sidecar) = CONTENT_SIDECAR_FIELD.data_type() else {
            panic!("content sidecar should be a struct");
        };
        let expected: Vec<StructField> =
            std::iter::once(StructField::not_null("type", DataType::STRING))
                .chain(Sidecar::to_schema().into_fields())
                .collect();
        let actual: Vec<StructField> = content_sidecar.fields().cloned().collect();
        assert_eq!(actual, expected);
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest]
    #[case::relative_path(
        "memory:///table/",
        "metadata/root.parquet",
        2048,
        "memory:///table/metadata/root.parquet",
        Ok(2048)
    )]
    #[case::negative_size(
        "memory:///table/",
        "metadata/root.parquet",
        -1,
        "memory:///table/metadata/root.parquet",
        Err("Failed to convert checkpoint contentRoot size -1")
    )]
    fn test_checkpoint_action_root_filemeta(
        #[case] table_root: &str,
        #[case] path: &str,
        #[case] size_in_bytes: i64,
        #[case] expected_location: &str,
        #[case] expected: KernelResult<FileSize, &str>,
    ) {
        let table_root = Url::parse(table_root).unwrap();
        let checkpoint_action = CheckpointAction {
            version: 1,
            content_root: ContentRoot {
                path: path.to_string(),
                size_in_bytes,
                version: 1,
            },
            protocol: Protocol::new_unchecked(1, 2, None, None),
            metadata: Metadata::default(),
            transactions: Vec::new(),
            domain_metadata: Vec::new(),
            txn_sidecars: Vec::new(),
            domain_metadata_sidecars: Vec::new(),
        };

        let result = checkpoint_action.root_filemeta(&table_root);
        match expected {
            Ok(expected_size) => {
                let file_meta = result.unwrap();
                assert_eq!(file_meta.location.as_str(), expected_location);
                assert_eq!(file_meta.size, expected_size);
                assert_eq!(file_meta.last_modified, i64::MAX);
            }
            Err(expected_message) => assert_result_error_with_message(result, expected_message),
        }
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    fn sample_checkpoint_action() -> CheckpointAction {
        let sidecar = |path: &str| Sidecar {
            path: path.to_string(),
            size_in_bytes: 100,
            modification_time: 1,
            tags: Some(HashMap::from([("tag".to_string(), "value".to_string())])),
        };
        CheckpointAction {
            version: 42,
            content_root: ContentRoot {
                path: "s3://bucket/manifest".to_string(),
                size_in_bytes: 1024,
                version: 40,
            },
            protocol: Protocol::new_unchecked(1, 2, None, None),
            metadata: Metadata::default(),
            transactions: vec![SetTransaction {
                app_id: "myApp".to_string(),
                version: 3,
                last_updated: None,
            }],
            domain_metadata: vec![DomainMetadata {
                domain: "myDomain".to_string(),
                configuration: "cfg".to_string(),
                removed: false,
            }],
            txn_sidecars: vec![sidecar("txn-sidecar.parquet")],
            domain_metadata_sidecars: vec![sidecar("dm-sidecar.parquet")],
        }
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_scalar_round_trip() -> KernelResult<()> {
        let engine = ExprEngine::new();
        let action = sample_checkpoint_action();
        let scalar = action.clone().try_into_scalar()?;
        let data = create_row(&engine, LOG_CHECKPOINT_SCHEMA.clone(), scalar)?;
        let back = CheckpointAction::try_new_from_data(data.as_ref())?
            .expect("checkpoint action should round-trip");
        assert_eq!(action, back);
        Ok(())
    }

    // The `contentRoot.version <= checkpointMetadata.version` invariant is enforced on the
    // serialize path too, not just when parsing. `content_root_version_too_high` in visitors.rs
    // covers the parse-path guard; this covers the `validate()` call during scalar conversion.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_scalar_rejects_invalid_content_root_version() {
        let base = sample_checkpoint_action();
        let action = CheckpointAction {
            content_root: ContentRoot {
                version: base.version + 1,
                ..base.content_root
            },
            ..sample_checkpoint_action()
        };
        let result = action.try_into_scalar();
        assert_result_error_with_message(result, "exceeds checkpointMetadata.version");
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_wire_format() -> KernelResult<()> {
        // Build the action's engine data, then write it out through the engine JSON writer and
        // pin the exact bytes. This is the only guard on the wire format: element order, camelCase
        // field names, the sidecar `type` discriminator, and the JSON writer's null omission (the
        // null union siblings collapse each element to a single-key tagged object).
        let engine = ExprEngine::new();
        let scalar = sample_checkpoint_action().try_into_scalar()?;
        let data = create_row(&engine, LOG_CHECKPOINT_SCHEMA.clone(), scalar)?;
        let filtered = FilteredEngineData::with_all_rows_selected(data);
        let bytes = to_json_bytes(std::iter::once(Ok(filtered)))?;
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();

        assert_eq!(
            json,
            json!({ "checkpoint": [
                { "checkpointMetadata": { "version": 42 } },
                { "contentRoot": { "path": "s3://bucket/manifest", "sizeInBytes": 1024, "version": 40 } },
                { "protocol": { "minReaderVersion": 1, "minWriterVersion": 2 } },
                { "metaData": {
                    "id": "",
                    "format": { "provider": "parquet", "options": {} },
                    "schemaString": "",
                    "partitionColumns": [],
                    "configuration": {},
                } },
                { "txn": { "appId": "myApp", "version": 3 } },
                { "domainMetadata": { "domain": "myDomain", "configuration": "cfg", "removed": false } },
                { "sidecar": { "type": "txn", "path": "txn-sidecar.parquet", "sizeInBytes": 100, "modificationTime": 1, "tags": { "tag": "value" } } },
                { "sidecar": { "type": "domainMetadata", "path": "dm-sidecar.parquet", "sizeInBytes": 100, "modificationTime": 1, "tags": { "tag": "value" } } },
            ] })
        );
        Ok(())
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_try_new_from_data_returns_none_when_no_checkpoint_action() -> KernelResult<()> {
        // `action_batch` carries many action kinds but no `checkpoint` array, so parsing yields
        // `Ok(None)` rather than an error.
        let data = crate::unit_test_utils::action_batch();
        assert!(CheckpointAction::try_new_from_data(data.as_ref())?.is_none());
        Ok(())
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_round_trip_multiple_and_empty_collections() -> KernelResult<()> {
        // Exercise the write loops and the reader's accumulation for count > 1 (two txns, two
        // domainMetadata, two same-type sidecars) and count 0 (empty domainMetadata sidecars).
        let sidecar = |path: &str| Sidecar {
            path: path.to_string(),
            size_in_bytes: 1,
            modification_time: 2,
            tags: None,
        };
        let action = CheckpointAction {
            version: 10,
            content_root: ContentRoot {
                path: "s3://bucket/manifest".to_string(),
                size_in_bytes: 8,
                version: 8,
            },
            protocol: Protocol::new_unchecked(1, 2, None, None),
            metadata: Metadata::default(),
            transactions: vec![
                SetTransaction {
                    app_id: "a1".to_string(),
                    version: 1,
                    last_updated: None,
                },
                SetTransaction {
                    app_id: "a2".to_string(),
                    version: 2,
                    last_updated: None,
                },
            ],
            domain_metadata: vec![
                DomainMetadata {
                    domain: "d1".to_string(),
                    configuration: "c1".to_string(),
                    removed: false,
                },
                DomainMetadata {
                    domain: "d2".to_string(),
                    configuration: "c2".to_string(),
                    removed: true,
                },
            ],
            txn_sidecars: vec![sidecar("t1.parquet"), sidecar("t2.parquet")],
            domain_metadata_sidecars: vec![],
        };
        let engine = ExprEngine::new();
        let scalar = action.clone().try_into_scalar()?;
        let data = create_row(&engine, LOG_CHECKPOINT_SCHEMA.clone(), scalar)?;
        let back = CheckpointAction::try_new_from_data(data.as_ref())?
            .expect("checkpoint action should round-trip");
        assert_eq!(action, back);
        Ok(())
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_round_trip_protocol_with_features() -> KernelResult<()> {
        // A (3, 7) protocol with the same ReaderWriter feature in both lists (required by the
        // read-time feature-consistency check) must survive scalar conversion -> parse.
        let action = CheckpointAction {
            protocol: Protocol::new_unchecked(
                3,
                7,
                Some(vec![TableFeature::AdaptiveMetadataPreview]),
                Some(vec![TableFeature::AdaptiveMetadataPreview]),
            ),
            ..sample_checkpoint_action()
        };
        let engine = ExprEngine::new();
        let scalar = action.clone().try_into_scalar()?;
        let data = create_row(&engine, LOG_CHECKPOINT_SCHEMA.clone(), scalar)?;
        let back = CheckpointAction::try_new_from_data(data.as_ref())?
            .expect("checkpoint action should round-trip");
        assert_eq!(action, back);
        Ok(())
    }

    /// The hand-written serde folds/expands the array losslessly: an action expands to the
    /// tagged-element array and folds back to the identical action.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_serde_round_trip() {
        let action = sample_checkpoint_action();
        let json = serde_json::to_value(&action).unwrap();
        assert!(json.is_array(), "checkpoint action serializes to an array");
        let back: CheckpointAction = serde_json::from_value(json).unwrap();
        assert_eq!(back, action);
    }

    /// The `Serialize` path validates before emitting, mirroring `try_into_scalar`: an action whose
    /// `contentRoot.version` exceeds the checkpoint version fails to serialize rather than writing
    /// an out-of-spec array.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_serde_rejects_invalid_content_root_version() {
        let base = sample_checkpoint_action();
        let action = CheckpointAction {
            content_root: ContentRoot {
                version: base.version + 1,
                ..base.content_root
            },
            ..sample_checkpoint_action()
        };
        let err = serde_json::to_value(&action).expect_err("invalid action must not serialize");
        assert!(
            err.to_string()
                .contains("exceeds checkpointMetadata.version"),
            "expected content-root-version error, got: {err}"
        );
    }

    /// The synthesized `checkpointMetadata` element omits `tags` entirely (via
    /// `skip_serializing_if`) rather than emitting `"tags": null`, matching the EngineData wire
    /// form. The symmetric round-trip tests cannot observe this because both an absent key and an
    /// explicit null parse back to `None`.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_serde_omits_checkpoint_metadata_tags() {
        let json = serde_json::to_value(sample_checkpoint_action()).unwrap();
        let checkpoint_metadata = json
            .as_array()
            .and_then(|elements| elements.first())
            .and_then(|element| element.get("checkpointMetadata"))
            .expect("first element is checkpointMetadata");
        assert_eq!(checkpoint_metadata, &json!({ "version": 42 }));
        assert!(
            checkpoint_metadata.get("tags").is_none(),
            "checkpointMetadata must omit the tags key, got: {checkpoint_metadata}"
        );
    }

    /// Fully-populated array elements, used to build valid and malformed variants for the serde
    /// fold-error cases below. Mirrors `checkpoint_elements` in `visitors.rs` so the serde path and
    /// the EngineData path are checked against the same inputs.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    mod checkpoint_serde_elements {
        pub(super) const CHECKPOINT_METADATA: &str = r#"{"checkpointMetadata":{"version":42}}"#;
        pub(super) const CONTENT_ROOT: &str =
            r#"{"contentRoot":{"path":"p","sizeInBytes":1,"version":40}}"#;
        pub(super) const PROTOCOL: &str =
            r#"{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}"#;
        pub(super) const METADATA: &str = r#"{"metaData":{"id":"id","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[]}","partitionColumns":[],"configuration":{}}}"#;
    }

    /// The serde fold applies the same enumerated checks as the EngineData `CheckpointVisitor`,
    /// with identical error messages (compare `test_parse_checkpoint_action_errors` in
    /// `visitors.rs`): required singletons, no duplicates, known sidecar `type`, and the
    /// `contentRoot.version` invariant. Unknown elements are the intended exception -- the
    /// serde path fails closed on them while the visitor skips them -- and are not covered
    /// here.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest]
    #[case::duplicate_metadata(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA, checkpoint_serde_elements::CONTENT_ROOT,
        checkpoint_serde_elements::PROTOCOL, checkpoint_serde_elements::METADATA,
        checkpoint_serde_elements::METADATA,
    ], "duplicate `metaData` element in checkpoint action")]
    #[case::duplicate_checkpoint_metadata(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA, checkpoint_serde_elements::CHECKPOINT_METADATA,
        checkpoint_serde_elements::CONTENT_ROOT, checkpoint_serde_elements::PROTOCOL,
        checkpoint_serde_elements::METADATA,
    ], "duplicate `checkpointMetadata` element in checkpoint action")]
    #[case::missing_protocol(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA, checkpoint_serde_elements::CONTENT_ROOT,
        checkpoint_serde_elements::METADATA,
    ], "checkpoint action is missing required `protocol` element")]
    #[case::missing_content_root(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA, checkpoint_serde_elements::PROTOCOL,
        checkpoint_serde_elements::METADATA,
    ], "checkpoint action is missing required `contentRoot` element")]
    #[case::missing_metadata(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA, checkpoint_serde_elements::CONTENT_ROOT,
        checkpoint_serde_elements::PROTOCOL,
    ], "checkpoint action is missing required `metaData` element")]
    #[case::empty_array(&[], "checkpoint action is missing required `checkpointMetadata` element")]
    #[case::bad_sidecar_type(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA, checkpoint_serde_elements::CONTENT_ROOT,
        checkpoint_serde_elements::PROTOCOL, checkpoint_serde_elements::METADATA,
        r#"{"sidecar":{"type":"bogus","path":"s.parquet","sizeInBytes":1,"modificationTime":0}}"#,
    ], "checkpoint sidecar has unsupported type `bogus`")]
    #[case::content_root_version_too_high(&[
        checkpoint_serde_elements::CHECKPOINT_METADATA,
        r#"{"contentRoot":{"path":"p","sizeInBytes":1,"version":99}}"#,
        checkpoint_serde_elements::PROTOCOL, checkpoint_serde_elements::METADATA,
    ], "checkpoint contentRoot.version 99 exceeds checkpointMetadata.version 42")]
    fn test_checkpoint_action_serde_fold_errors(
        #[case] elements: &[&str],
        #[case] expected_msg: &str,
    ) {
        let array = format!("[{}]", elements.join(","));
        let err = serde_json::from_str::<CheckpointAction>(&array)
            .expect_err("checkpoint action should fail to fold");
        assert!(
            err.to_string().contains(expected_msg),
            "expected error containing {expected_msg:?}, got: {err}"
        );
    }

    /// The serde path deliberately fails closed on an unrecognized element key, unlike the
    /// forward-compatible EngineData `CheckpointElementVisitor`, which skips unknown elements. Pins
    /// that intended asymmetry: an otherwise-valid array carrying a future element kind fails the
    /// whole-hint parse (so the reader falls back to log replay).
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_checkpoint_action_serde_fails_closed_on_unknown_element() {
        let array = format!(
            "[{},{},{},{},{}]",
            checkpoint_serde_elements::CHECKPOINT_METADATA,
            checkpoint_serde_elements::CONTENT_ROOT,
            checkpoint_serde_elements::PROTOCOL,
            checkpoint_serde_elements::METADATA,
            r#"{"someNewAction":{"foo":1}}"#,
        );
        let err = serde_json::from_str::<CheckpointAction>(&array)
            .expect_err("unknown element must fail the parse");
        assert!(
            err.to_string().contains("unknown variant"),
            "expected an unknown-variant error, got: {err}"
        );
    }
}