delta_kernel 0.30.0

use std::collections::{HashMap, HashSet};
use std::iter;
use std::ops::Deref;
use std::sync::{Arc, LazyLock};
use std::time::{Duration, Instant};

use delta_kernel_derive::internal_api;
use tracing::instrument;

use crate::actions::{
    as_log_add_schema, CommitInfo, DomainMetadata, Metadata, Protocol, SetTransaction,
    LOG_DOMAIN_METADATA_SCHEMA, LOG_METADATA_SCHEMA, LOG_PROTOCOL_SCHEMA, LOG_REMOVE_SCHEMA,
    LOG_TXN_SCHEMA, MAX_VALUES, MIN_VALUES, NULL_COUNT, NUM_RECORDS, TIGHT_BOUNDS,
};
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::actions::{BackReference, CheckpointAction};
use crate::committer::{
    CommitMetadata, CommitProtocolMetadata, CommitResponse, CommitType, Committer,
};
use crate::crc::{CrcDelta, FileStatsDelta};
use crate::engine_data::FilteredEngineData;
use crate::error::KernelError;
use crate::expressions::UnaryExpressionOp::ToJson;
use crate::expressions::{
    col, column_name, lit, null_lit, ArrayData, ColumnName, ExpressionStructPatch,
    ExpressionStructPatchBuilder,
};
use crate::log_replay::HasSelectionVector;
use crate::log_segment::LogSegment;
use crate::metrics::events::TRANSACTION_COMMIT_SPAN;
use crate::metrics::{CommitFailureReason, MetricId};
use crate::path::{LogRoot, ParsedLogPath};
use crate::row_tracking::{RowTrackingDomainMetadata, RowTrackingVisitor};
use crate::scan::data_skipping::stats_schema::schema_with_all_fields_nullable;
use crate::scan::log_replay::{
    BASE_ROW_ID_NAME, DEFAULT_ROW_COMMIT_VERSION_NAME, FILE_CONSTANT_VALUES_NAME,
    PARTITION_VALUES_NAME, PARTITION_VALUES_PARSED_NAME, SIZE_NAME, STATS_NAME, STATS_PARSED_NAME,
    TAGS_NAME,
};
use crate::scan::scan_row_schema;
use crate::schema::void_utils::validate_schema_for_write;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::schema::ToSchema;
use crate::schema::{
    lazy_schema_ref, schema_ref, ArrayType, ColumnDefault, SchemaRef, SchemaStructPatchBuilder,
    StructField, StructType,
};
use crate::snapshot::{Snapshot, SnapshotRef};
use crate::struct_patch::ProjectionStructPatchBuilder;
use crate::table_configuration::TableConfiguration;
use crate::table_features::TableFeature;
use crate::utils::{require, PhantomType};
use crate::{
    create_row, version_as_i64, DataType, Engine, EngineData, Expression, FileMeta, KernelResult,
    KernelResultIterator, Predicate, RowVisitor, Version,
};

#[cfg(feature = "internal-api")]
pub mod builder;
#[cfg(not(feature = "internal-api"))]
pub(crate) mod builder;

#[cfg(feature = "internal-api")]
pub mod create_table;
#[cfg(not(feature = "internal-api"))]
pub(crate) mod create_table;

#[cfg(feature = "internal-api")]
pub mod data_layout;
#[cfg(not(feature = "internal-api"))]
pub(crate) mod data_layout;

pub use builder::update_table::UpdateTableTransactionBuilder;
use builder::TransactionBuilderState;
pub use builder::{TransactionBuilder, TransactionType};
use commit_info::collect_operation_metadata;
mod bound_write_context;
mod commit_info;
mod domain_metadata;
mod exec_mode;
mod operation;
pub(crate) use operation::CommitOperation;
pub use operation::UpdateTableOperation;
#[cfg(feature = "adaptive-metadata-in-dev")]
mod leaf_writer;
#[cfg(feature = "adaptive-metadata-in-dev")]
mod manifest_commit_state;
#[cfg(feature = "adaptive-metadata-in-dev")]
mod root_manifest_file;
pub(crate) mod schema_evolution;
#[internal_api]
#[cfg_attr(not(feature = "internal-api"), allow(unused_imports))]
pub(crate) use schema_evolution::SchemaOperation;
#[cfg(feature = "internal-api")]
pub mod stats_verifier;
#[cfg(not(feature = "internal-api"))]
mod stats_verifier;
mod update_table;
mod write_state;
mod write_validation;

pub use bound_write_context::BoundWriteContext;
pub use exec_mode::{ExecutionMode, Imperative};
#[cfg(feature = "adaptive-metadata-in-dev")]
#[cfg_attr(not(feature = "internal-api"), allow(unused_imports))]
#[internal_api]
pub(crate) use leaf_writer::{LeafNodeWriter, LeafNodeWriterResult};
#[cfg(feature = "adaptive-metadata-in-dev")]
#[cfg_attr(not(feature = "internal-api"), allow(unused_imports))]
#[internal_api]
pub(crate) use manifest_commit_state::ManifestCommitState;
#[cfg(feature = "adaptive-metadata-in-dev")]
use root_manifest_file::RootManifestFile;
use stats_verifier::StatsColumnVerifier;
use update_table::{intermediate_dv_schema, new_dv_column_schema};
pub use write_state::{BoundWriteContextBuilder, RowTrackingMetadataColumns, WriteState};

/// Type alias for an iterator of [`EngineData`] results.
pub(crate) type EngineDataResultIterator<'a> = KernelResultIterator<'a, Box<dyn EngineData>>;

/// The static instance referenced by [`add_files_schema`] that doesn't contain the dataChange
/// column.
pub(crate) static MANDATORY_ADD_FILE_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    not_null "path": STRING,
    not_null "partitionValues": { STRING => nullable STRING },
    not_null "size": LONG,
    not_null "modificationTime": LONG,
};

/// Returns a reference to the mandatory fields in an add action.
///
/// Note this does not include "dataChange" which is a required field but
/// should be set on the transaction level. Getting the full schema
/// can be done with [`Transaction::add_files_schema`].
pub(crate) fn mandatory_add_file_schema() -> &'static SchemaRef {
    &MANDATORY_ADD_FILE_SCHEMA
}

/// The base schema for add file metadata, referenced by [`Transaction::add_files_schema`].
///
/// The `stats` field represents the minimum structure. The actual stats written by
/// `DefaultEngine::write_parquet` include additional fields computed from the data:
/// - `nullCount`: nested struct mirroring the data schema (all fields LONG)
/// - `minValues`: nested struct with min/max eligible column types
/// - `maxValues`: nested struct with min/max eligible column types
///
/// The nested structures within nullCount/minValues/maxValues depend on the table's data schema
/// and which columns have statistics enabled. Use [`Transaction::stats_schema`] to get the
/// expected stats schema for a specific table.
pub(crate) static BASE_ADD_FILES_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    ..(mandatory_add_file_schema().fields().cloned()),
    nullable STATS_NAME: {
        nullable NUM_RECORDS: LONG,
        // nullCount, minValues, maxValues are dynamic based on data schema. Empty struct
        // placeholders indicate these fields exist but their inner structure depends on the
        // table schema and stats column configuration.
        nullable NULL_COUNT: {},
        nullable MIN_VALUES: {},
        nullable MAX_VALUES: {},
        nullable TIGHT_BOUNDS: BOOLEAN,
    },
};

static DATA_CHANGE_COLUMN: LazyLock<StructField> =
    LazyLock::new(|| StructField::not_null("dataChange", DataType::BOOLEAN));

/// Extend a schema with row tracking columns and return a new SchemaRef.
///
/// Note that this method is only useful to extend an Add action schema.
fn with_row_tracking_cols(schema: &SchemaRef) -> KernelResult<SchemaRef> {
    let patch = SchemaStructPatchBuilder::new()
        .append(StructField::nullable("baseRowId", DataType::LONG))
        .append(StructField::nullable(
            "defaultRowCommitVersion",
            DataType::LONG,
        ));
    Ok(Arc::new(patch.build(schema)?))
}

/// The row-tracking-augmented add-file write-metadata schema: [`BASE_ADD_FILES_SCHEMA`] extended
/// with the row-tracking columns.
pub(crate) fn augmented_write_metadata_schema() -> KernelResult<SchemaRef> {
    with_row_tracking_cols(&BASE_ADD_FILES_SCHEMA)
}

/// Marker type for transactions that update existing tables.
///
/// This is the default state for [`Transaction`] and supports schema changes as well as file
/// additions, removals, deletion-vector updates, and blind append semantics.
#[derive(Debug)]
pub struct UpdateTable;

/// Marker type for create-table transactions.
///
/// Transactions in this state support file additions but not file removals, deletion-vector
/// updates, or domain metadata removals.
#[derive(Debug)]
pub struct CreateTable;

/// A transaction represents an in-progress write to a table. After creating a transaction, changes
/// to the table may be staged via the transaction methods before calling `commit` to commit the
/// changes to the table.
///
/// The type parameter `STATE` controls which table operations are available:
/// - [`UpdateTable`] (default): Full API for modifying existing tables.
/// - [`CreateTable`]: Restricted API for table creation (see
///   [`CreateTableTransaction`](create_table::CreateTableTransaction)).
///
/// The independent type parameter `MODE` selects the [`ExecutionMode`], which provides an
/// abstraction over 1) how staged changes are represented and 2) how to inspect the changes. The
/// default, [`Imperative`], stages [`EngineData`] and provides [`commit`](Self::commit).
///
/// Both states support file additions. Schema-changing transactions use [`UpdateTable`] rather
/// than a separate schema-only state, allowing schema and file changes in the same transaction.
/// The builder validates schema intent, and commit validates the staged actions. `ALTER TABLE`
/// commits reject staged data-file batches and manifest writes.
///
/// Configure schema and data-change intent on the builder. Built transactions do not expose
/// setters for either, including when the `internal-api` feature is enabled.
///
/// # Examples
///
/// ```rust,ignore
/// let txn = snapshot
///     .update_table(UpdateTableOperation::Write, "my-app/1.0", /* data_change */ true)
///     .build(&engine, Box::new(FileSystemCommitter::new()))?;
/// txn.commit(&engine)?;
/// ```
///
/// Schema changes cannot be added to a built transaction:
///
/// ```compile_fail
/// # use delta_kernel::transaction::Transaction;
/// fn change_schema(txn: Transaction) {
///     let _ = txn.with_schema_changes(Vec::new());
/// }
/// ```
///
/// The data-change flag cannot be overridden on a built transaction:
///
/// ```compile_fail,E0599
/// # use delta_kernel::transaction::Transaction;
/// fn change_data_change<STATE>(txn: Transaction<STATE>) {
///     let _ = txn.with_data_change(false);
/// }
/// ```
///
/// Blind-append intent cannot be added to a built transaction:
///
/// ```compile_fail
/// # use delta_kernel::transaction::Transaction;
/// fn change_blind_append(txn: Transaction) {
///     let _ = txn.with_blind_append();
/// }
/// ```
///
/// The operation cannot be changed on a built transaction:
///
/// ```compile_fail
/// # use delta_kernel::transaction::{Transaction, UpdateTableOperation};
/// fn change_operation(txn: Transaction) {
///     let _ = txn.with_operation(UpdateTableOperation::Write);
/// }
/// ```
pub struct Transaction<STATE = UpdateTable, MODE: ExecutionMode = Imperative> {
    span: tracing::Span,
    // Correlates all metric events emitted by this transaction.
    operation_id: MetricId,
    // Opaque, caller-supplied id recorded on this transaction's commit metric events alongside
    // `operation_id`. Set via `with_correlation_id`; not interpreted by kernel.
    correlation_id: Option<Arc<str>>,
    // The snapshot this transaction is based on. None for CREATE TABLE (no pre-existing table).
    // Use `read_snapshot()` to access; it returns an error if None.
    read_snapshot_opt: Option<SnapshotRef>,
    // The table configuration that this commit will produce. For writes that don't change the
    // config, this is cloned from the read snapshot; when the config changes (e.g. schema
    // evolution), it is constructed separately with the new schema/protocol.
    effective_table_config: TableConfiguration,
    // Whether to emit a Protocol action. True for CREATE TABLE, false otherwise.
    should_emit_protocol: bool,
    // Whether to emit a Metadata action. True for CREATE TABLE and transactions with staged
    // schema changes, false otherwise.
    should_emit_metadata: bool,
    committer: Box<dyn Committer>,
    operation: CommitOperation,
    operation_parameters: Option<HashMap<String, Option<String>>>,
    operation_metrics: Option<HashMap<String, Option<String>>>,
    engine_info: String,
    engine_commit_info: Option<(Box<dyn EngineData>, SchemaRef)>,
    staged_data_changes: MODE,
    // NB: hashmap would require either duplicating the appid or splitting SetTransaction
    // key/payload. HashSet requires Borrow<&str> with matching Eq, Ord, and Hash. Plus,
    // HashSet::insert drops the to-be-inserted value without returning the existing one, which
    // would make error messaging unnecessarily difficult. Thus, we keep Vec here and deduplicate
    // in the commit method.
    set_transactions: Vec<SetTransaction>,
    // commit-wide timestamp (in milliseconds since epoch) - used in ICT, `txn` action, etc. to
    // keep all timestamps within the same commit consistent.
    commit_timestamp: i64,
    // User-provided domain metadata additions (via with_domain_metadata API).
    user_domain_metadata_additions: Vec<DomainMetadata>,
    // System-generated domain metadata produced by create-table transforms, e.g. clustering.
    system_domain_metadata_additions: Vec<DomainMetadata>,
    // Row-tracking high-water mark explicitly provided by the caller.
    provided_row_tracking_high_water_mark: Option<i64>,
    // Domain names to remove in this transaction. The configuration values are fetched during
    // commit from the log to preserve the pre-image in tombstones.
    user_domain_removals: Vec<String>,
    // Data-change value used by commit validation, file actions, and transaction metrics.
    data_change: bool,
    // TODO(#2499): Replace this state when Conntector responsibilities encode column-default
    // handling. Whether the connector acknowledged responsibility for applying column
    // defaults.
    column_defaults_acknowledged: bool,
    // Whether the connector acknowledged responsibility for preserving Row IDs and Row Commit
    // Versions.
    row_tracking_preservation_acknowledged: bool,
    // Whether this transaction should be marked as a blind append.
    is_blind_append: bool,
    // The manifest this transaction will write, if any. The two ways of writing it are mutually
    // exclusive, so a single field makes staging both unrepresentable.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    manifest_write: Option<ManifestWrite>,
    // Clustering columns from domain metadata. Only populated if the ClusteredTable feature is
    // enabled. Used for determining which columns require statistics collection. Expected to be
    // physical column names.
    physical_clustering_columns: Option<Vec<ColumnName>>,
    // PhantomType marker for transaction state (UpdateTable or CreateTable).
    // Zero-sized; only affects the type system.
    _state: PhantomType<STATE>,
}

/// The manifest a transaction will write. Root-file and content-tree commits are mutually
/// exclusive ways of writing the manifest, so representing them as one enum makes staging both
/// unrepresentable.
#[cfg(feature = "adaptive-metadata-in-dev")]
enum ManifestWrite {
    /// Caller-supplied root manifest file, staged via
    /// [`with_root_manifest_file`](Transaction::with_root_manifest_file).
    RootFile(RootManifestFile),
    /// In-progress manifest (content-tree) commit, staged via
    /// [`with_manifest_commit`](Transaction::with_manifest_commit).
    Commit(ManifestCommitState),
}

impl<STATE, MODE: ExecutionMode> std::fmt::Debug for Transaction<STATE, MODE> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let version_info = match &self.read_snapshot_opt {
            Some(snap) => format!("{}", snap.version()),
            None => "create_table".to_string(),
        };
        f.write_str(&format!(
            "Transaction {{ read_snapshot version: {}, engine_info: {:?} }}",
            version_info, self.engine_info
        ))
    }
}

/// Builds the projection for converting add file metadata into commit-ready Add actions.
fn build_add_action_projection(
    input_schema: &StructType,
    data_change: bool,
) -> KernelResult<(SchemaRef, Expression)> {
    let (output_schema, patch) = ProjectionStructPatchBuilder::new(input_schema)
        .insert_after(
            "modificationTime",
            DATA_CHANGE_COLUMN.clone(),
            lit(data_change),
        )
        .replace(
            "stats",
            StructField::nullable("stats", DataType::STRING),
            Expression::unary(ToJson, col!("stats")),
        )
        .build()?;
    let patch = Expression::struct_from([patch]);
    Ok((output_schema, patch))
}

/// Transforms add file metadata into commit-ready add actions by converting stats to JSON and
/// setting the `dataChange` field.
fn build_add_actions<'a, I, T>(
    engine: &dyn Engine,
    add_files_metadata: I,
    input_schema: SchemaRef,
    data_change: bool,
) -> KernelResult<impl Iterator<Item = KernelResult<Box<dyn EngineData>>> + 'a>
where
    I: Iterator<Item = KernelResult<T>> + Send + 'a,
    T: Deref<Target = dyn EngineData> + Send + 'a,
{
    let evaluation_handler = engine.evaluation_handler();
    let (output_schema, adds_expr) = build_add_action_projection(&input_schema, data_change)?;
    let adds_expr = Arc::new(adds_expr);
    Ok(add_files_metadata.map(move |add_files_batch| {
        let adds_evaluator = evaluation_handler.new_expression_evaluator(
            input_schema.clone(),
            adds_expr.clone(),
            as_log_add_schema(output_schema.clone()).into(),
        )?;
        adds_evaluator.evaluate(add_files_batch?.deref())
    }))
}

// =============================================================================
// Imperative commit APIs
// =============================================================================
impl<STATE> Transaction<STATE, Imperative> {
    /// Consume the transaction and commit it to the table. The result is a result of
    /// [CommitResult] with the following semantics:
    /// - Ok(CommitResult) for either success or a recoverable error (includes the failed
    ///   transaction in case of a conflict so the user can retry, etc.)
    /// - Err(KernelError) indicates a non-retryable error (e.g. logic/validation error).
    ///
    /// `ALTER TABLE` returns a validation error if any data-file batches or manifest writes are
    /// staged. Empty or fully unselected data-file batches are also rejected.
    #[instrument(
        parent = &self.span,
        name = TRANSACTION_COMMIT_SPAN,
        skip_all,
        fields(
            report,
            operation_id = %self.operation_id,
            is_catalog_managed = self.effective_table_config.is_catalog_managed(),
            correlation_id = self.correlation_id.as_deref().unwrap_or(""),
            commit_version = self.get_commit_version(),
            num_add_files,
            num_remove_files,
            num_dv_updates,
            add_files_bytes,
            remove_files_bytes,
            is_blind_append,
            data_change,
            operation,
            prepare_duration_ns,
            committer_duration_ns,
            failure_reason,
        ),
        err
    )]
    pub fn commit(self, engine: &dyn Engine) -> KernelResult<CommitResult<STATE>> {
        let commit_start = Instant::now();

        // === Pre-commit validation ===
        self.validate_operation_compatibility()?;
        self.validate_commit()?;
        self.validate_file_actions()?;

        // === Action generation ===
        let NonfileCommitActions {
            commit_version,
            in_commit_timestamp,
            kernel_commit_info,
            protocol,
            metadata,
            set_transactions,
        } = self.prepare_nonfile_actions(engine)?;
        let CommitBatches {
            actions,
            dm_changes,
        } = self.assemble_commit_batches(
            engine,
            commit_version,
            kernel_commit_info,
            protocol.as_ref(),
            metadata.as_ref(),
            set_transactions,
        )?;

        // === Commit ===
        let commit_metadata = self.create_commit_metadata(
            commit_version,
            in_commit_timestamp,
            protocol,
            metadata,
            dm_changes.clone(),
        )?;
        let prepare_duration = commit_start.elapsed();
        let (commit_response, committer_duration) =
            self.perform_commit(engine, actions, commit_metadata);

        // === Result translation ===
        let post_commit_inputs = PostCommitInputs {
            in_commit_timestamp,
            dm_changes,
        };
        self.translate_commit_result(
            commit_response,
            post_commit_inputs,
            prepare_duration,
            committer_duration,
        )
    }
}

// =============================================================================
// Imperative commit-info configuration
// =============================================================================
impl<STATE> Transaction<STATE, Imperative> {
    /// Set the content of the commitInfo action for this transaction. Note that kernel will
    /// _always_ write a commitInfo, this function simply allows engines to add their own data
    /// into that action if they wish. Kernel overrides the following fields if they are set in
    /// `engine_commit_info`, so connectors should not set them:
    ///
    /// - `timestamp`
    /// - `inCommitTimestamp`
    /// - `operation`
    /// - `operationParameters`
    /// - `operationMetrics`
    /// - `kernelVersion`
    /// - `isBlindAppend`
    /// - `engineInfo`
    /// - `txnId`
    /// - `dataChange` (adaptiveMetadata)
    ///
    /// Kernel merges the following field if it is set:
    ///
    /// - `tags`: When a connector tag has the same key as a Kernel-provided tag, Kernel's value
    ///   takes precedence. Otherwise, the connector-provided tag is preserved.
    pub fn with_commit_info(
        mut self,
        engine_commit_info: Box<dyn EngineData>,
        commit_info_schema: SchemaRef,
    ) -> Self {
        self.engine_commit_info = Some((engine_commit_info, commit_info_schema));
        self
    }
}

// =============================================================================
// Shared methods available on ALL transaction types
// =============================================================================
impl<STATE, MODE: ExecutionMode> Transaction<STATE, MODE> {
    /// Replaces operation metrics recorded in `commitInfo`.
    ///
    /// Present values must already be stringified as expected in table history. `None` writes a
    /// null map value. When unset, `operationMetrics` is omitted. An explicitly empty map is
    /// written as `{}`. A later call replaces the complete map, and the last value wins when a key
    /// occurs more than once.
    pub fn with_operation_metrics<I, K, V>(mut self, metrics: I) -> Self
    where
        I: IntoIterator<Item = (K, Option<V>)>,
        K: Into<String>,
        V: Into<String>,
    {
        self.operation_metrics = Some(collect_operation_metadata(metrics));
        self
    }

    /// Adds a [`SetTransaction`] action for `app_id` and `version`. Returns the updated
    /// transaction.
    ///
    /// The action's `lastUpdated` uses this transaction's commit timestamp. Each application id may
    /// occur only once per transaction, including ids added through the builder. Duplicates are not
    /// checked here; [`Self::commit`] rejects them.
    pub fn with_transaction_id(mut self, app_id: String, version: i64) -> Self {
        let set_transaction = SetTransaction::new(app_id, version, Some(self.commit_timestamp));
        self.set_transactions.push(set_transaction);
        self
    }

    /// Set domain metadata to be written to the Delta log.
    /// Note that each domain can only appear once per transaction. That is, multiple configurations
    /// of the same domain are disallowed in a single transaction, as well as setting and removing
    /// the same domain in a single transaction. If a duplicate domain is included, the commit will
    /// fail (that is, we don't eagerly check domain validity here).
    /// Setting metadata for multiple distinct domains is allowed.
    pub fn with_domain_metadata(mut self, domain: String, configuration: String) -> Self {
        self.user_domain_metadata_additions
            .push(DomainMetadata::new(domain, configuration));
        self
    }

    /// The schema that the [`Engine`]'s [`ParquetHandler`] is expected to use when reporting
    /// information about a Parquet write operation back to Kernel.
    ///
    /// Concretely, it is the expected schema for [`EngineData`] passed to [`add_files`], as it is
    /// the base for constructing an add_file. Each row represents metadata about a
    /// file to be added to the table. Kernel takes this information and extends it to the full
    /// add_file action schema, adding internal fields (e.g., baseRowID) as necessary.
    ///
    /// The `stats` field contains file-level statistics. The schema returned here shows the base
    /// structure; the actual stats written by `DefaultEngine::write_parquet` include dynamically
    /// computed fields (numRecords, nullCount, minValues, maxValues, tightBounds) based on the
    /// data schema and table configuration. See [`stats_schema`] for the table-specific expected
    /// stats schema.
    ///
    /// Note: While currently static, in the future the schema might change depending on
    /// options set on the transaction or features enabled on the table.
    ///
    /// [`add_files`]: crate::transaction::Transaction::add_files
    /// [`ParquetHandler`]: crate::ParquetHandler
    /// [`stats_schema`]: Transaction::stats_schema
    pub fn add_files_schema(&self) -> &'static SchemaRef {
        &BASE_ADD_FILES_SCHEMA
    }
}

// =============================================================================
// Shared data file methods
// =============================================================================
impl<STATE, MODE: ExecutionMode> Transaction<STATE, MODE> {
    // TODO(#2499): Remove this API when Engine responsibilities encode column-default handling.
    /// Acknowledges that the connector applies column defaults before writing data files.
    ///
    /// Call this before requesting write state for a table that enables the
    /// `allowColumnDefaults` feature and declares at least one column default. The connector must
    /// materialize every omitted column's default itself; this method records that responsibility
    /// but does not apply any defaults. Without this acknowledgement, write-state creation fails
    /// with an error.
    pub fn ack_column_defaults(&mut self) {
        self.column_defaults_acknowledged = true;
    }

    /// Returns the expected schema for file statistics.
    ///
    /// The schema structure is derived from table configuration:
    /// - `delta.dataSkippingStatsColumns`: Explicit column list (if set)
    /// - `delta.dataSkippingNumIndexedCols`: Column count limit (default 32)
    /// - Partition columns: Always excluded
    ///
    /// The returned schema has the following structure:
    /// ```ignore
    /// {
    ///   numRecords: long,
    ///   nullCount: { ... },   // Nested struct mirroring data schema, all fields LONG
    ///   minValues: { ... },   // Nested struct, only min/max eligible types
    ///   maxValues: { ... },   // Nested struct, only min/max eligible types
    ///   tightBounds: boolean,
    /// }
    /// ```
    ///
    /// Engines should collect statistics matching this schema structure when writing files.
    ///
    /// Per the Delta protocol, required columns (e.g. clustering columns) are always included
    /// in statistics, regardless of `dataSkippingStatsColumns` or `dataSkippingNumIndexedCols`
    /// settings.
    #[allow(unused)]
    pub fn stats_schema(&self) -> KernelResult<SchemaRef> {
        self.effective_table_config
            .stats_schema_builder()
            .with_required_physical_columns(self.physical_clustering_columns.as_deref())
            .build()
    }

    /// Returns the list of column names that should have statistics collected.
    ///
    /// This returns leaf column paths as [`ColumnName`] objects. Each `ColumnName`
    /// stores path components separately (e.g., `column_name!("nested.field")`).
    /// See [`ColumnName`'s `Display` implementation][ColumnName#impl-Display-for-ColumnName]
    /// for details on string formatting and escaping.
    ///
    /// Engines can use this to determine which columns need stats during writes.
    ///
    /// Per the Delta protocol, clustering columns are always included in statistics,
    /// regardless of `dataSkippingStatsColumns` or `dataSkippingNumIndexedCols` settings.
    #[allow(unused)]
    pub fn stats_columns(&self) -> Vec<ColumnName> {
        self.effective_table_config
            .physical_stats_column_names(self.physical_clustering_columns.as_deref())
    }

    /// Returns the logical partition column names for this table.
    pub fn logical_partition_columns(&self) -> &[String] {
        self.effective_table_config.logical_partition_columns()
    }

    // TODO(#2630): Expose nested column defaults through the transaction API.
    /// Returns the column default for every top-level column in this table's logical schema that
    /// declares one, keyed by logical column name.
    ///
    /// Connectors use this to discover which columns have defaults, then call
    /// [`ColumnDefault::to_scalar`] on each (or fall back to [`ColumnDefault::raw_sql`] when the
    /// kernel cannot parse the default) to materialize the column before writing. After handling
    /// every omitted column, call [`ack_column_defaults`](Self::ack_column_defaults) before
    /// requesting write state.
    ///
    /// Keys are `String` rather than [`ColumnName`] because the kernel currently surfaces defaults
    /// only for top-level columns, consistent with partition columns. This is a kernel limitation,
    /// not a protocol one.
    ///
    /// Malformed defaults (a non-string `CURRENT_DEFAULT`, or a non-`NULL` default on a Variant
    /// column) are rejected eagerly at snapshot load (when constructing the
    /// [`TableConfiguration`]), so by the time this runs the metadata is already validated.
    /// Orphaned metadata (a `CURRENT_DEFAULT` without the `allowColumnDefaults` writer feature)
    /// is tolerated at table load but is not surfaced through this method.
    ///
    /// # Errors
    ///
    /// Propagates any error from [`StructField::column_default`].
    pub fn top_level_column_defaults(&self) -> KernelResult<HashMap<String, ColumnDefault<'_>>> {
        if !self
            .effective_table_config
            .is_feature_enabled(&TableFeature::AllowColumnDefaults)
        {
            tracing::info!(
                "allowColumnDefaults is not enabled; the schema may contain orphaned column-default metadata"
            );
            return Ok(HashMap::new());
        }
        let mut defaults = HashMap::new();
        for field in self.effective_table_config.logical_schema_ref().fields() {
            if let Some(column_default) = field.column_default()? {
                defaults.insert(field.name().clone(), column_default);
            }
        }
        Ok(defaults)
    }

    /// Validates that the table's logical schema supports data writes.
    ///
    /// Called by [`write_state`](Self::write_state), before any Parquet is written, so connectors
    /// fail fast when the schema contains unsupported data types or void placements that cannot
    /// produce valid files.
    ///
    /// We perform this same check during [`commit`](Transaction::commit) as defense-in-depth, for
    /// callers that directly use [`add_files`](Transaction::add_files) without going through
    /// [`write_state`](Self::write_state)
    fn validate_for_data_write(&self) -> KernelResult<()> {
        validate_schema_for_write(&self.effective_table_config.logical_schema())
    }

    /// Creates the table-wide state needed to write data files.
    ///
    /// The returned shared state can create [`BoundWriteContext`] instances in this process,
    /// or it can be encoded and transported to distributed writers. Each context retains a
    /// reference to the same immutable state. All table-wide write validation runs before the
    /// state is returned so a writer does not begin producing files for a table that kernel cannot
    /// write to safely. Use [`WriteState::write_context_builder`] to bind values for each partition
    /// being written.
    ///
    /// The state captures the transaction configuration when this method is called. If the
    /// transaction is subsequently modified, call this method again to capture the updated
    /// configuration.
    ///
    /// Returns an error if the table has an empty or unsupported schema, or if the table declares
    /// column defaults that the connector has not acknowledged.
    pub fn write_state(&self) -> KernelResult<Arc<WriteState>> {
        self.ensure_schema_non_empty_for_write_state()?;
        self.ensure_column_defaults_acknowledged()?;
        self.validate_for_data_write()?;
        // The effective table configuration can change while building a transaction, so this
        // state must be derived on demand rather than cached on the transaction. TODO(#3149):
        // revisit caching if transaction construction becomes immutable.
        Ok(Arc::new(WriteState::new(
            &self.effective_table_config,
            self.stats_columns(),
        )))
    }
}

impl<STATE> Transaction<STATE, Imperative> {
    /// Add files to include in this transaction. This API generally enables the engine to
    /// add/append/insert data (files) to the table. Note that this API can be called multiple times
    /// to add multiple batches.
    ///
    /// The expected schema for `add_metadata` is given by [`Transaction::add_files_schema`].
    ///
    /// TODO(#3553): Support writing clusteringProvider to AddFile actions.
    pub fn add_files(&mut self, add_metadata: Box<dyn EngineData>) {
        self.staged_data_changes
            .add_files_metadata
            .push(add_metadata);
    }
}

// =============================================================================
// Internal methods available on ALL transaction types (used by commit path)
// =============================================================================
/// In-memory representation of non-file commit actions, built from the transaction.
struct NonfileCommitActions {
    commit_version: Version,
    in_commit_timestamp: Option<i64>,
    kernel_commit_info: CommitInfo,
    protocol: Option<Protocol>,
    metadata: Option<Metadata>,
    set_transactions: Vec<SetTransaction>,
}

/// The encoded commit actions, plus the state the post-commit CRC and snapshot are built from.
struct CommitBatches<'a> {
    actions: KernelResultIterator<'a, FilteredEngineData>,
    dm_changes: Vec<DomainMetadata>,
}

/// Commit state produced during action generation that the post-commit CRC and snapshot are built
/// from.
struct PostCommitInputs {
    in_commit_timestamp: Option<i64>,
    dm_changes: Vec<DomainMetadata>,
}

impl<STATE, MODE: ExecutionMode> Transaction<STATE, MODE> {
    /// Initializes shared transaction fields from builder intent and the supplied table
    /// configuration, committer, operation, span, and timestamp. Builders fill table-specific
    /// fields via struct update before returning the transaction.
    fn new(
        common_state: TransactionBuilderState,
        effective_table_config: TableConfiguration,
        committer: Box<dyn Committer>,
        operation: CommitOperation,
        span: tracing::Span,
        commit_timestamp: i64,
    ) -> Self {
        let set_transactions = common_state
            .transaction_ids
            .into_iter()
            .map(|(app_id, version)| SetTransaction::new(app_id, version, Some(commit_timestamp)))
            .collect();

        Self {
            span,
            operation_id: MetricId::new(),
            correlation_id: common_state.correlation_id,
            read_snapshot_opt: None,
            effective_table_config,
            should_emit_protocol: false,
            should_emit_metadata: false,
            committer,
            operation,
            operation_parameters: common_state.operation_parameters,
            operation_metrics: common_state.operation_metrics,
            engine_info: common_state.engine_info,
            engine_commit_info: common_state.engine_commit_info,
            staged_data_changes: MODE::default(),
            set_transactions,
            commit_timestamp,
            user_domain_metadata_additions: common_state.domain_metadata_additions,
            system_domain_metadata_additions: vec![],
            provided_row_tracking_high_water_mark: None,
            user_domain_removals: vec![],
            data_change: common_state.data_change,
            column_defaults_acknowledged: false,
            row_tracking_preservation_acknowledged: false,
            is_blind_append: false,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            manifest_write: None,
            physical_clustering_columns: None,
            _state: PhantomType::default(),
        }
    }

    /// Determines the commit type based on whether this is a create-table operation and whether
    /// the table is catalog-managed.
    fn determine_commit_type(
        is_create: bool,
        table_config: &crate::table_configuration::TableConfiguration,
    ) -> CommitType {
        let is_catalog_managed = table_config.is_catalog_managed();

        // TODO: Handle UpgradeToCatalogManaged and DowngradeToPathBased when ALTER TABLE
        // SET TBLPROPERTIES is supported.
        match (is_create, is_catalog_managed) {
            (true, true) => CommitType::CatalogManagedCreate,
            (true, false) => CommitType::PathBasedCreate,
            (false, true) => CommitType::CatalogManagedWrite,
            (false, false) => CommitType::PathBasedWrite,
        }
    }

    /// Validates that the committer type matches the commit type. A catalog committer must be
    /// used for catalog-managed operations, and a non-catalog committer for path-based operations.
    fn validate_commit_type(
        is_catalog_committer: bool,
        commit_type: &CommitType,
    ) -> KernelResult<()> {
        match (
            is_catalog_committer,
            commit_type.requires_catalog_committer(),
        ) {
            (true, true) | (false, false) => Ok(()),
            (false, true) => Err(KernelError::generic(
                "This table is catalog-managed and requires a catalog committer. \
                 Please provide a catalog committer via UpdateTableTransactionBuilder::build().",
            )),
            (true, false) => Err(KernelError::generic(
                "This table is path-based and cannot be committed to with a catalog committer.",
            )),
        }
    }

    /// Builds the [`CommitMetadata`] for this transaction. Determines the commit type,
    /// validates the committer, and assembles the protocol/metadata state.
    fn create_commit_metadata(
        &self,
        commit_version: Version,
        in_commit_timestamp: Option<i64>,
        new_protocol: Option<Protocol>,
        new_metadata: Option<Metadata>,
        domain_metadata_changes: Vec<crate::actions::DomainMetadata>,
    ) -> KernelResult<CommitMetadata> {
        let log_root = LogRoot::new(self.effective_table_config.table_root().clone())?;
        let is_create = self.is_create_table();
        let commit_type = Self::determine_commit_type(is_create, &self.effective_table_config);
        Self::validate_commit_type(self.committer.is_catalog_committer(), &commit_type)?;
        // For create-table: previous P&M is None (no prior table), new P&M is set.
        // For existing table with metadata change: previous P&M is from snapshot, new P&M
        // is from effective config.
        // For existing table without metadata change: previous P&M is from snapshot, new is None.
        let (read_protocol, read_metadata, max_published_version) = if is_create {
            (None, None, None)
        } else {
            let snap = self.read_snapshot()?;
            let read_config = snap.table_configuration();
            (
                Some(read_config.protocol().clone()),
                Some(read_config.metadata().clone()),
                snap.log_segment().listed.max_published_version,
            )
        };
        let protocol_metadata = CommitProtocolMetadata::try_new(
            read_protocol,
            read_metadata,
            new_protocol,
            new_metadata,
        )?;
        Ok(CommitMetadata::new(
            log_root,
            commit_version,
            commit_type,
            in_commit_timestamp.unwrap_or(self.commit_timestamp),
            max_published_version,
            protocol_metadata,
            domain_metadata_changes,
        ))
    }

    /// Validate that the transaction is eligible to be marked as a blind append.
    ///
    /// Note: Domain metadata additions/removals are allowed; blind append only constrains
    /// data-file operations and read predicates. Conflict resolution determines whether
    /// metadata changes are problematic.
    fn validate_blind_append_semantics(&self) -> KernelResult<()> {
        if !self.is_blind_append {
            return Ok(());
        }
        require!(
            !self.is_create_table(),
            KernelError::invalid_transaction_state(
                "Blind append is not supported for create-table transactions",
            )
        );
        require!(
            self.staged_data_changes.has_adds(),
            KernelError::invalid_transaction_state(
                "Blind append requires at least one added data file"
            )
        );
        require!(
            self.data_change,
            KernelError::invalid_transaction_state("Blind append requires data_change to be true")
        );
        require!(
            !self.staged_data_changes.has_removes(),
            KernelError::invalid_transaction_state("Blind append cannot remove files")
        );
        require!(
            !self.staged_data_changes.has_dv_inputs(),
            KernelError::invalid_transaction_state("Blind append cannot update deletion vectors")
        );

        Ok(())
    }

    /// Validate the staged manifest write, if any. A root-manifest-file commit must carry no data
    /// file actions; a manifest (content-tree) commit cannot be committed yet (its write path is
    /// not built). The `adaptiveMetadata-preview` feature and root/commit mutual exclusion are
    /// enforced when staging, so they need no check here.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    fn validate_manifest_write_semantics(&self) -> KernelResult<()> {
        match &self.manifest_write {
            Some(ManifestWrite::RootFile(_)) => {
                require!(
                    !self.has_data_file_actions(),
                    KernelError::generic("root manifest file commit cannot include file actions")
                );
            }
            Some(ManifestWrite::Commit(_)) => {
                return Err(KernelError::unsupported(
                    "committing a manifest commit is not yet supported",
                ));
            }
            None => {}
        }
        Ok(())
    }

    /// Reject file actions on a commit to an `icebergNativeV4-preview` table.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    fn validate_iceberg_native_v4_semantics(&self) -> KernelResult<()> {
        if !self
            .effective_table_config
            .is_feature_enabled(&TableFeature::IcebergNativeV4Preview)
        {
            return Ok(());
        }
        require!(
            !self.has_data_file_actions(),
            KernelError::invalid_transaction_state(
                "commits to icebergNativeV4-preview tables cannot include file actions"
            )
        );
        Ok(())
    }

    /// Builds the `checkpoint` action committing the configured root manifest file, or `None` if
    /// this transaction has none.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    fn generate_checkpoint_action(
        &self,
        engine: &dyn Engine,
        commit_version: Version,
        dm_changes: &[DomainMetadata],
    ) -> KernelResult<Option<CheckpointAction>> {
        let Some(ManifestWrite::RootFile(root_manifest_file)) = &self.manifest_write else {
            return Ok(None);
        };
        let action = root_manifest_file.compute_checkpoint_action(
            engine,
            commit_version,
            &self.effective_table_config,
            dm_changes,
            &self.set_transactions,
        )?;
        Ok(Some(action))
    }

    fn is_adaptive_metadata_enabled(&self) -> bool {
        self.effective_table_config
            .is_feature_enabled(&TableFeature::AdaptiveMetadataPreview)
    }

    /// Reject data file writes (add/remove/DV) against an empty-schema table.
    /// CREATE TABLE and metadata-only commits are exempt.
    fn ensure_schema_non_empty_for_data_writes(&self) -> KernelResult<()> {
        if self.is_create_table() {
            return Ok(());
        }
        if self.has_data_file_actions() {
            self.ensure_schema_non_empty_for_write_state()?;
        }
        Ok(())
    }

    /// Reject write-state creation on empty-schema tables, so engines fail before staging any
    /// parquet. CREATE TABLE is exempt.
    fn ensure_schema_non_empty_for_write_state(&self) -> KernelResult<()> {
        if self.is_create_table() {
            return Ok(());
        }
        if self.effective_table_config.logical_schema().num_fields() == 0 {
            return Err(KernelError::generic(
                "Cannot write data files to a Delta table with empty schema; \
                 use `snapshot.update_table(UpdateTableOperation::AlterTable, engine_info, false)` and \
                 `add_column(...)` to add at least one column before writing data",
            ));
        }
        Ok(())
    }

    /// Rejects write-state creation when a table declares column defaults and the connector has
    /// not acknowledged handling them.
    fn ensure_column_defaults_acknowledged(&self) -> KernelResult<()> {
        require!(
            self.column_defaults_acknowledged
                || !self
                    .effective_table_config
                    .is_feature_enabled(&TableFeature::AllowColumnDefaults)
                || !self.effective_table_config.has_column_with_default(),
            KernelError::invalid_transaction_state(
                "Writing data to a table with column defaults requires calling \
                 Transaction::ack_column_defaults() first",
            )
        );
        Ok(())
    }

    fn ensure_row_tracking_preservation_acknowledged(&self) -> KernelResult<()> {
        if !self
            .effective_table_config
            .is_feature_enabled(&TableFeature::RowTracking)
        {
            return Ok(());
        }
        require!(
            self.row_tracking_preservation_acknowledged,
            KernelError::invalid_transaction_state(
                "Data manipulation on a table with Row Tracking enabled requires preserving stable \
                 Row IDs for copied or updated rows and stable Row Commit Versions for copied \
                 rows. See Transaction::ack_row_tracking_preservation() for more details",
            )
        );
        Ok(())
    }

    /// Returns true if this is a create-table transaction.
    /// A create-table transaction has no read snapshot (no pre-existing table).
    fn is_create_table(&self) -> bool {
        self.read_snapshot_opt.is_none()
    }

    fn validate_operation_compatibility(&self) -> KernelResult<()> {
        self.operation
            .validate()
            .map_err(KernelError::invalid_transaction_state)?;
        match (self.is_create_table(), &self.operation) {
            (true, CommitOperation::CreateTable) => Ok(()),
            (false, CommitOperation::UpdateTable(UpdateTableOperation::AlterTable))
                if !self.should_emit_metadata =>
            {
                Err(KernelError::invalid_transaction_state(
                    "ALTER TABLE requires at least one schema change",
                ))
            }
            (false, CommitOperation::UpdateTable(UpdateTableOperation::AlterTable))
                if self.data_change =>
            {
                Err(KernelError::invalid_transaction_state(
                    "ALTER TABLE requires data_change to be false",
                ))
            }
            (false, CommitOperation::UpdateTable(UpdateTableOperation::AlterTable))
                if self.has_data_file_actions() =>
            {
                Err(KernelError::invalid_transaction_state(
                    "ALTER TABLE cannot include data-file actions",
                ))
            }
            #[cfg(feature = "adaptive-metadata-in-dev")]
            (false, CommitOperation::UpdateTable(UpdateTableOperation::AlterTable))
                if self.manifest_write.is_some() =>
            {
                Err(KernelError::invalid_transaction_state(
                    "ALTER TABLE cannot include manifest writes",
                ))
            }
            (false, CommitOperation::UpdateTable(_)) => Ok(()),
            (true, _) => Err(KernelError::invalid_transaction_state(
                "create-table transactions must use the CREATE TABLE operation",
            )),
            (false, CommitOperation::CreateTable) => Err(KernelError::invalid_transaction_state(
                "CREATE TABLE cannot use an update-table transaction",
            )),
        }
    }

    /// True iff this transaction stages any data-file batch (add, remove, or DV update).
    fn has_data_file_actions(&self) -> bool {
        self.staged_data_changes.has_adds()
            || self.staged_data_changes.has_removes()
            || self.staged_data_changes.has_dv_inputs()
    }

    // Returns the read snapshot. Returns an error if this is a create-table transaction.
    // To get the `Option<SnapshotRef>` directly, use the `read_snapshot_opt` field.
    fn read_snapshot(&self) -> KernelResult<&Snapshot> {
        self.read_snapshot_opt.as_deref().ok_or_else(|| {
            KernelError::internal_error("read_snapshot() called on create-table transaction")
        })
    }

    /// Computes the in-commit timestamp for this transaction if ICT is enabled.
    /// Returns `None` if ICT is not enabled on the table. A feature being in the protocol
    /// (`is_feature_supported`) is not sufficient -- the `delta.enableInCommitTimestamps`
    /// property must also be `true` (`is_feature_enabled`).
    fn get_in_commit_timestamp(&self, engine: &dyn Engine) -> KernelResult<Option<i64>> {
        let has_ict = self
            .effective_table_config
            .is_feature_enabled(&TableFeature::InCommitTimestamp);

        if !has_ict {
            return Ok(None);
        }

        if self.is_create_table() {
            // For CREATE TABLE there are no prior commits -- use the wall-clock time directly.
            return Ok(Some(self.commit_timestamp));
        }

        // Existing table: enforce monotonicity per the Delta protocol. The timestamp
        // must be the larger of:
        // - The time at which the writer attempted the commit
        // - One millisecond later than the previous commit's inCommitTimestamp
        Ok(self
            .read_snapshot()?
            .get_in_commit_timestamp(engine)?
            .map(|prev_ict| self.commit_timestamp.max(prev_ict + 1)))
    }

    /// Returns the commit version for this transaction.
    /// For existing table transactions, this is snapshot.version() + 1.
    /// For create-table transactions, this is 0.
    fn get_commit_version(&self) -> Version {
        match &self.read_snapshot_opt {
            Some(snap) => snap.version() + 1,
            None => 0,
        }
    }

    #[cfg(test)]
    fn replace_effective_table_config(&mut self, table_config: TableConfiguration) {
        self.effective_table_config = table_config;
    }

    /// Performs all pre-commit validations based on configured transaction settings.
    /// This method does not introspect into staged file-actions (see validate_file_actions
    /// instead).
    fn validate_commit(&self) -> KernelResult<()> {
        // Kernel cannot distinguish Remove actions and DV updates that only delete rows from those
        // that accompany copied or updated rows, so both require the preservation acknowledgment.
        if self.staged_data_changes.has_removes() || self.staged_data_changes.has_dv_updates() {
            self.effective_table_config
                .validate_feature_support_for_remove()?;
            self.ensure_row_tracking_preservation_acknowledged()?;
        }

        // TODO(zach): we currently do this in two passes - can we do it in one and still keep refs
        // in the HashSet?
        let mut app_ids = HashSet::with_capacity(self.set_transactions.len());
        if let Some(dup) = self
            .set_transactions
            .iter()
            .find(|t| !app_ids.insert(&t.app_id))
        {
            return Err(KernelError::generic(format!(
                "app_id {} already exists in transaction",
                dup.app_id
            )));
        }

        self.validate_blind_append_semantics()?;
        self.ensure_schema_non_empty_for_data_writes()?;
        #[cfg(feature = "adaptive-metadata-in-dev")]
        self.validate_manifest_write_semantics()?;
        #[cfg(feature = "adaptive-metadata-in-dev")]
        self.validate_iceberg_native_v4_semantics()?;

        // Validate that the schema supports data writes when files are being added. Reads and
        // metadata-only commits are always allowed.
        if self.staged_data_changes.has_adds() {
            validate_schema_for_write(&self.effective_table_config.logical_schema())?;
        }

        // If a data-changing transaction has add files together with remove files or DV updates,
        // block it when CDF is enabled. Kernel cannot discern DML operations. DML operations that
        // update rows require a `cdc` file, but Kernel does not currently support writing CDC
        // files.
        if !self.is_create_table()
            && self.staged_data_changes.has_adds()
            && (self.staged_data_changes.has_removes() || self.staged_data_changes.has_dv_updates())
            && self.data_change
        {
            let cdf_enabled = self
                .effective_table_config
                .is_feature_enabled(&TableFeature::ChangeDataFeed);
            require!(
                !cdf_enabled,
                KernelError::generic(
                    "Cannot add and remove data in the same transaction when Change Data Feed is enabled (delta.enableChangeDataFeed = true). \
                     This would require writing CDC files for DML operations, which is not yet supported. \
                     Consider using separate transactions: one to add files, another to remove files or update deletion vectors."
                )
            );
        }

        self.validate_domain_metadata_operations()?;
        Ok(())
    }

    /// Generate non-file commit actions as in-memory structs, that can then be encoded into the
    /// committable format (e.g EngineData)
    fn prepare_nonfile_actions(&self, engine: &dyn Engine) -> KernelResult<NonfileCommitActions> {
        let set_transactions = self.set_transactions.clone();
        let in_commit_timestamp = self.get_in_commit_timestamp(engine)?;
        let mut kernel_commit_info = CommitInfo::new(
            self.commit_timestamp,
            in_commit_timestamp,
            Some(self.operation.to_string()),
            Some(self.engine_info.clone()),
            self.is_blind_append,
        );
        if let Some(operation_parameters) = self.operation_parameters.clone() {
            kernel_commit_info.set_operation_parameters(operation_parameters);
        }
        if let Some(operation_metrics) = self.operation_metrics.clone() {
            kernel_commit_info.set_operation_metrics(operation_metrics);
        }

        // Kernel requires every commit on an existing Row Tracking-enabled table to preserve
        // Stable Row IDs and Stable Row Commit Versions, so it always emits true. CREATE TABLE has
        // no existing Row Tracking state to preserve and therefore omits the flag.
        if !self.is_create_table()
            && self
                .effective_table_config
                .is_feature_enabled(&TableFeature::RowTracking)
        {
            kernel_commit_info.set_row_tracking_preserved();
        }

        #[cfg(feature = "adaptive-metadata-in-dev")]
        if self.is_adaptive_metadata_enabled() {
            kernel_commit_info.set_data_change(self.data_change);
        }
        Ok(NonfileCommitActions {
            commit_version: self.get_commit_version(),
            in_commit_timestamp,
            kernel_commit_info,
            protocol: self
                .should_emit_protocol
                .then(|| self.effective_table_config.protocol().clone()),
            metadata: self
                .should_emit_metadata
                .then(|| self.effective_table_config.metadata().clone()),
            set_transactions,
        })
    }

    fn record_commit_success_metrics(
        &self,
        file_stats: &FileStatsDelta,
        num_dv_updates: u64,
        prepare_duration: Duration,
        committer_duration: Duration,
    ) {
        let span = tracing::Span::current();
        span.record("num_add_files", file_stats.gross_add_files);
        span.record("num_remove_files", file_stats.gross_remove_files);
        span.record("num_dv_updates", num_dv_updates);
        span.record("add_files_bytes", file_stats.gross_add_bytes);
        span.record("remove_files_bytes", file_stats.gross_remove_bytes);
        span.record("is_blind_append", self.is_blind_append);
        span.record("data_change", self.data_change);
        span.record("operation", self.operation.as_str());
        span.record("prepare_duration_ns", prepare_duration.as_nanos() as u64);
        span.record(
            "committer_duration_ns",
            committer_duration.as_nanos() as u64,
        );
    }

    fn into_committed(
        self,
        file_meta: FileMeta,
        crc_delta: CrcDelta,
    ) -> KernelResult<CommittedTransaction> {
        let parsed_commit = ParsedLogPath::parse_commit(file_meta)?;

        let commit_version = parsed_commit.version;

        let (post_commit_stats, post_commit_snapshot) = match &self.read_snapshot_opt {
            Some(snap) => {
                // Existing table path: use the read snapshot to compute post-commit state.
                let stats = PostCommitStats {
                    commits_since_checkpoint: snap.log_segment().commits_since_checkpoint() + 1,
                    commits_since_log_compaction: snap
                        .log_segment()
                        .commits_since_log_compaction_or_checkpoint()
                        + 1,
                };
                let snapshot = snap.new_post_commit(parsed_commit, crc_delta)?;
                (stats, Arc::new(snapshot))
            }
            None => {
                // CREATE TABLE path: build a fresh Snapshot at version 0.
                let log_root = self
                    .effective_table_config
                    .table_root()
                    .join("_delta_log/")?;
                let log_segment = LogSegment::new_for_version_zero(log_root, parsed_commit)?;
                let crc = crc_delta.into_complete_crc(0).ok_or_else(|| {
                    KernelError::internal_error(
                        "CREATE TABLE CRC delta is missing protocol or metadata",
                    )
                })?;
                let stats = PostCommitStats {
                    commits_since_checkpoint: 1,
                    commits_since_log_compaction: 1,
                };
                let snapshot = Snapshot::new_with_crc(
                    log_segment,
                    self.effective_table_config,
                    Some(Arc::new(crc)),
                    true,  /* built_as_latest */
                    false, /* skipped_new_checkpoints */
                )?;
                (stats, Arc::new(snapshot))
            }
        };

        Ok(CommittedTransaction {
            commit_version,
            post_commit_stats,
            post_commit_snapshot: Some(post_commit_snapshot),
        })
    }

    /// Build a [`CrcDelta`] from the transaction's commit state and a precomputed
    /// [`FileStatsDelta`].
    fn build_crc_delta(
        &self,
        file_stats: FileStatsDelta,
        in_commit_timestamp: Option<i64>,
        dm_changes: Vec<DomainMetadata>,
    ) -> KernelResult<CrcDelta> {
        // TODO: drop these conversions by migrating the upstream chain
        //       (`CommitMetadata.domain_metadata_changes`, `Transaction.set_transactions`)
        //       to `HashMap<String, _>`, lifting protocol-mandated uniqueness from runtime
        //       checks into the type system.
        let domain_metadata = dm_changes
            .into_iter()
            .map(|dm| (dm.domain().to_string(), dm))
            .collect();
        let set_transactions = self
            .set_transactions
            .iter()
            .map(|txn| (txn.app_id.clone(), txn.clone()))
            .collect();
        // Although `remove.size` is optional per the Delta protocol, the kernel write path
        // enforces presence: `try_compute_for_txn` above errors with `MissingData` if any
        // add or remove row lacks `size` (see `FileStatsVisitor::visit` in
        // `kernel/src/crc/file_stats.rs`). So at this point every size is known to be
        // present, and only operation classification can flip `is_incremental_safe`.
        let is_incremental_safe = self.operation.is_incremental_safe();
        Ok(CrcDelta {
            file_stats,
            protocol: self
                .should_emit_protocol
                .then(|| self.effective_table_config.protocol().clone()),
            metadata: self
                .should_emit_metadata
                .then(|| self.effective_table_config.metadata().clone()),
            domain_metadata,
            set_transactions,
            in_commit_timestamp,
            is_incremental_safe,
        })
    }

    fn into_conflicted(self, conflict_version: Version) -> ConflictedTransaction<STATE, MODE> {
        ConflictedTransaction {
            transaction: self,
            conflict_version,
        }
    }

    fn into_retryable(self, error: KernelError) -> RetryableTransaction<STATE, MODE> {
        RetryableTransaction {
            transaction: self,
            error,
        }
    }
}

// =============================================================================
// Imperative action processing
// =============================================================================
impl<STATE> Transaction<STATE, Imperative> {
    // Reject data-file removals / DV updates on appendOnly tables when `data_change` is true.
    fn validate_append_only_semantics(&self) -> KernelResult<()> {
        if !self.data_change
            || !self
                .effective_table_config
                .is_feature_enabled(&TableFeature::AppendOnly)
        {
            return Ok(());
        }

        let removes_data = self
            .staged_data_changes
            .remove_files_metadata
            .iter()
            .chain(&self.staged_data_changes.dv_matched_files)
            .any(HasSelectionVector::has_selected_rows);
        require!(
            !removes_data,
            KernelError::invalid_transaction_state(
                "Append-only tables cannot remove files or update deletion vectors when data_change is true",
            )
        );
        Ok(())
    }

    /// Validates the rows and required fields in staged addFile, removeFile, and dv updates.
    fn validate_file_actions(&self) -> KernelResult<()> {
        self.validate_append_only_semantics()?;

        // Validate protocol-required add-file statistics.
        // Note: Stats validation cannot use `StagedDataValidator` because its columns and types
        // are determined at runtime, whereas `RowVisitor::selected_column_names_and_types` must
        // return a static projection. It therefore uses dynamic per-column visitors.
        self.validate_add_files_stats(&self.staged_data_changes.add_files_metadata)?;

        // Validate required fields for addFile.
        write_validation::StagedDataValidator::staged_add_file(
            self.effective_table_config.physical_partition_columns(),
        )
        .validate(&self.staged_data_changes.add_files_metadata)?;

        write_validation::StagedDataValidator::staged_dv_matched_file(
            self.effective_table_config.physical_partition_columns(),
        )?
        .validate_filtered(&self.staged_data_changes.dv_matched_files)?;

        // Validate required fields for RemoveFile.
        write_validation::StagedDataValidator::staged_remove_file()
            .validate_filtered(&self.staged_data_changes.remove_files_metadata)?;

        Ok(())
    }

    /// Encodes actions (both non-file and file) into a committable iterator of EngineData,
    /// retaining per-action errors in the commit iterator. Returns the action iterator with the
    /// domain changes used by the post-commit CRC and snapshot logic.
    fn assemble_commit_batches<'a>(
        &'a self,
        engine: &'a dyn Engine,
        commit_version: Version,
        kernel_commit_info: CommitInfo,
        protocol: Option<&Protocol>,
        metadata: Option<&Metadata>,
        set_transactions: Vec<SetTransaction>,
    ) -> KernelResult<CommitBatches<'a>> {
        let set_transaction_actions = set_transactions
            .into_iter()
            .map(|txn| create_row(engine, LOG_TXN_SCHEMA.clone(), txn));
        let commit_info_action = self.generate_commit_info(engine, kernel_commit_info);
        let protocol_action = protocol
            .map(|protocol| create_row(engine, LOG_PROTOCOL_SCHEMA.clone(), protocol.clone()))
            .transpose()?;
        let metadata_action = metadata
            .map(|metadata| create_row(engine, LOG_METADATA_SCHEMA.clone(), metadata.clone()))
            .transpose()?;

        let (add_actions, row_tracking_domain_metadata) =
            self.generate_adds(engine, commit_version)?;

        let dm_changes =
            self.generate_domain_metadata_actions(engine, row_tracking_domain_metadata)?;
        let domain_metadata_actions: Vec<_> = dm_changes
            .iter()
            .cloned()
            .map(|dm| create_row(engine, LOG_DOMAIN_METADATA_SCHEMA.clone(), dm))
            .collect();

        let dv_update_actions = self.generate_dv_update_actions(engine)?;

        let remove_actions = self.generate_remove_actions(
            engine,
            self.staged_data_changes.remove_files_metadata.iter(),
            false, /* has_dv_update_columns */
        )?;

        #[cfg(feature = "adaptive-metadata-in-dev")]
        let checkpoint_action = self
            .generate_checkpoint_action(engine, commit_version, &dm_changes)?
            .map(|action| action.into_engine_data(engine))
            .transpose()?;
        #[cfg(not(feature = "adaptive-metadata-in-dev"))]
        let checkpoint_action: Option<Box<dyn EngineData>> = None;

        // CommitInfo must precede all other actions.
        let actions = iter::once(commit_info_action)
            .chain(protocol_action.map(Ok))
            .chain(metadata_action.map(Ok))
            .chain(checkpoint_action.map(Ok))
            .chain(add_actions)
            .chain(set_transaction_actions)
            .chain(domain_metadata_actions);

        let filtered_actions = actions
            .map(|action_result| action_result.map(FilteredEngineData::with_all_rows_selected))
            .chain(remove_actions)
            .chain(dv_update_actions);

        Ok(CommitBatches {
            actions: Box::new(filtered_actions),
            dm_changes,
        })
    }

    /// Submits the action iterator and returns the committer's response + elapsed time.
    fn perform_commit(
        &self,
        engine: &dyn Engine,
        actions: KernelResultIterator<'_, FilteredEngineData>,
        commit_metadata: CommitMetadata,
    ) -> (KernelResult<CommitResponse>, Duration) {
        let committer_start = Instant::now();
        let commit_response = self.committer.commit(engine, actions, commit_metadata);
        (commit_response, committer_start.elapsed())
    }

    /// Translates a commit response into a CommitResult, while recording metrics and preserving the
    /// transaction on retryable failure. Computes file statistics only for a committed response
    /// and propagates post-commit errors.
    fn translate_commit_result(
        self,
        commit_response: KernelResult<CommitResponse>,
        post_commit_inputs: PostCommitInputs,
        prepare_duration: Duration,
        committer_duration: Duration,
    ) -> KernelResult<CommitResult<STATE>> {
        match commit_response {
            Ok(CommitResponse::Committed { file_meta }) => {
                let PostCommitInputs {
                    in_commit_timestamp,
                    dm_changes,
                } = post_commit_inputs;
                // TODO(#2717): the commit already succeeded atomically; the post-commit `?`
                //              below must not fail the txn (and must not mislabel the metric).
                let bin_boundaries = self
                    .read_snapshot_opt
                    .as_ref()
                    .and_then(|snap| snap.get_file_stats_if_present())
                    .and_then(|s| s.file_size_histogram)
                    .map(|h| h.sorted_bin_boundaries);
                let file_stats = FileStatsDelta::try_compute_for_txn(
                    &self.staged_data_changes.add_files_metadata,
                    &self.staged_data_changes.remove_files_metadata,
                    bin_boundaries.as_deref(),
                )?;
                self.record_commit_success_metrics(
                    &file_stats,
                    self.staged_data_changes.num_dv_updates as u64,
                    prepare_duration,
                    committer_duration,
                );
                let crc_delta =
                    self.build_crc_delta(file_stats, in_commit_timestamp, dm_changes)?;
                Ok(CommitResult::Committed(
                    self.into_committed(file_meta, crc_delta)?,
                ))
            }
            Ok(CommitResponse::Conflict { version }) => {
                // Flips the metric event from success -> failure.
                tracing::Span::current()
                    .record("failure_reason", CommitFailureReason::Conflict.as_ref());
                Ok(CommitResult::Conflicted(self.into_conflicted(version)))
            }
            // TODO: we may want to be more or less selective about what is retryable (this is tied
            // to the idea of "what kind of Errors should write_json_file return?")
            Err(e @ KernelError::IOError(_)) => {
                // Flips the metric event from success -> failure.
                tracing::Span::current()
                    .record("failure_reason", CommitFailureReason::RetryableIo.as_ref());
                Ok(CommitResult::Retryable(self.into_retryable(e)))
            }
            Err(e) => Err(e),
        }
    }

    /// Validate that add files carry the per-file statistics required by the table's protocol.
    ///
    /// Currently checks two protocol requirements:
    /// - `stats.numRecords` must be present when [`requires_stats_num_records`] returns true.
    /// - Per-file min/max/nullCount must be present for clustering columns when the
    ///   `ClusteredTable` feature is enabled.
    ///
    /// Other stat columns (e.g. the conventional "first 32 columns") are not validated here
    /// because they are not protocol-required.
    ///
    /// Only add files are validated(remove files do not carry statistics).
    ///
    /// [`requires_stats_num_records`]: crate::table_configuration::TableConfiguration::requires_stats_num_records
    fn validate_add_files_stats(&self, add_files: &[Box<dyn EngineData>]) -> KernelResult<()> {
        if add_files.is_empty() {
            return Ok(());
        }
        if self.effective_table_config.requires_stats_num_records() {
            // TODO: Likely it's better to merge this with the clustering column validation below,
            // benchmark it and see if it's faster. If so, refactor this to do both validations in
            // one pass.
            stats_verifier::verify_num_records_present(add_files)?;
        }
        if let Some(ref clustering_cols) = self.physical_clustering_columns {
            if !clustering_cols.is_empty() {
                let physical_schema = self.effective_table_config.physical_schema();
                let columns_with_types: Vec<(ColumnName, DataType)> = clustering_cols
                    .iter()
                    .map(|col| {
                        let data_type = physical_schema
                            .fields_of_path(col)?
                            .last()
                            .map(|field| field.data_type().clone())
                            .ok_or_else(|| {
                                KernelError::internal_error(format!(
                                    "Required column '{col}' not found in table schema"
                                ))
                            })?;
                        Ok((col.clone(), data_type))
                    })
                    .collect::<KernelResult<_>>()?;
                let verifier = StatsColumnVerifier::new(columns_with_types);
                verifier.verify(add_files)?;
            }
        }
        Ok(())
    }

    /// Generates add actions and row tracking domain metadata for a commit.
    #[instrument(name = "txn.gen_adds", skip_all, err)]
    fn generate_adds<'a>(
        &'a self,
        engine: &dyn Engine,
        commit_version: u64,
    ) -> KernelResult<(
        EngineDataResultIterator<'a>,
        Option<RowTrackingDomainMetadata>,
    )> {
        // Note: this does not require delta.enableRowTracking=true. "supported" is sufficient
        // for writers to assign row IDs.
        let assign_fresh_row_tracking_metadata = self
            .effective_table_config
            .should_assign_fresh_row_tracking_metadata();

        if self.staged_data_changes.add_files_metadata.is_empty() {
            // No files to add. For an empty CREATE TABLE with row tracking, emit the initial
            // high water mark domain metadata (rowIdHighWaterMark = -1) so subsequent writes
            // have a valid starting point. For all other empty commits (metadata-only, etc.),
            // nothing row-tracking-related needs to be written.
            let row_tracking_dm = (assign_fresh_row_tracking_metadata && self.is_create_table())
                .then(RowTrackingDomainMetadata::initial);
            return Ok((Box::new(iter::empty()), row_tracking_dm));
        }

        let commit_version = version_as_i64(commit_version)?;

        if assign_fresh_row_tracking_metadata {
            self.generate_adds_with_row_tracking(engine, commit_version)
        } else {
            let add_actions = build_add_actions(
                engine,
                self.staged_data_changes
                    .add_files_metadata
                    .iter()
                    .map(|a| Ok(a.deref())),
                self.add_files_schema().clone(),
                self.data_change,
            )?;
            Ok((Box::new(add_actions), None))
        }
    }

    /// Generates add actions with row tracking columns and the row ID high water mark
    /// domain metadata.
    ///
    /// Reads `numRecords` in one pass to compute the final high water mark, then assigns unique
    /// non-overlapping `baseRowId` ranges in a second pass. The initial high water mark is read
    /// from the snapshot for existing tables, or defaults to -1 for create-table (no prior log
    /// to read from).
    fn generate_adds_with_row_tracking<'a>(
        &'a self,
        engine: &dyn Engine,
        commit_version: i64,
    ) -> KernelResult<(
        EngineDataResultIterator<'a>,
        Option<RowTrackingDomainMetadata>,
    )> {
        let row_id_high_water_mark = if self.is_create_table() {
            None
        } else {
            self.read_snapshot()?
                .get_row_tracking_high_water_mark(engine)?
        };

        // Assign base row IDs starting from the current high water mark.
        let mut row_tracking_visitor = RowTrackingVisitor::new(row_id_high_water_mark);
        let extended_add_files =
            self.staged_data_changes
                .add_files_metadata
                .iter()
                .map(move |add_files_batch| {
                    row_tracking_visitor.visit_rows_of(add_files_batch.deref())?;
                    let base_row_ids = row_tracking_visitor.take_base_row_ids();
                    let commit_versions = vec![commit_version; base_row_ids.len()];
                    let base_row_ids_array =
                        ArrayData::try_new(ArrayType::new(DataType::LONG, true), base_row_ids)?;
                    let commit_versions_array =
                        ArrayData::try_new(ArrayType::new(DataType::LONG, true), commit_versions)?;

                    let row_tracking_schema = with_row_tracking_cols(&schema_ref! {})?;
                    add_files_batch.append_columns(
                        row_tracking_schema,
                        vec![base_row_ids_array, commit_versions_array],
                    )
                });

        // Generate add actions including row tracking metadata
        let add_actions = build_add_actions(
            engine,
            extended_add_files,
            augmented_write_metadata_schema()?,
            self.data_change,
        )?;

        // Do a dry run to eagerly compute the new high water mark so we can store it. The row
        // tracking calculation above happens only lazily as our caller drives the iterator.
        let mut dry_run = RowTrackingVisitor::dry_run(row_id_high_water_mark);
        for add_files_batch in &self.staged_data_changes.add_files_metadata {
            dry_run.visit_rows_of(add_files_batch.deref())?;
        }
        let row_tracking_domain_metadata =
            RowTrackingDomainMetadata::new(dry_run.row_id_high_water_mark);

        Ok((Box::new(add_actions), Some(row_tracking_domain_metadata)))
    }

    /// Generates Remove actions from scan file metadata.
    ///
    /// This internal method transforms scan row metadata into Remove actions for the Delta log.
    /// It's called during commit to process files staged via [`remove_files`] or files being
    /// updated with new deletion vectors via [`update_deletion_vectors`].
    ///
    /// # Parameters
    ///
    /// - `engine`: The engine used for expression evaluation
    /// - `remove_files_metadata`: Iterator over scan file metadata to transform into Remove actions
    /// - `has_dv_update_columns`: Whether `remove_files_metadata` contains the temporary columns
    ///   added for a deletion vector update
    ///
    /// # Returns
    ///
    /// An iterator of FilteredEngineData containing Remove actions in the log schema format.
    ///
    /// [`remove_files`]: Transaction::remove_files
    /// [`update_deletion_vectors`]: Transaction::update_deletion_vectors
    #[instrument(name = "txn.gen_removes", skip_all, err)]
    fn generate_remove_actions<'a>(
        &'a self,
        engine: &dyn Engine,
        remove_files_metadata: impl Iterator<Item = &'a FilteredEngineData> + Send + 'a,
        has_dv_update_columns: bool,
    ) -> KernelResult<impl Iterator<Item = KernelResult<FilteredEngineData>> + Send + 'a> {
        // Create-table transactions should not have any remove actions.
        // Only error if there are actually files queued for removal.
        if self.is_create_table() && !self.staged_data_changes.remove_files_metadata.is_empty() {
            return Err(KernelError::internal_error(
                "CREATE TABLE transaction cannot have remove actions",
            ));
        }

        let target_schema = schema_with_all_fields_nullable(&LOG_REMOVE_SCHEMA);
        let evaluation_handler = engine.evaluation_handler();
        // TODO(#3263): `remove_files_metadata` may contain `stats_parsed` and
        // `partitionValues_parsed`; provide its full schema to both evaluators.
        let input_schema = if has_dv_update_columns {
            intermediate_dv_schema().clone()
        } else {
            scan_row_schema()
        };
        let columns_to_drop: Vec<_> = has_dv_update_columns
            .then(new_dv_column_schema)
            .into_iter()
            .flat_map(|schema| schema.fields().map(|field| field.name().to_owned()))
            .collect();

        // adaptiveMetadata removes must carry a null deletionTimestamp and extendedFileMetadata =
        // true (see `Remove` and `build_remove_struct_patch`).
        let adaptive_metadata_enabled = self.is_adaptive_metadata_enabled();

        let make_eval = |coalesce_stats_with_parsed: bool| {
            let columns_to_drop: Vec<_> = columns_to_drop.iter().map(String::as_str).collect();
            let patch = build_remove_struct_patch(
                self.commit_timestamp,
                self.data_change,
                &columns_to_drop,
                coalesce_stats_with_parsed,
                adaptive_metadata_enabled,
            )?;
            let expr = Arc::new(Expression::struct_from([Expression::struct_patch(patch)?]));
            evaluation_handler.new_expression_evaluator(
                input_schema.clone(),
                expr,
                target_schema.clone().into(),
            )
        };

        // Build two evaluators: one for the common case where scan files do not include a
        // stats_parsed column, and one for predicate-based scans that include stats_parsed.
        // The stats_parsed evaluator coalesces stats with ToJson(stats_parsed) to handle the
        // case where stats is null (e.g., on V2 checkpoints with writeStatsAsJson=false) and
        // then drops the stats_parsed column.
        let base_eval = make_eval(false /* coalesce_stats_with_parsed */)?;
        let stats_parsed_eval = make_eval(true /* coalesce_stats_with_parsed */)?;
        let stats_parsed_col = column_name!(STATS_PARSED_NAME);

        Ok(remove_files_metadata.map(move |file_metadata_batch| {
            let data = file_metadata_batch.data();
            let evaluator = if data.has_field(&stats_parsed_col) {
                &stats_parsed_eval
            } else {
                &base_eval
            };
            let updated_engine_data = evaluator.evaluate(data)?;
            FilteredEngineData::try_new(
                updated_engine_data,
                file_metadata_batch.selection_vector().to_vec(),
            )
        }))
    }
}

/// Builds the struct patch for converting scan row metadata into a Remove action.
///
/// Handles two "parsed" columns that predicate-based scans add to scan metadata:
///
/// - `stats_parsed`: when `coalesce_stats_with_parsed` is true, the `stats` field is replaced with
///   `COALESCE(stats, TO_JSON(stats_parsed))` and `stats_parsed` is dropped. The coalesce handles
///   cases where `stats` is null (e.g., V2 checkpoints with `writeStatsAsJson=false`) by
///   reconstructing the JSON from the parsed representation.
/// - `partitionValues_parsed`: dropped if present. Unlike stats, no reconstruction is needed: the
///   Remove action's `partitionValues` is sourced from `fileConstantValues.partitionValues`, which
///   scans always populate from `add.partitionValues`.
///
/// When `adaptive_metadata_enabled` is set, the RFC requires `deletionTimestamp` to be null
/// (cleanup uses tree reachability, not timestamp expiry), so it is emitted as a fixed literal
/// rather than derived from the input.
fn build_remove_struct_patch(
    commit_timestamp: i64,
    data_change: bool,
    columns_to_drop: &[&str],
    coalesce_stats_with_parsed: bool,
    adaptive_metadata_enabled: bool,
) -> KernelResult<ExpressionStructPatch> {
    let deletion_timestamp = if adaptive_metadata_enabled {
        null_lit(DataType::LONG)
    } else {
        lit(commit_timestamp)
    };
    // Note: The Delta protocol requires `partitionValues`, `size`, and `tags` when
    // `extendedFileMetadata` is true. We require only `partitionValues` and `size` to match Spark.
    // Under adaptiveMetadata both are guaranteed present
    let extended_file_metadata = Predicate::and_from([
        col!(SIZE_NAME).is_not_null(),
        col!(FILE_CONSTANT_VALUES_NAME, PARTITION_VALUES_NAME).is_not_null(),
    ]);
    let mut patch = ExpressionStructPatchBuilder::new()
        // deletionTimestamp
        .insert_after("path", deletion_timestamp)
        // dataChange
        .insert_after("path", lit(data_change))
        // extended_file_metadata
        .insert_after("path", Expression::from(extended_file_metadata))
        .insert_after(
            "path",
            col!(FILE_CONSTANT_VALUES_NAME, PARTITION_VALUES_NAME),
        );

    if coalesce_stats_with_parsed {
        // Replace stats with COALESCE(stats, TO_JSON(stats_parsed)) and drop stats_parsed.
        let coalesce_stats = Expression::coalesce([
            col!("stats"),
            Expression::unary(ToJson, col!(STATS_PARSED_NAME)),
        ]);
        patch = patch
            .replace("stats", coalesce_stats)
            .drop(STATS_PARSED_NAME);
    }

    patch = patch
        .insert_after("stats", col!(FILE_CONSTANT_VALUES_NAME, TAGS_NAME))
        .insert_after(
            "deletionVector",
            col!(FILE_CONSTANT_VALUES_NAME, BASE_ROW_ID_NAME),
        )
        .insert_after(
            "deletionVector",
            col!(FILE_CONSTANT_VALUES_NAME, DEFAULT_ROW_COMMIT_VERSION_NAME),
        )
        .drop(FILE_CONSTANT_VALUES_NAME)
        .drop("modificationTime")
        // Added to scan output when the predicate touches a partition column.
        .drop_if_exists(PARTITION_VALUES_PARSED_NAME);

    // Kernel does not populate adaptive-metadata-tree back references on writes, so emit a null to
    // keep the produced struct aligned with the `backReference` field of LOG_REMOVE_SCHEMA.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    {
        patch = patch.append(null_lit(BackReference::to_schema()));
    }

    for column_to_drop in columns_to_drop {
        patch = patch.drop(*column_to_drop);
    }

    patch.build()
}

/// Kernel exposes information about the state of the table that engines might want to use to
/// trigger actions like checkpointing or log compaction. This struct holds that information.
#[derive(Debug)]
pub struct PostCommitStats {
    /// The number of commits since this table has been checkpointed. Note that commit 0 is
    /// considered a checkpoint for the purposes of this computation.
    pub commits_since_checkpoint: u64,
    /// The number of commits since the log has been compacted on this table. Note that a
    /// checkpoint is considered a compaction for the purposes of this computation. Thus this
    /// is really the number of commits since a compaction OR a checkpoint.
    pub commits_since_log_compaction: u64,
}

/// The result of attempting to commit this transaction. If the commit was
/// successful/conflicted/retryable, the result is Ok(CommitResult), otherwise, if a nonrecoverable
/// error occurred, the result is Err(KernelError).
///
/// The commit result can be one of the following:
/// - [`CommitResult::Committed`]: the transaction was successfully committed. [PostCommitStats] and
///   in the future a post-commit snapshot can be obtained from the committed transaction.
/// - [`CommitResult::Conflicted`]: the transaction conflicted with an existing version. This
///   transcation must be rebased before retrying. (currently no rebase APIs exist, caller must
///   create new txn)
/// - [`CommitResult::Retryable`]: an IO (retryable) error occurred during the commit. This
///   transaction can be retried without rebasing.
#[derive(Debug)]
#[must_use]
pub enum CommitResult<STATE = UpdateTable, MODE: ExecutionMode = Imperative> {
    /// The transaction was successfully committed.
    Committed(CommittedTransaction),
    /// This transaction conflicted with an existing version (see
    /// [ConflictedTransaction::conflict_version]). The transaction
    /// is returned so the caller can resolve the conflict (along with the version which
    /// conflicted).
    // TODO(zach): in order to make the returning of a transaction useful, we need to add APIs to
    // update the transaction to a new version etc.
    Conflicted(ConflictedTransaction<STATE, MODE>),
    /// An IO (retryable) error occurred during the commit.
    Retryable(RetryableTransaction<STATE, MODE>),
}

impl<STATE, MODE: ExecutionMode> CommitResult<STATE, MODE> {
    /// Returns true if the commit was successful.
    pub fn is_committed(&self) -> bool {
        matches!(self, CommitResult::Committed(_))
    }
}

impl<STATE: std::fmt::Debug, MODE: ExecutionMode> CommitResult<STATE, MODE> {
    /// Unwraps the [`CommittedTransaction`], panicking if the commit was not successful.
    #[cfg(any(test, feature = "test-utils"))]
    #[allow(clippy::panic)]
    pub fn unwrap_committed(self) -> CommittedTransaction {
        match self {
            CommitResult::Committed(c) => c,
            other => panic!("Expected Committed, got: {other:?}"),
        }
    }

    /// Unwraps the post-commit snapshot of the [`CommittedTransaction`], panicking if the
    /// commit was not successful or the post-commit snapshot is missing.
    /// TODO(#2494): Refactor existing tests to use this.
    #[cfg(any(test, feature = "test-utils"))]
    #[allow(clippy::panic, clippy::expect_used)]
    pub fn unwrap_post_commit_snapshot(self) -> SnapshotRef {
        self.unwrap_committed()
            .post_commit_snapshot()
            .expect("expected post-commit snapshot")
            .clone()
    }
}

/// This is the result of a successfully committed [Transaction]. One can retrieve the
/// [post_commit_stats], [commit version], and optionally the [post-commit snapshot] from this
/// struct.
///
/// [post_commit_stats]: Self::post_commit_stats
/// [commit version]: Self::commit_version
/// [post-commit snapshot]: Self::post_commit_snapshot
#[derive(Debug)]
pub struct CommittedTransaction {
    /// The version of the table that was just committed.
    commit_version: Version,
    /// The [`PostCommitStats`] for this transaction.
    post_commit_stats: PostCommitStats,
    /// The [`SnapshotRef`] of the table after this transaction was committed.
    ///
    /// This is optional to allow incremental development of new features (e.g., table creation,
    /// transaction retries) without blocking on implementing post-commit snapshot support.
    post_commit_snapshot: Option<SnapshotRef>,
}

impl CommittedTransaction {
    /// The version of the table that was just sucessfully committed
    pub fn commit_version(&self) -> Version {
        self.commit_version
    }

    /// The [`PostCommitStats`] for this transaction
    pub fn post_commit_stats(&self) -> &PostCommitStats {
        &self.post_commit_stats
    }

    /// The [`SnapshotRef`] of the table after this transaction was committed.
    pub fn post_commit_snapshot(&self) -> Option<&SnapshotRef> {
        self.post_commit_snapshot.as_ref()
    }
}

/// This is the result of a conflicted [Transaction]. One can retrieve the [conflict version] from
/// this struct. In the future a rebase API will be provided (issue #1389).
///
/// [conflict version]: Self::conflict_version
#[derive(Debug)]
pub struct ConflictedTransaction<STATE = UpdateTable, MODE: ExecutionMode = Imperative> {
    // TODO: remove after rebase APIs
    #[allow(dead_code)]
    transaction: Transaction<STATE, MODE>,
    conflict_version: Version,
}

impl<STATE, MODE: ExecutionMode> ConflictedTransaction<STATE, MODE> {
    /// The version attempted commit that yielded a conflict
    pub fn conflict_version(&self) -> Version {
        self.conflict_version
    }
}

/// A transaction that failed to commit due to a retryable error (e.g. IO error). The transaction
/// can be recovered with `RetryableTransaction::transaction` and retried without rebasing. The
/// associated error can be inspected via `RetryableTransaction::error`.
#[derive(Debug)]
pub struct RetryableTransaction<STATE = UpdateTable, MODE: ExecutionMode = Imperative> {
    /// The transaction that failed to commit due to a retryable error.
    pub transaction: Transaction<STATE, MODE>,
    /// Transient error that caused the commit to fail.
    pub error: KernelError,
}

#[cfg(test)]
mod tests {
    // Manifest-commit and root-manifest tests live in their own file
    // (tests/manifest_commit_tests.rs) but as a submodule of `tests`, so they reuse this
    // module's private helpers (e.g. `create_existing_table_txn`) without widening their
    // visibility.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    mod manifest_commit_tests;

    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::sync::Mutex;

    use ::test_utils::get_column;
    use rstest::rstest;
    use url::Url;

    use super::exec_mode::StagedDataChanges;
    use super::*;
    use crate::actions::deletion_vector::DeletionVectorDescriptor;
    use crate::actions::CommitInfo;
    use crate::arrow::array::builder::{MapBuilder, MapFieldNames, StringBuilder};
    use crate::arrow::array::{
        new_null_array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, NullArray,
        StringArray, StructArray,
    };
    use crate::arrow::datatypes::{
        DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema,
    };
    use crate::arrow::record_batch::RecordBatch;
    use crate::committer::{FileSystemCommitter, PublishMetadata};
    use crate::engine::arrow_conversion::{TryFromArrow, TryIntoArrow};
    use crate::engine::arrow_data::ArrowEngineData;
    use crate::engine::arrow_expression::ArrowEvaluationHandler;
    use crate::engine::sync::SyncEngine;
    use crate::expressions::{MapData, Scalar, StructData};
    use crate::metrics::{MetricEvent, TableType, TransactionCommitFailure};
    use crate::object_store::memory::InMemory;
    use crate::object_store::path::Path;
    use crate::object_store::ObjectStoreExt as _;
    use crate::scan::log_replay::PATH_NAME;
    use crate::scan::state_info::tests::RowTrackingState;
    use crate::schema::{schema, schema_ref, MapType};
    use crate::table_features::{ColumnMappingMode, TableFeature};
    use crate::table_properties::{APPEND_ONLY, ENABLE_CHANGE_DATA_FEED, ENABLE_ROW_TRACKING};
    use crate::transaction::create_table::create_table;
    use crate::transaction::data_layout::DataLayout;
    #[cfg(feature = "adaptive-metadata-in-dev")]
    use crate::unit_test_utils::{
        adaptive_metadata_table_configuration, test_schema_flat_with_column_mapping,
    };
    use crate::unit_test_utils::{
        assert_result_error_with_message, copy_test_table, create_valid_add_file_batch,
        install_thread_local_metrics_reporter, load_test_table, string_array_to_engine_data,
        test_schema_flat, test_schema_nested, test_schema_with_array, test_schema_with_map,
        CapturingReporter, MockProtocolBuilder, MockTableConfigurationBuilder,
    };
    use crate::{EvaluationHandler, KernelResultIterator, Snapshot};

    #[rstest]
    fn transaction_constructor_initializes_all_defaults(
        #[values(false, true)] expected_data_change: bool,
        #[values("", "test-engine")] expected_engine_info: &str,
    ) -> KernelResult<()> {
        let (_engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let _guard = tracing::subscriber::set_default(tracing_subscriber::registry());
        let expected_span = tracing::info_span!("transaction_constructor_defaults");
        assert!(expected_span.id().is_some());
        let expected_commit_timestamp = 1234;

        let transaction = Transaction::<UpdateTable>::new(
            TransactionBuilderState {
                engine_info: expected_engine_info.into(),
                data_change: expected_data_change,
                ..Default::default()
            },
            snapshot.table_configuration().clone(),
            Box::new(FileSystemCommitter::new()),
            UpdateTableOperation::Write.into(),
            expected_span.clone(),
            expected_commit_timestamp,
        );
        let Transaction {
            span,
            operation_id,
            correlation_id,
            read_snapshot_opt,
            effective_table_config,
            should_emit_protocol,
            should_emit_metadata,
            committer,
            operation,
            operation_parameters,
            operation_metrics,
            engine_info,
            engine_commit_info,
            staged_data_changes:
                Imperative {
                    add_files_metadata,
                    remove_files_metadata,
                    dv_matched_files,
                    num_dv_updates,
                },
            set_transactions,
            commit_timestamp,
            user_domain_metadata_additions,
            system_domain_metadata_additions,
            provided_row_tracking_high_water_mark,
            user_domain_removals,
            data_change,
            column_defaults_acknowledged,
            row_tracking_preservation_acknowledged,
            is_blind_append,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            manifest_write,
            physical_clustering_columns,
            _state: _,
        } = transaction;

        assert_eq!(span.id(), expected_span.id());
        assert_ne!(operation_id, MetricId::nil());
        assert!(correlation_id.is_none());
        assert!(read_snapshot_opt.is_none());
        assert_eq!(&effective_table_config, snapshot.table_configuration());
        assert!(!should_emit_protocol);
        assert!(!should_emit_metadata);
        assert!(!committer.is_catalog_committer());
        assert_eq!(operation, UpdateTableOperation::Write.into());
        assert!(operation_parameters.is_none());
        assert!(operation_metrics.is_none());
        assert_eq!(engine_info, expected_engine_info);
        assert!(engine_commit_info.is_none());
        assert!(add_files_metadata.is_empty());
        assert!(remove_files_metadata.is_empty());
        assert!(set_transactions.is_empty());
        assert_eq!(commit_timestamp, expected_commit_timestamp);
        assert!(user_domain_metadata_additions.is_empty());
        assert!(system_domain_metadata_additions.is_empty());
        assert!(provided_row_tracking_high_water_mark.is_none());
        assert!(user_domain_removals.is_empty());
        assert_eq!(data_change, expected_data_change);
        assert!(!column_defaults_acknowledged);
        assert!(!row_tracking_preservation_acknowledged);
        assert!(!is_blind_append);
        assert!(dv_matched_files.is_empty());
        assert_eq!(num_dv_updates, 0);
        #[cfg(feature = "adaptive-metadata-in-dev")]
        assert!(manifest_write.is_none());
        assert!(physical_clustering_columns.is_none());
        Ok(())
    }

    #[test]
    fn transaction_constructor_generates_unique_operation_ids() -> KernelResult<()> {
        let (_engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let first = Transaction::<UpdateTable>::new(
            TransactionBuilderState::default(),
            snapshot.table_configuration().clone(),
            Box::new(FileSystemCommitter::new()),
            UpdateTableOperation::Write.into(),
            tracing::Span::none(),
            1234,
        );
        let second = Transaction::<UpdateTable>::new(
            TransactionBuilderState::default(),
            snapshot.table_configuration().clone(),
            Box::new(FileSystemCommitter::new()),
            UpdateTableOperation::Write.into(),
            tracing::Span::none(),
            1234,
        );

        assert_ne!(first.operation_id, second.operation_id);
        Ok(())
    }

    #[test]
    fn transaction_constructor_preserves_shared_intent() -> KernelResult<()> {
        let (_engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let expected_parameters = HashMap::from([("mode".into(), Some("Create".into()))]);
        let expected_metrics = HashMap::from([("numFiles".into(), Some("3".into()))]);
        let expected_domains = vec![DomainMetadata::new("app.domain".into(), "{}".into())];
        let expected_commit_info_schema = schema_ref! { nullable "tag": STRING };
        let commit_info = string_array_to_engine_data(StringArray::from(vec!["connector-tag"]));
        let expected_commit_info = std::ptr::from_ref(commit_info.as_ref());
        let expected_commit_timestamp = 5678;

        let transaction = Transaction::<CreateTable>::new(
            TransactionBuilderState {
                correlation_id: Some("correlation-id".into()),
                operation_parameters: Some(expected_parameters.clone()),
                operation_metrics: Some(expected_metrics.clone()),
                engine_info: "connector/1.0".into(),
                engine_commit_info: Some((commit_info, expected_commit_info_schema.clone())),
                transaction_ids: vec![("app-one".into(), 7), ("app-two".into(), 8)],
                domain_metadata_additions: expected_domains.clone(),
                data_change: false,
            },
            snapshot.table_configuration().clone(),
            Box::new(MockCatalogCommitter),
            CommitOperation::CreateTable,
            tracing::Span::none(),
            expected_commit_timestamp,
        );

        assert_eq!(transaction.operation, CommitOperation::CreateTable);
        assert_eq!(transaction.engine_info, "connector/1.0");
        assert_eq!(
            transaction.correlation_id.as_deref(),
            Some("correlation-id")
        );
        assert_eq!(transaction.operation_parameters, Some(expected_parameters));
        assert_eq!(transaction.operation_metrics, Some(expected_metrics));
        assert_eq!(transaction.user_domain_metadata_additions, expected_domains);
        assert!(transaction.committer.is_catalog_committer());
        assert_eq!(transaction.commit_timestamp, expected_commit_timestamp);
        assert_eq!(
            transaction.set_transactions,
            vec![
                SetTransaction::new("app-one".into(), 7, Some(expected_commit_timestamp)),
                SetTransaction::new("app-two".into(), 8, Some(expected_commit_timestamp)),
            ]
        );
        let (commit_info, commit_info_schema) = transaction.engine_commit_info.unwrap();
        assert!(std::ptr::eq(expected_commit_info, commit_info.as_ref()));
        assert_eq!(commit_info_schema, expected_commit_info_schema);
        assert!(!transaction.data_change);
        Ok(())
    }

    impl Transaction {
        /// Set clustering columns for testing purposes without needing a table
        /// with the ClusteredTable feature enabled.
        fn with_clustering_columns_for_test(mut self, columns: Vec<ColumnName>) -> Self {
            self.physical_clustering_columns = Some(columns);
            self
        }
    }

    /// A mock committer that always returns an IOError, used to test the retryable error path.
    struct IoErrorCommitter;

    impl Committer for IoErrorCommitter {
        fn commit(
            &self,
            _engine: &dyn Engine,
            _actions: KernelResultIterator<'_, FilteredEngineData>,
            _commit_metadata: CommitMetadata,
        ) -> KernelResult<CommitResponse> {
            Err(KernelError::IOError(std::io::Error::other(
                "simulated IO error",
            )))
        }
        fn is_catalog_committer(&self) -> bool {
            false
        }
        fn publish(
            &self,
            _engine: &dyn Engine,
            _publish_metadata: PublishMetadata,
        ) -> KernelResult<()> {
            Ok(())
        }
    }

    /// A mock committer that always returns a non-retryable (non-IO) error, used to test the
    /// terminal error path.
    struct GenericErrorCommitter;

    impl Committer for GenericErrorCommitter {
        fn commit(
            &self,
            _engine: &dyn Engine,
            _actions: KernelResultIterator<'_, FilteredEngineData>,
            _commit_metadata: CommitMetadata,
        ) -> KernelResult<CommitResponse> {
            Err(KernelError::generic("simulated commit error"))
        }
        fn is_catalog_committer(&self) -> bool {
            false
        }
        fn publish(
            &self,
            _engine: &dyn Engine,
            _publish_metadata: PublishMetadata,
        ) -> KernelResult<()> {
            Ok(())
        }
    }

    /// A mock catalog committer, used to test catalog committer validation.
    struct MockCatalogCommitter;

    impl Committer for MockCatalogCommitter {
        fn commit(
            &self,
            _engine: &dyn Engine,
            _actions: KernelResultIterator<'_, FilteredEngineData>,
            _commit_metadata: CommitMetadata,
        ) -> KernelResult<CommitResponse> {
            // This won't be reached in tests — the validation error fires before commit.
            Ok(CommitResponse::Conflict { version: 0 })
        }
        fn is_catalog_committer(&self) -> bool {
            true
        }
        fn publish(
            &self,
            _engine: &dyn Engine,
            _publish_metadata: PublishMetadata,
        ) -> KernelResult<()> {
            Ok(())
        }
    }

    /// Sets up a snapshot for a table with deletion vector support at version 1
    fn setup_dv_enabled_table() -> (SyncEngine, Arc<Snapshot>) {
        let engine = SyncEngine::new();
        let path =
            std::fs::canonicalize(PathBuf::from("./tests/data/table-with-dv-small/")).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url)
            .at_version(1)
            .build(&engine)
            .unwrap();
        (engine, snapshot)
    }

    fn setup_non_dv_table() -> (SyncEngine, Arc<Snapshot>) {
        let engine = SyncEngine::new();
        let path =
            std::fs::canonicalize(PathBuf::from("./tests/data/table-without-dv-small/")).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url).build(&engine).unwrap();
        (engine, snapshot)
    }

    fn setup_dv_supported_but_disabled_table() -> KernelResult<(Arc<dyn Engine>, Arc<Snapshot>)> {
        let storage = Arc::new(InMemory::new());
        let table_root = url::Url::parse("memory:///").unwrap();
        let engine = Arc::new(SyncEngine::new_with_store(storage.clone()));
        let schema_json = serde_json::json!({
            "type": "struct",
            "fields": [{
                "name": "id",
                "type": "integer",
                "nullable": true,
                "metadata": {}
            }]
        });
        let actions = [
            r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["deletionVectors"]}}"#.to_string(),
            serde_json::json!({
                "metaData": {
                    "id": "test-id",
                    "format": {"provider": "parquet", "options": {}},
                    "schemaString": schema_json.to_string(),
                    "partitionColumns": [],
                    "configuration": {},
                    "createdTime": 1234567890
                }
            })
            .to_string(),
        ]
        .join("\n");

        let commit_path = Path::from("_delta_log/00000000000000000000.json");
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(storage.put(&commit_path, actions.into()))?;
        let engine: Arc<dyn Engine> = engine;
        let snapshot = Snapshot::builder_for(table_root).build(engine.as_ref())?;
        Ok((engine, snapshot))
    }

    /// Creates a test deletion vector descriptor with default values (the DV might not exist on
    /// disk)
    fn create_test_dv_descriptor(path_suffix: &str) -> DeletionVectorDescriptor {
        use crate::actions::deletion_vector::{
            DeletionVectorDescriptor, DeletionVectorStorageType,
        };
        DeletionVectorDescriptor {
            storage_type: DeletionVectorStorageType::PersistedRelative,
            path_or_inline_dv: format!("dv_{path_suffix}"),
            offset: Some(0),
            size_in_bytes: 100,
            cardinality: 1,
        }
    }

    fn create_dv_transaction(
        snapshot: Arc<Snapshot>,
        engine: &dyn Engine,
    ) -> KernelResult<Transaction> {
        snapshot
            .update_table(UpdateTableOperation::Delete, "test_engine", true)
            .build(engine, Box::new(FileSystemCommitter::new()))
    }

    // TODO: create a finer-grained unit tests for transactions (issue#1091)
    #[test]
    fn test_add_files_schema() -> KernelResult<()> {
        let engine = SyncEngine::new();
        let path =
            std::fs::canonicalize(PathBuf::from("./tests/data/table-with-dv-small/")).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url)
            .at_version(1)
            .build(&engine)
            .unwrap();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "default engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))?;

        let schema = txn.add_files_schema();
        let expected = schema! {
            not_null "path": STRING,
            not_null "partitionValues": { STRING => nullable STRING },
            not_null "size": LONG,
            not_null "modificationTime": LONG,
            nullable "stats": {
                nullable NUM_RECORDS: LONG,
                nullable NULL_COUNT: {},
                nullable MIN_VALUES: {},
                nullable MAX_VALUES: {},
                nullable TIGHT_BOUNDS: BOOLEAN,
            },
        };
        assert_eq!(*schema, expected.into());
        Ok(())
    }

    #[rstest]
    #[case::base(false)]
    #[case::row_tracking(true)]
    fn test_add_action_projection_schema(#[case] row_tracking: bool) -> KernelResult<()> {
        let input_schema = if row_tracking {
            with_row_tracking_cols(&BASE_ADD_FILES_SCHEMA)?
        } else {
            BASE_ADD_FILES_SCHEMA.clone()
        };
        let (schema, _) = build_add_action_projection(input_schema.as_ref(), true)?;
        let field_names: Vec<_> = schema.fields().map(|f| f.name().as_str()).collect();
        let expected_field_names = if row_tracking {
            vec![
                "path",
                "partitionValues",
                "size",
                "modificationTime",
                "dataChange",
                "stats",
                "baseRowId",
                "defaultRowCommitVersion",
            ]
        } else {
            vec![
                "path",
                "partitionValues",
                "size",
                "modificationTime",
                "dataChange",
                "stats",
            ]
        };
        assert_eq!(field_names, expected_field_names);
        assert_eq!(schema.field("dataChange"), Some(&*DATA_CHANGE_COLUMN));
        assert_eq!(
            schema.field("stats").unwrap().data_type(),
            &DataType::STRING
        );
        Ok(())
    }

    /// Verifies the Remove projection's `deletionTimestamp` and `extendedFileMetadata` fields.
    /// `extendedFileMetadata` is always a presence predicate; `deletionTimestamp` derives from the
    /// input outside adaptiveMetadata and is fixed to null under it per the RFC.
    #[rstest]
    #[case::standard(false)]
    #[case::adaptive_metadata(true)]
    fn test_remove_action_projection_deletion_timestamp_and_extended_metadata(
        #[case] adaptive_metadata_enabled: bool,
    ) -> KernelResult<()> {
        let commit_timestamp = 123;
        let patch = build_remove_struct_patch(
            commit_timestamp,
            true,  /* data_change */
            &[],   /* columns_to_drop */
            false, /* coalesce_stats_with_parsed */
            adaptive_metadata_enabled,
        )?;
        let path_patch = patch
            .field_patches
            .get("path")
            .expect("path should have inserted fields");
        // Insertions preserve `insert_after` call order: deletionTimestamp, dataChange,
        // extendedFileMetadata, partitionValues.
        let deletion_timestamp = path_patch
            .insertions
            .first()
            .expect("deletionTimestamp should be the first inserted field");
        let extended_file_metadata = path_patch
            .insertions
            .get(2)
            .expect("extendedFileMetadata should follow deletionTimestamp and dataChange");

        let expected_deletion_timestamp = if adaptive_metadata_enabled {
            null_lit(DataType::LONG)
        } else {
            lit(commit_timestamp)
        };
        let expected_extended_file_metadata = Expression::from_pred(Predicate::and_from([
            col!(SIZE_NAME).is_not_null(),
            col!(FILE_CONSTANT_VALUES_NAME, PARTITION_VALUES_NAME).is_not_null(),
        ]));
        assert_eq!(deletion_timestamp.as_ref(), &expected_deletion_timestamp);
        assert_eq!(
            extended_file_metadata.as_ref(),
            &expected_extended_file_metadata
        );
        Ok(())
    }

    #[test]
    fn test_new_deletion_vector_path() -> KernelResult<()> {
        let engine = SyncEngine::new();
        let path =
            std::fs::canonicalize(PathBuf::from("./tests/data/table-with-dv-small/")).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url.clone())
            .at_version(1)
            .build(&engine)
            .unwrap();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "default engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))?;
        let write_state = txn.write_state().unwrap();
        let write_context = write_state.write_context_builder().build().unwrap();

        // Test with empty prefix
        let dv_path1 = write_context.new_deletion_vector_path(String::from(""));
        let abs_path1 = dv_path1.absolute_path()?;
        assert!(abs_path1.as_str().contains(url.as_str()));

        // Test with non-empty prefix
        let prefix = String::from("dv_test");
        let dv_path2 = write_context.new_deletion_vector_path(prefix.clone());
        let abs_path2 = dv_path2.absolute_path()?;
        assert!(abs_path2.as_str().contains(url.as_str()));
        assert!(abs_path2.as_str().contains(&prefix));

        // Test that two paths with same prefix are different (unique UUIDs)
        let dv_path3 = write_context.new_deletion_vector_path(prefix.clone());
        let abs_path3 = dv_path3.absolute_path()?;
        assert_ne!(abs_path2, abs_path3);

        Ok(())
    }

    #[test]
    fn write_state_reflects_schema_configured_on_builder() -> KernelResult<()> {
        let (engine, snapshot) = setup_non_dv_table();
        let txn = snapshot
            .clone()
            .update_table(UpdateTableOperation::Write, "default engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))?;

        // Independently built transactions must retain their own write schemas.
        let initial_write_state = txn.write_state()?;
        let initial_write_context = initial_write_state.write_context_builder().build()?;
        assert!(!initial_write_context
            .logical_data_schema()
            .contains("fresh_column"));

        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .add_column(StructField::nullable("fresh_column", DataType::INTEGER))
            .build(&engine, Box::new(FileSystemCommitter::new()))?;

        let updated_write_state = txn.write_state()?;
        let updated_write_context = updated_write_state.write_context_builder().build()?;
        assert!(updated_write_context
            .logical_data_schema()
            .contains("fresh_column"));
        assert!(updated_write_context
            .physical_data_schema()
            .contains("fresh_column"));
        assert!(!initial_write_context
            .logical_data_schema()
            .contains("fresh_column"));

        Ok(())
    }

    #[test]
    fn schema_changes_are_applied_once_and_persisted_on_commit() -> KernelResult<()> {
        let (engine, txn, _tempdir) = create_existing_table_txn_with(true, |builder| {
            builder.add_column(StructField::nullable("first_column", DataType::INTEGER))
        })?;

        let snapshot = txn.commit(engine.as_ref())?.unwrap_post_commit_snapshot();
        assert!(snapshot.schema().contains("first_column"));

        let snapshot = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .add_column(StructField::nullable("second_column", DataType::STRING))
            .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();

        let schema = snapshot.schema();
        let field_names: Vec<&str> = schema.fields().map(|field| field.name().as_str()).collect();
        let first_index = field_names
            .iter()
            .position(|name| *name == "first_column")
            .expect("first schema change must be present");
        let second_index = field_names
            .iter()
            .position(|name| *name == "second_column")
            .expect("second schema change must be present");
        assert!(first_index < second_index);
        Ok(())
    }

    #[test]
    fn schema_set_nullable_is_persisted_on_commit() -> KernelResult<()> {
        let engine: Arc<dyn Engine> =
            Arc::new(SyncEngine::new_with_store(Arc::new(InMemory::new())));
        let schema = Arc::new(StructType::try_new([
            StructField::not_null("id", DataType::INTEGER),
            StructField::nullable("name", DataType::STRING),
        ])?);
        let snapshot = create_table("memory:///set_nullable", schema, "test", true)
            .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();

        let snapshot = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .set_nullable(column_name!("id"))
            .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
            .commit(engine.as_ref())?
            .unwrap_post_commit_snapshot();

        assert!(snapshot.schema().field("id").unwrap().is_nullable());
        assert!(snapshot.schema().field("name").unwrap().is_nullable());
        Ok(())
    }

    #[test]
    fn schema_add_struct_then_nested_field_is_persisted_on_commit() -> KernelResult<()> {
        let (engine, txn, _tempdir) = create_existing_table_txn_with(true, |builder| {
            builder
                .add_column(StructField::nullable(
                    "address",
                    StructType::try_new([]).unwrap(),
                ))
                .add_column_at(
                    column_name!("address"),
                    StructField::nullable("city", DataType::STRING),
                )
        })?;
        let snapshot = txn.commit(engine.as_ref())?.unwrap_post_commit_snapshot();

        let schema = snapshot.schema();
        let address = schema.field("address").expect("address must exist");
        let DataType::Struct(address) = address.data_type() else {
            panic!("address must be a struct");
        };
        let city = address
            .field("city")
            .expect("city must be nested in address");
        assert_eq!(city.data_type(), &DataType::STRING);
        assert!(city.is_nullable());
        Ok(())
    }

    #[test]
    fn schema_changes_can_precede_staged_data_in_the_same_commit() -> KernelResult<()> {
        let (engine, mut txn, _tempdir) = create_existing_table_txn_with(true, |builder| {
            builder.add_column(StructField::nullable("fresh_column", DataType::INTEGER))
        })?;
        add_dummy_file(&mut txn);

        let snapshot = txn.commit(engine.as_ref())?.unwrap_post_commit_snapshot();

        assert!(snapshot.schema().contains("fresh_column"));
        Ok(())
    }

    #[test]
    fn empty_builder_schema_changes_are_rejected() {
        let (_engine, snapshot) = setup_non_dv_table();
        let result = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .with_schema_changes([]);
        assert_result_error_with_message(
            result,
            "with_schema_changes requires at least one schema operation",
        );
    }

    mod column_defaults {
        use super::*;
        use crate::schema::column_default::{field_with_default, field_with_invalid_default};

        // NB: `test_utils::schema_with_column_defaults` cannot be used here. In `--lib` unit tests
        // the crate under test and the `delta_kernel` that `test_utils` links are two distinct
        // crate instances, so kernel schema types don't unify across the `test_utils` boundary.
        // The `field_with_*` helpers live in-crate (`schema::column_default`) for the same reason.

        /// Builds a transaction whose effective logical schema is `schema`, with the
        /// `allowColumnDefaults` writer feature enabled so any declared defaults are honored.
        fn txn_with_schema(schema: StructType) -> Transaction {
            txn_with_schema_and_writer_features(schema, [TableFeature::AllowColumnDefaults])
        }

        /// Like [`txn_with_schema`] but with an explicit writer-feature list, so a test can
        /// exercise a table that does *not* enable `allowColumnDefaults`. Panics if the table
        /// configuration fails to construct; use [`try_table_config`] to assert that error.
        fn txn_with_schema_and_writer_features(
            schema: StructType,
            writer_features: impl IntoIterator<Item = TableFeature>,
        ) -> Transaction {
            let (engine, snapshot) = setup_non_dv_table();
            let mut txn = snapshot
                .update_table(UpdateTableOperation::Write, "test-engine", true)
                .build(&engine, Box::new(FileSystemCommitter::new()))
                .unwrap();
            let table_config = try_table_config(&txn, schema, writer_features).unwrap();
            txn.replace_effective_table_config(table_config);
            txn
        }

        /// Builds the [`TableConfiguration`] a transaction would carry for `schema` and
        /// `writer_features`, swapping a synthetic schema/protocol onto a real snapshot's config so
        /// the validation `try_new` runs at construction can be exercised without `create_table`.
        /// Returns the construction result so a test can assert eager-validation errors.
        fn try_table_config(
            base: &Transaction,
            schema: StructType,
            writer_features: impl IntoIterator<Item = TableFeature>,
        ) -> KernelResult<TableConfiguration> {
            let metadata = base
                .effective_table_config
                .metadata()
                .clone()
                .with_schema(Arc::new(schema))
                .unwrap();
            let protocol =
                Protocol::try_new_modern(TableFeature::EMPTY_LIST, writer_features).unwrap();
            let version = base.effective_table_config.version();
            TableConfiguration::try_new_from(
                &base.effective_table_config,
                Some(metadata),
                Some(protocol),
                version,
            )
        }

        /// A transaction over a real (non-DV) table, to use as the base config in
        /// [`try_table_config`].
        fn base_txn() -> Transaction {
            let (engine, snapshot) = setup_non_dv_table();
            snapshot
                .update_table(UpdateTableOperation::Write, "test-engine", true)
                .build(&engine, Box::new(FileSystemCommitter::new()))
                .unwrap()
        }

        #[test]
        fn collects_present_defaults_and_skips_columns_without_one() {
            let schema = schema! {
                (field_with_default("parsable", DataType::INTEGER, "42")),
                (field_with_default(
                    "unparsable",
                    DataType::TIMESTAMP,
                    "current_timestamp()",
                )),
                nullable "no_default": STRING,
            };
            let txn = txn_with_schema(schema);

            let defaults = txn.top_level_column_defaults().unwrap();
            assert_eq!(
                defaults.len(),
                2,
                "only columns with a default are returned"
            );
            assert!(!defaults.contains_key("no_default"));

            let parsable = &defaults["parsable"];
            assert_eq!(parsable.raw_sql(), "42");
            assert!(parsable.to_scalar().unwrap().is_some());

            let unparsable = &defaults["unparsable"];
            assert_eq!(unparsable.raw_sql(), "current_timestamp()");
            assert!(unparsable.to_scalar().unwrap().is_none());
        }

        #[test]
        fn returns_empty_map_when_no_column_has_a_default() {
            let schema = schema! {
                nullable "a": INTEGER,
                nullable "b": STRING,
            };
            let txn = txn_with_schema(schema);
            assert!(txn.top_level_column_defaults().unwrap().is_empty());
        }

        #[test]
        fn load_rejects_malformed_default() {
            let schema = schema! {
                (field_with_invalid_default("c")),
            };

            let err = try_table_config(&base_txn(), schema, [TableFeature::AllowColumnDefaults])
                .expect_err("non-string CURRENT_DEFAULT must error at load")
                .to_string();
            assert!(err.contains("non-string"), "got: {err}");
        }

        #[test]
        fn load_tolerates_default_present_but_feature_not_enabled() {
            let schema = schema! {
                (field_with_default("c", DataType::INTEGER, "42")),
            };

            // Orphaned column-default metadata (no `allowColumnDefaults` feature) is tolerated.
            let txn = txn_with_schema_and_writer_features(schema, []);
            assert!(txn.top_level_column_defaults().unwrap().is_empty());
        }
    }

    #[test]
    fn test_write_context_schemas_exclude_partition_columns() -> KernelResult<()> {
        let engine = SyncEngine::new();
        let path = std::fs::canonicalize(PathBuf::from("./tests/data/basic_partitioned/")).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url).build(&engine).unwrap();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "default engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))?;

        let write_state = txn.write_state()?;
        let write_context = write_state
            .write_context_builder()
            .with_partition_values(HashMap::from([(
                "letter".to_string(),
                Scalar::String("a".into()),
            )]))
            .build()?;
        let logical_schema = write_context.logical_data_schema();
        let physical_schema = write_context.physical_data_schema();

        // Both schemas exclude partition columns.
        assert!(
            !logical_schema.contains("letter"),
            "Logical schema should not contain partition column 'letter'"
        );
        assert!(
            !physical_schema.contains("letter"),
            "Physical schema should not contain partition column 'letter' (stored in path)"
        );

        // Both should contain the non-partition columns
        assert!(
            logical_schema.contains("number"),
            "Logical schema should contain data column 'number'"
        );

        assert!(
            physical_schema.contains("number"),
            "Physical schema should contain data column 'number'"
        );

        Ok(())
    }

    /// Loads a snapshot from `table_path` and builds a partitioned write context for the given
    /// partition values. The table must be partitioned.
    fn snapshot_and_partitioned_write_context(
        table_path: &str,
        partition_values: HashMap<String, Scalar>,
    ) -> KernelResult<(Arc<Snapshot>, BoundWriteContext)> {
        let engine = SyncEngine::new();
        let path = std::fs::canonicalize(PathBuf::from(table_path)).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url).build(&engine)?;
        let txn = snapshot
            .clone()
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))?;
        let write_state = txn.write_state()?;
        let wc = write_state
            .write_context_builder()
            .with_partition_values(partition_values)
            .build()?;
        Ok((snapshot, wc))
    }

    /// Helper: evaluates the logical-to-physical transform on the given batch and returns the
    /// output RecordBatch.
    fn eval_logical_to_physical(
        wc: &BoundWriteContext,
        batch: RecordBatch,
    ) -> KernelResult<RecordBatch> {
        let input_schema = StructType::try_from_arrow(batch.schema())?;
        let physical_schema = wc.physical_data_schema();
        let l2p = wc.logical_to_physical();

        let handler = ArrowEvaluationHandler;
        let evaluator = handler.new_expression_evaluator(
            input_schema.into(),
            l2p,
            physical_schema.clone().into(),
        )?;
        let result = ArrowEngineData::try_from_engine_data(
            evaluator.evaluate(&ArrowEngineData::new(batch))?,
        )?;
        Ok(result.record_batch().clone())
    }

    #[rstest]
    #[case::not_materialized("./tests/data/basic_partitioned/", false)]
    #[case::materialized("./tests/data/partitioned_with_materialize_feature/", true)]
    fn test_partition_columns_materialized_in_logical_to_physical(
        #[case] table_path: &str,
        #[case] materialized: bool,
    ) -> KernelResult<()> {
        let (snapshot, wc) = snapshot_and_partitioned_write_context(
            table_path,
            HashMap::from([("letter".to_string(), Scalar::String("a".into()))]),
        )?;
        assert_eq!(
            snapshot
                .table_configuration()
                .protocol()
                .has_table_feature(&TableFeature::MaterializePartitionColumns),
            materialized
        );

        // The input data must exclude the partition column "letter".
        let input_schema = Arc::new(ArrowSchema::new(vec![
            ArrowField::new("number", ArrowDataType::Int64, true),
            ArrowField::new("a_float", ArrowDataType::Float64, true),
        ]));
        let batch = RecordBatch::try_new(
            input_schema,
            vec![
                Arc::new(Int64Array::from(vec![42])) as ArrayRef,
                Arc::new(Float64Array::from(vec![1.5])),
            ],
        )?;
        let rb = eval_logical_to_physical(&wc, batch)?;

        let rb_schema = rb.schema();
        let names: Vec<&str> = rb_schema
            .fields()
            .iter()
            .map(|f| f.name().as_str())
            .collect();
        if materialized {
            assert_eq!(names, vec!["letter", "number", "a_float"]);
            assert_eq!(get_column!(rb, "letter", StringArray).value(0), "a");
        } else {
            assert_eq!(names, vec!["number", "a_float"]);
        }
        Ok(())
    }

    #[rstest]
    #[case::cm_none(ColumnMappingMode::None)]
    #[case::cm_name(ColumnMappingMode::Name)]
    #[case::cm_id(ColumnMappingMode::Id)]
    fn test_materialized_partition_column_insert(
        #[case] cm_mode: ColumnMappingMode,
    ) -> KernelResult<()> {
        let cm = match cm_mode {
            ColumnMappingMode::None => "none",
            ColumnMappingMode::Name => "name",
            ColumnMappingMode::Id => "id",
        };
        let engine: Arc<dyn Engine> =
            Arc::new(SyncEngine::new_with_store(Arc::new(InMemory::new())));
        // Logical order: [p1, p2, d1, v(void), p3, p4, d2]; partition cols = p1, p2, p3, p4.
        let schema = schema_ref! {
            nullable "p1": STRING,
            nullable "p2": INTEGER,
            nullable "d1": INTEGER,
            nullable "v": VOID,
            nullable "p3": STRING,
            nullable "p4": INTEGER,
            nullable "d2": INTEGER,
        };
        let txn = create_table("memory:///t", schema, "DefaultEngine", true)
            .with_data_layout(DataLayout::partitioned(["p1", "p2", "p3", "p4"]))
            .with_table_properties([
                ("delta.feature.materializePartitionColumns", "supported"),
                ("delta.columnMapping.mode", cm),
            ])
            .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;

        let write_state = txn.write_state()?;
        let wc = write_state
            .write_context_builder()
            .with_partition_values(HashMap::from([
                ("p1".to_string(), Scalar::String("aa".into())),
                ("p2".to_string(), Scalar::Integer(7)),
                ("p3".to_string(), Scalar::String("cc".into())),
                ("p4".to_string(), Scalar::Integer(9)),
            ]))
            .build()?;

        // Input excludes partition columns but keeps the void column, in logical schema
        // order: [d1, v, d2].
        let input_schema = Arc::new(ArrowSchema::new(vec![
            ArrowField::new("d1", ArrowDataType::Int32, true),
            ArrowField::new("v", ArrowDataType::Null, true),
            ArrowField::new("d2", ArrowDataType::Int32, true),
        ]));
        let batch = RecordBatch::try_new(
            input_schema,
            vec![
                Arc::new(Int32Array::from(vec![10])) as ArrayRef,
                Arc::new(NullArray::new(1)),
                Arc::new(Int32Array::from(vec![20])),
            ],
        )?;
        let rb = eval_logical_to_physical(&wc, batch)?;

        // With void stripped and partition literals inserted, the output names/order must match
        // the physical schema exactly.
        let rb_schema = rb.schema();
        let names: Vec<&str> = rb_schema
            .fields()
            .iter()
            .map(|f| f.name().as_str())
            .collect();
        let physical_schema = wc.physical_data_schema();
        let expected_names: Vec<&str> = physical_schema
            .fields()
            .map(|f| f.name().as_str())
            .collect();
        assert_eq!(names, expected_names);

        // Verify the transformed data.
        assert_eq!(get_column!(rb, names[0], StringArray).value(0), "aa"); // p1 (prepended)
        assert_eq!(get_column!(rb, names[1], Int32Array).value(0), 7); // p2 (prepended)
        assert_eq!(get_column!(rb, names[2], Int32Array).value(0), 10); // d1
        assert_eq!(get_column!(rb, names[3], StringArray).value(0), "cc"); // p3 (after d1, void
                                                                           // skipped)
        assert_eq!(get_column!(rb, names[4], Int32Array).value(0), 9); // p4 (after d1)
        assert_eq!(get_column!(rb, names[5], Int32Array).value(0), 20); // d2
        Ok(())
    }

    /// Physical schema should include partition columns when materializePartitionColumns is on.
    #[test]
    fn test_physical_schema_includes_partition_columns_when_materialized() -> KernelResult<()> {
        let (_snapshot, write_context) = snapshot_and_partitioned_write_context(
            "./tests/data/partitioned_with_materialize_feature/",
            HashMap::from([("letter".to_string(), Scalar::String("a".into()))]),
        )?;
        let physical_schema = write_context.physical_data_schema();

        assert!(
            physical_schema.contains("letter"),
            "Partition column 'letter' should be in physical schema when materialized"
        );
        assert!(
            physical_schema.contains("number"),
            "Non-partition column 'number' should be in physical schema"
        );
        Ok(())
    }

    // Building the wrong write context for the table's partitioning returns an error.
    #[rstest]
    #[case::partition_value_on_unpartitioned(
        "./tests/data/table-without-dv-small/",
        Some(HashMap::from([("x".to_string(), Scalar::Integer(1))])),
        "not partitioned",
        None
    )]
    #[case::missing_values_on_partitioned(
        "./tests/data/basic_partitioned/",
        None,
        "table is partitioned",
        None
    )]
    #[case::empty_values_on_partitioned(
        "./tests/data/basic_partitioned/",
        Some(HashMap::new()),
        "missing partition column",
        Some("required")
    )]
    fn test_write_context_builder_requires_matching_partition_values(
        #[case] table_path: &str,
        #[case] partition_values: Option<HashMap<String, Scalar>>,
        #[case] expected_msg: &str,
        #[case] unexpected_msg: Option<&str>,
    ) -> KernelResult<()> {
        let engine = SyncEngine::new();
        let path = std::fs::canonicalize(PathBuf::from(table_path)).unwrap();
        let url = url::Url::from_directory_path(path).unwrap();
        let snapshot = Snapshot::builder_for(url).build(&engine)?;
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))?;
        let write_state = txn.write_state()?;
        let mut builder = write_state.write_context_builder();
        if let Some(partition_values) = partition_values {
            builder = builder.with_partition_values(partition_values);
        }
        let err = builder.build().unwrap_err();
        assert!(
            matches!(err, KernelError::InvalidPartitionValues(_)),
            "unexpected error: {err}"
        );
        let err = err.to_string();
        assert!(
            err.contains(expected_msg),
            "expected '{expected_msg}' in error, got: {err}"
        );
        if let Some(unexpected_msg) = unexpected_msg {
            assert!(
                !err.contains(unexpected_msg),
                "did not expect '{unexpected_msg}' in error, got: {err}"
            );
        }
        Ok(())
    }

    /// Tests that update_deletion_vectors validates table protocol requirements.
    /// Validates that attempting DV updates on unsupported tables returns protocol error.
    #[test]
    fn test_update_deletion_vectors_unsupported_table() -> KernelResult<()> {
        let (engine, snapshot) = setup_non_dv_table();
        let mut txn = create_dv_transaction(snapshot, &engine)?;

        let dv_map = HashMap::new();
        let result = txn.update_deletion_vectors(dv_map, std::iter::empty());

        let err = result.expect_err("Should fail on table without DV support");
        let err_msg = err.to_string();
        assert!(
            err_msg.contains("Deletion vector")
                && (err_msg.contains("require") || err_msg.contains("version")),
            "Expected protocol error about DV requirements, got: {err_msg}"
        );
        Ok(())
    }

    #[test]
    fn test_update_deletion_vectors_requires_enablement_property() -> KernelResult<()> {
        let (engine, snapshot) = setup_dv_supported_but_disabled_table()?;
        let mut txn = create_dv_transaction(snapshot, engine.as_ref())?;

        let err = txn
            .update_deletion_vectors(HashMap::new(), std::iter::empty())
            .expect_err("DV updates should require delta.enableDeletionVectors=true");

        assert!(
            matches!(err, KernelError::Unsupported(_)),
            "unexpected error: {err}"
        );
        assert!(
            err.to_string().contains("delta.enableDeletionVectors"),
            "error should mention the enablement property, got: {err}"
        );
        Ok(())
    }

    /// Tests that update_deletion_vectors validates DV descriptors match scan files.
    /// Validates detection of mismatch between provided DV descriptors and actual files.
    #[test]
    fn test_update_deletion_vectors_mismatch_count() -> KernelResult<()> {
        let (engine, snapshot) = setup_dv_enabled_table();
        let mut txn = create_dv_transaction(snapshot, &engine)?;

        let mut dv_map = HashMap::new();
        let descriptor = create_test_dv_descriptor("non_existent");
        dv_map.insert("non_existent_file.parquet".to_string(), descriptor);

        let result = txn.update_deletion_vectors(dv_map, std::iter::empty());

        assert!(
            result.is_err(),
            "Should fail when DV descriptors don't match scan files"
        );
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("matched") && err_msg.contains("does not match"),
            "Expected error about mismatched count (expected 1 descriptor, 0 matched files), got: {err_msg}");
        Ok(())
    }

    /// Tests that a mismatch after scanning some files does not leave staged DV updates behind.
    #[test]
    fn test_update_deletion_vectors_mismatch_does_not_mutate_transaction() -> KernelResult<()> {
        let (engine, snapshot) = setup_dv_enabled_table();
        let mut txn = create_dv_transaction(snapshot.clone(), &engine)?;
        let scan = snapshot.scan_builder().build()?;
        let scan_metadata = scan
            .scan_metadata(&engine)?
            .collect::<KernelResult<Vec<_>>>()?;

        let mut paths = Vec::new();
        for metadata in &scan_metadata {
            paths =
                metadata.visit_scan_files(paths, |paths, scan_file| paths.push(scan_file.path))?;
        }
        let existing_path = paths
            .into_iter()
            .next()
            .ok_or_else(|| KernelError::generic("expected at least one scan file"))?;

        let mut dv_map = HashMap::new();
        dv_map.insert(existing_path, create_test_dv_descriptor("matched"));
        dv_map.insert(
            "non_existent_file.parquet".to_string(),
            create_test_dv_descriptor("missing"),
        );

        let result = txn.update_deletion_vectors(
            dv_map,
            scan_metadata
                .into_iter()
                .map(|metadata| Ok(metadata.scan_files)),
        );

        assert!(
            result.is_err(),
            "Should fail when only some DV descriptors match scan files"
        );
        assert!(
            txn.staged_data_changes.dv_matched_files.is_empty(),
            "Failed DV update should not leave staged file updates"
        );
        Ok(())
    }

    #[test]
    fn test_update_deletion_vectors_iter_error_does_not_mutate_transaction() -> KernelResult<()> {
        let (engine, snapshot) = setup_dv_enabled_table();
        let mut txn = create_dv_transaction(snapshot.clone(), &engine)?;
        txn.staged_data_changes
            .dv_matched_files
            .push(FilteredEngineData::with_all_rows_selected(
                string_array_to_engine_data(StringArray::from(vec!["sentinel"])),
            ));
        let staged_len_before = txn.staged_data_changes.dv_matched_files.len();
        let scan = snapshot.scan_builder().build()?;
        let scan_metadata = scan
            .scan_metadata(&engine)?
            .collect::<KernelResult<Vec<_>>>()?;

        let mut paths = Vec::new();
        for metadata in &scan_metadata {
            paths =
                metadata.visit_scan_files(paths, |paths, scan_file| paths.push(scan_file.path))?;
        }
        let existing_path = paths
            .into_iter()
            .next()
            .ok_or_else(|| KernelError::generic("expected at least one scan file"))?;

        let mut dv_map = HashMap::new();
        dv_map.insert(existing_path, create_test_dv_descriptor("matched"));

        let result = txn.update_deletion_vectors(
            dv_map,
            scan_metadata
                .into_iter()
                .map(|metadata| Ok(metadata.scan_files))
                .chain(std::iter::once(Err(KernelError::generic(
                    "simulated scan metadata failure",
                )))),
        );

        assert!(result.is_err(), "iterator error should propagate");
        assert_eq!(
            txn.staged_data_changes.dv_matched_files.len(),
            staged_len_before,
            "Failed DV update should not stage additional file updates"
        );
        Ok(())
    }

    /// Tests that update_deletion_vectors handles empty DV updates correctly as a no-op.
    /// This edge case occurs when a DELETE operation matches no rows.
    #[test]
    fn test_update_deletion_vectors_empty_inputs() -> KernelResult<()> {
        let (engine, snapshot) = setup_dv_enabled_table();
        let mut txn = create_dv_transaction(snapshot, &engine)?;

        let dv_map = HashMap::new();
        let result = txn.update_deletion_vectors(dv_map, std::iter::empty());

        assert!(
            result.is_ok(),
            "Empty DV updates should succeed as no-op, got error: {result:?}"
        );

        Ok(())
    }

    #[test]
    fn unmatched_dv_inputs_do_not_require_row_tracking_acknowledgment() -> KernelResult<()> {
        let (engine, snapshot) = setup_dv_enabled_table();
        let mut txn = create_dv_transaction(snapshot.clone(), &engine)?;
        let scan = snapshot.scan_builder().build()?;
        txn.update_deletion_vectors(
            HashMap::new(),
            Transaction::scan_metadata_to_engine_data(scan.scan_metadata(&engine)?),
        )?;

        // Since we staged a DV update (with no files matched), the transaction bypasses
        // row-tracking + CDF checks and does not require explicit row tracking acknowledgement.
        assert!(txn.has_data_file_actions());
        assert!(!txn.staged_data_changes.has_dv_updates());
        txn.effective_table_config = MockTableConfigurationBuilder::new()
            .with_protocol(
                MockProtocolBuilder::new()
                    .with_features([TableFeature::RowTracking, TableFeature::ChangeDataFeed])
                    .build(),
            )
            .with_properties([(ENABLE_ROW_TRACKING, true), (ENABLE_CHANGE_DATA_FEED, true)])
            .build();
        add_dummy_file(&mut txn);
        txn.validate_commit()?;

        // Blind append validations only consider staged DV updates (not whether any selection
        // vector actually matched), so a blind append is still disallowed.
        txn.is_blind_append = true;
        assert_result_error_with_message(
            txn.validate_commit(),
            "Blind append cannot update deletion vectors",
        );
        Ok(())
    }

    // ============================================================================
    // validate_blind_append tests
    // ============================================================================
    fn add_dummy_file<STATE>(txn: &mut Transaction<STATE>) {
        let batch = create_valid_add_file_batch(false /* all_nullable */);
        txn.add_files(Box::new(ArrowEngineData::new(batch)));
    }

    #[derive(Clone, Copy, Debug)]
    enum DataRemoval {
        RemoveFile,
        DeletionVectorUpdate,
    }

    fn set_append_only(txn: &mut Transaction, enabled: bool) -> KernelResult<()> {
        let metadata = txn
            .effective_table_config
            .metadata()
            .clone()
            .with_configuration_entry(APPEND_ONLY, enabled.to_string());
        let table_config = TableConfiguration::try_new_from(
            &txn.effective_table_config,
            Some(metadata),
            None,
            txn.effective_table_config.version(),
        )?;
        txn.replace_effective_table_config(table_config);
        Ok(())
    }

    fn make_scan_files(selection_vector: &[bool]) -> FilteredEngineData {
        let schema: ArrowSchema = scan_row_schema().as_ref().try_into_arrow().unwrap();
        let row_count = selection_vector.len();
        let columns = schema
            .fields()
            .iter()
            .map(|field| {
                if field.name() == PATH_NAME {
                    Arc::new(StringArray::from_iter_values(
                        (0..row_count).map(|index| format!("file-{index}.parquet")),
                    )) as ArrayRef
                } else if field.name() == SIZE_NAME {
                    Arc::new(Int64Array::from(vec![1; row_count]))
                } else if field.name() == FILE_CONSTANT_VALUES_NAME {
                    let ArrowDataType::Struct(fields) = field.data_type() else {
                        panic!("fileConstantValues should be a struct");
                    };
                    let child_arrays = fields
                        .iter()
                        .map(|field| {
                            if field.name() == PARTITION_VALUES_NAME {
                                let names = MapFieldNames {
                                    entry: "key_value".to_string(),
                                    key: "key".to_string(),
                                    value: "value".to_string(),
                                };
                                let mut builder = MapBuilder::new(
                                    Some(names),
                                    StringBuilder::new(),
                                    StringBuilder::new(),
                                );
                                for _ in 0..row_count {
                                    builder.append(true).unwrap();
                                }
                                Arc::new(builder.finish()) as ArrayRef
                            } else {
                                new_null_array(field.data_type(), row_count)
                            }
                        })
                        .collect();
                    Arc::new(StructArray::new(fields.clone(), child_arrays, None))
                } else {
                    new_null_array(field.data_type(), row_count)
                }
            })
            .collect();
        let batch = RecordBatch::try_new(Arc::new(schema), columns).unwrap();
        FilteredEngineData::try_new(
            Box::new(ArrowEngineData::new(batch)),
            selection_vector.to_vec(),
        )
        .unwrap()
    }

    fn stage_data_removal(txn: &mut Transaction, removal: DataRemoval, selection_vector: &[bool]) {
        let data = make_scan_files(selection_vector);
        match removal {
            DataRemoval::RemoveFile => txn.remove_files(data),
            DataRemoval::DeletionVectorUpdate => {
                txn.staged_data_changes.dv_matched_files.push(data)
            }
        }
    }

    /// Build a transaction on a writable copy of the `table-without-dv-small` fixture.
    fn create_existing_table_txn() -> KernelResult<(Arc<dyn Engine>, Transaction, tempfile::TempDir)>
    {
        create_existing_table_txn_with(true, |builder| builder)
    }

    fn create_existing_table_txn_with(
        data_change: bool,
        configure: impl FnOnce(UpdateTableTransactionBuilder) -> UpdateTableTransactionBuilder,
    ) -> KernelResult<(Arc<dyn Engine>, Transaction, tempfile::TempDir)> {
        let (url, tempdir) = copy_test_table("table-without-dv-small")?;
        let engine: Arc<dyn Engine> = Arc::new(SyncEngine::new());
        let snapshot = Snapshot::builder_for(url).build(engine.as_ref())?;
        let txn = configure(snapshot.update_table(
            UpdateTableOperation::Write,
            "test-engine",
            data_change,
        ))
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
        Ok((engine, txn, tempdir))
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn test_commit_rejects_file_actions_on_iceberg_native_v4_table() -> KernelResult<()> {
        let (engine, mut txn, _tempdir) = create_existing_table_txn()?;
        txn.replace_effective_table_config(adaptive_metadata_table_configuration(
            test_schema_flat_with_column_mapping(),
            &[TableFeature::IcebergNativeV4Preview],
        ));
        add_dummy_file(&mut txn);
        let err = txn.commit(engine.as_ref()).unwrap_err();
        assert!(
            err.to_string().contains("icebergNativeV4"),
            "unexpected error: {err}"
        );
        Ok(())
    }

    #[rstest]
    #[case::create_with_update_operation(
        true,
        CommitOperation::UpdateTable(UpdateTableOperation::Write),
        "create-table transactions must use the CREATE TABLE operation"
    )]
    #[case::update_with_create_operation(
        false,
        CommitOperation::CreateTable,
        "CREATE TABLE cannot use an update-table transaction"
    )]
    #[case::alter_without_schema_changes(
        false,
        CommitOperation::UpdateTable(UpdateTableOperation::AlterTable),
        "ALTER TABLE requires at least one schema change"
    )]
    #[case::empty_custom_operation(
        false,
        CommitOperation::UpdateTable(UpdateTableOperation::Custom(String::new())),
        "custom operation name cannot be empty"
    )]
    #[case::reserved_custom_operation(
        false,
        CommitOperation::UpdateTable(UpdateTableOperation::Custom("WRITE".to_string())),
        "custom operation name 'WRITE' is reserved"
    )]
    fn operation_compatibility_rejects_invalid_internal_state(
        #[case] is_create_table: bool,
        #[case] operation: CommitOperation,
        #[case] expected: &str,
    ) -> KernelResult<()> {
        // Public builders reject these states. Mutate private intent to test the runtime guard.
        let result = if is_create_table {
            let tempdir = tempfile::tempdir()?;
            let engine = SyncEngine::new();
            let mut txn = create_table(
                tempdir.path().to_string_lossy(),
                test_schema_flat(),
                "test-engine",
                true,
            )
            .build(&engine, Box::new(FileSystemCommitter::new()))?;
            txn.operation = operation;
            txn.validate_operation_compatibility()
        } else {
            let (_engine, mut txn, _tempdir) = create_existing_table_txn()?;
            txn.operation = operation;
            txn.validate_operation_compatibility()
        };

        let error = result.expect_err("invalid internal operation must be rejected");
        assert!(matches!(&error, KernelError::InvalidTransactionState(_)));
        assert!(error.to_string().contains(expected), "{error}");
        Ok(())
    }

    #[test]
    fn test_validate_blind_append_success() -> KernelResult<()> {
        let (_engine, mut txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        add_dummy_file(&mut txn);
        txn.validate_blind_append_semantics()?;
        Ok(())
    }

    #[rstest]
    fn create_table_add_actions_preserve_data_change(
        #[values(false, true)] data_change: bool,
    ) -> KernelResult<()> {
        let tempdir = tempfile::tempdir()?;
        let engine = SyncEngine::new();
        let mut transaction = create_table(
            tempdir.path().to_string_lossy(),
            test_schema_flat(),
            "test-engine",
            data_change,
        )
        .build(&engine, Box::new(FileSystemCommitter::new()))?;
        add_dummy_file(&mut transaction);

        let (mut adds, _) = transaction.generate_adds(&engine, 0)?;
        let add = ArrowEngineData::try_from_engine_data(adds.next().expect("Add action")?)?;
        let add = get_column!(add.record_batch(), "add", StructArray);
        assert_eq!(
            get_column!(add, "dataChange", BooleanArray).value(0),
            data_change
        );
        assert!(adds.next().is_none());
        Ok(())
    }

    #[rstest]
    #[case::append_only_disabled(
        false, /* append_only */
        true, /* data_change */
        &[true, true, true], /* selection_vector */
        false, /* expected_error */
    )]
    #[case::no_data_change(
        true, /* append_only */
        false, /* data_change */
        &[true, true, true], /* selection_vector */
        false, /* expected_error */
    )]
    #[case::all_unselected(
        true, /* append_only */
        true, /* data_change */
        &[false, false, false], /* selection_vector */
        false, /* expected_error */
    )]
    #[case::partial_selected(
        true, /* append_only */
        true, /* data_change */
        &[false, true, false], /* selection_vector */
        true, /* expected_error */
    )]
    #[case::all_selected(
        true, /* append_only */
        true, /* data_change */
        &[true, true, true], /* selection_vector */
        true, /* expected_error */
    )]
    fn append_only_rejects_data_removal_when_data_change(
        #[values(DataRemoval::RemoveFile, DataRemoval::DeletionVectorUpdate)] removal: DataRemoval,
        #[values(0, 1)] batch_index: usize,
        #[case] append_only: bool,
        #[case] data_change: bool,
        #[case] selection_vector: &[bool],
        #[case] expected_error: bool,
    ) -> KernelResult<()> {
        let (_engine, mut txn, _tempdir) =
            create_existing_table_txn_with(data_change, |builder| builder)?;
        set_append_only(&mut txn, append_only)?;
        for index in 0..2 {
            let selection_vector = if index == batch_index {
                selection_vector
            } else {
                &[false, false, false]
            };
            stage_data_removal(&mut txn, removal, selection_vector);
        }

        let result = txn.validate_append_only_semantics();
        if expected_error {
            assert!(matches!(
                result,
                Err(KernelError::InvalidTransactionState(_))
            ));
        } else {
            result?;
        }
        Ok(())
    }

    #[test]
    fn test_validate_blind_append_requires_adds() -> KernelResult<()> {
        let (_engine, txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        let result = txn.validate_blind_append_semantics();
        assert!(matches!(
            result,
            Err(KernelError::InvalidTransactionState(_))
        ));
        Ok(())
    }

    #[test]
    fn test_validate_blind_append_requires_data_change() -> KernelResult<()> {
        let (_engine, mut txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        // Exercise defensive commit validation of an invalid internal state.
        txn.data_change = false;
        add_dummy_file(&mut txn);
        let result = txn.validate_blind_append_semantics();
        assert!(matches!(
            result,
            Err(KernelError::InvalidTransactionState(_))
        ));
        Ok(())
    }

    #[test]
    fn test_validate_blind_append_rejects_removes() -> KernelResult<()> {
        let (_engine, mut txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        add_dummy_file(&mut txn);
        let remove_data = FilteredEngineData::with_all_rows_selected(string_array_to_engine_data(
            StringArray::from(vec!["remove"]),
        ));
        txn.remove_files(remove_data);
        let result = txn.validate_blind_append_semantics();
        assert!(matches!(
            result,
            Err(KernelError::InvalidTransactionState(_))
        ));
        Ok(())
    }

    #[test]
    fn test_validate_blind_append_rejects_dv_updates() -> KernelResult<()> {
        let (_engine, mut txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        add_dummy_file(&mut txn);
        let dv_data = FilteredEngineData::with_all_rows_selected(string_array_to_engine_data(
            StringArray::from(vec!["dv"]),
        ));
        txn.staged_data_changes.dv_matched_files.push(dv_data);
        let result = txn.validate_blind_append_semantics();
        assert!(matches!(
            result,
            Err(KernelError::InvalidTransactionState(_))
        ));
        Ok(())
    }

    #[test]
    fn test_validate_blind_append_rejects_create_table() -> KernelResult<()> {
        let tempdir = tempfile::tempdir()?;
        let schema = schema_ref! { nullable "id": INTEGER };
        let engine = Arc::new(crate::engine::sync::SyncEngine::new());
        let mut txn = create_table(
            tempdir.path().to_str().expect("valid temp path"),
            schema,
            "test_engine",
            true,
        )
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
        // Create-table builders do not expose with_blind_append(). Set the field directly to
        // exercise the defensive runtime check.
        txn.is_blind_append = true;
        add_dummy_file(&mut txn);
        let result = txn.validate_blind_append_semantics();
        assert!(matches!(
            result,
            Err(KernelError::InvalidTransactionState(_))
        ));
        Ok(())
    }

    #[test]
    fn test_blind_append_sets_commit_info_flag() -> KernelResult<()> {
        let commit_info = CommitInfo::new(1, None, None, None, true);
        assert_eq!(commit_info.is_blind_append, Some(true));

        let commit_info_false = CommitInfo::new(1, None, None, None, false);
        assert_eq!(commit_info_false.is_blind_append, None);
        Ok(())
    }

    #[test]
    fn test_blind_append_commit_rejects_no_adds() -> KernelResult<()> {
        let (_engine, txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        // No files added — commit should fail with blind append validation
        let err = txn
            .commit(_engine.as_ref())
            .expect_err("Blind append with no adds should fail");
        assert!(
            err.to_string()
                .contains("Blind append requires at least one added data file"),
            "Unexpected error: {err}"
        );
        Ok(())
    }

    #[test]
    fn test_blind_append_commit_success() -> KernelResult<()> {
        let (engine, mut txn, _tempdir) =
            create_existing_table_txn_with(true, UpdateTableTransactionBuilder::with_blind_append)?;
        add_dummy_file(&mut txn);
        // Blind append with add files should pass validation and proceed to commit.
        // The commit itself may fail due to schema mismatch with the dummy data,
        // but we verify validation (line 415) passes on the Ok path.
        let result = txn.commit(engine.as_ref());
        // If it fails, it should NOT be an InvalidTransactionState error
        if let Err(e) = result {
            assert!(
                !matches!(e, KernelError::InvalidTransactionState(_)),
                "Blind append validation should have passed, got: {e}"
            );
        }
        Ok(())
    }

    // Note: Additional test coverage for partial file matching (where some files in a scan
    // have DV updates but others don't) is provided by the end-to-end integration test
    // kernel/tests/features/dv.rs and kernel/tests/write/remove_dv.rs, which exercise
    // the full deletion vector write workflow including the DvMatchVisitor logic.

    #[test]
    fn test_commit_io_error_returns_retryable_transaction() -> KernelResult<()> {
        let (engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let mut txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(engine.as_ref(), Box::new(IoErrorCommitter))?;
        add_dummy_file(&mut txn);
        let result = txn.commit(engine.as_ref())?;
        assert!(
            matches!(result, CommitResult::Retryable(_)),
            "Expected Retryable, got: {result:?}"
        );
        if let CommitResult::Retryable(retryable) = result {
            assert!(
                retryable.error.to_string().contains("simulated IO error"),
                "Unexpected error: {}",
                retryable.error
            );
        }
        Ok(())
    }

    #[test]
    fn test_existing_table_txn_debug() -> KernelResult<()> {
        let (_engine, txn, _tempdir) = create_existing_table_txn()?;
        let debug_str = format!("{txn:?}");
        // Existing-table transactions should include the snapshot version number
        assert!(
            debug_str.contains("Transaction") && debug_str.contains("read_snapshot version"),
            "Debug output should contain Transaction info: {debug_str}"
        );
        // Should NOT contain "create_table"
        assert!(
            !debug_str.contains("create_table"),
            "Existing table debug should not contain create_table: {debug_str}"
        );
        Ok(())
    }

    // Input schemas have no CM metadata; create_table automatically assigns IDs and
    // physical names when mode is Name or Id.
    #[rstest]
    #[case::flat_none(test_schema_flat(), ColumnMappingMode::None)]
    #[case::flat_name(test_schema_flat(), ColumnMappingMode::Name)]
    #[case::flat_id(test_schema_flat(), ColumnMappingMode::Id)]
    #[case::nested_none(test_schema_nested(), ColumnMappingMode::None)]
    #[case::nested_name(test_schema_nested(), ColumnMappingMode::Name)]
    #[case::nested_id(test_schema_nested(), ColumnMappingMode::Id)]
    #[case::map_none(test_schema_with_map(), ColumnMappingMode::None)]
    #[case::map_name(test_schema_with_map(), ColumnMappingMode::Name)]
    #[case::map_id(test_schema_with_map(), ColumnMappingMode::Id)]
    #[case::array_none(test_schema_with_array(), ColumnMappingMode::None)]
    #[case::array_name(test_schema_with_array(), ColumnMappingMode::Name)]
    #[case::array_id(test_schema_with_array(), ColumnMappingMode::Id)]
    fn test_physical_schema_column_mapping(
        #[case] schema: SchemaRef,
        #[case] mode: ColumnMappingMode,
    ) -> KernelResult<()> {
        let (_engine, txn) = crate::unit_test_utils::setup_column_mapping_txn(schema, mode)?;
        let write_state = txn.write_state().unwrap();
        let write_context = write_state.write_context_builder().build().unwrap();
        crate::unit_test_utils::validate_physical_schema_column_mapping(
            write_context.logical_data_schema(),
            write_context.physical_data_schema(),
            mode,
        );
        Ok(())
    }

    /// Builds two-row [`EngineData`] with logical field names matching [`test_schema_nested`].
    fn build_test_record_batch() -> KernelResult<Box<dyn EngineData>> {
        let schema = test_schema_nested();
        let tag_type = MapType::new(DataType::STRING, DataType::STRING, true);
        let score_type = ArrayType::new(DataType::INTEGER, true);
        let info_fields = vec![
            StructField::nullable("name", DataType::STRING),
            StructField::nullable("age", DataType::INTEGER),
            StructField::nullable("tags", tag_type.clone()),
            StructField::nullable("scores", score_type.clone()),
        ];
        let info1 = Scalar::Struct(StructData::try_new(
            info_fields.clone(),
            vec![
                "alice".into(),
                30i32.into(),
                Scalar::Map(MapData::try_new(tag_type.clone(), [("k1", "v1")])?),
                Scalar::Array(ArrayData::try_new(score_type.clone(), [10i32, 20i32])?),
            ],
        )?);
        let info2 = Scalar::Struct(StructData::try_new(
            info_fields,
            vec![
                "bob".into(),
                25i32.into(),
                Scalar::Map(MapData::try_new(tag_type, [("k2", "v2")])?),
                Scalar::Array(ArrayData::try_new(score_type, [30i32])?),
            ],
        )?);
        let rows = vec![vec![1i64.into(), info1], vec![2i64.into(), info2]];
        ArrowEvaluationHandler.create_many(schema, rows)
    }

    /// Validates that [`BoundWriteContext::logical_to_physical`] correctly renames fields at all
    /// nesting
    /// levels. Builds a RecordBatch with logical names, evaluates the transform, and checks
    /// that the output uses physical names from the physical schema — including nested struct
    /// children.
    fn validate_logical_to_physical_transform(mode: ColumnMappingMode) -> KernelResult<()> {
        let schema = test_schema_nested();
        let (_engine, txn) = crate::unit_test_utils::setup_column_mapping_txn(schema, mode)?;
        let write_state = txn.write_state().unwrap();
        let write_context = write_state.write_context_builder().build().unwrap();
        let logical_schema = write_context.logical_data_schema();
        let physical_schema = write_context.physical_data_schema();
        let logical_to_physical_expression = write_context.logical_to_physical();

        if mode != ColumnMappingMode::None {
            assert_ne!(
                logical_schema, physical_schema,
                "Physical schema should differ from logical schema when column mapping is enabled"
            );
        }

        let data = build_test_record_batch()?;

        // Evaluate the logical_to_physical expression
        let input_schema: SchemaRef = logical_schema.clone();
        let handler = ArrowEvaluationHandler;
        let evaluator = handler.new_expression_evaluator(
            input_schema,
            logical_to_physical_expression.clone(),
            physical_schema.clone().into(),
        )?;
        let result = evaluator.evaluate(data.as_ref())?;
        let result = ArrowEngineData::try_from_engine_data(result)?;
        let result_batch = result.record_batch();

        // Verify: all field names, types, and metadata match the physical schema
        let expected_arrow_schema: ArrowSchema = physical_schema.as_ref().try_into_arrow()?;
        assert_eq!(result_batch.schema().as_ref(), &expected_arrow_schema);

        // Verify: data is preserved (id values)
        let id_col = result_batch
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .expect("id column should be Int64");
        assert_eq!(id_col.values(), &[1i64, 2]);

        Ok(())
    }

    #[rstest]
    #[case::name_mode(ColumnMappingMode::Name)]
    #[case::id_mode(ColumnMappingMode::Id)]
    #[case::none_mode(ColumnMappingMode::None)]
    fn test_logical_to_physical_transform(#[case] mode: ColumnMappingMode) -> KernelResult<()> {
        validate_logical_to_physical_transform(mode)
    }

    #[rstest]
    fn test_logical_to_physical_transform_and_row_tracking_state(
        #[values(
            ColumnMappingMode::None,
            ColumnMappingMode::Name,
            ColumnMappingMode::Id
        )]
        mode: ColumnMappingMode,
        #[values(
            RowTrackingState::Enabled,
            RowTrackingState::SupportedNotEnabled,
            RowTrackingState::Suspended
        )]
        row_tracking_state: RowTrackingState,
    ) -> KernelResult<()> {
        // === Given every column-mapping mode and Row Tracking state ===
        let column_mapping_mode = match mode {
            ColumnMappingMode::None => "none",
            ColumnMappingMode::Name => "name",
            ColumnMappingMode::Id => "id",
        };
        let engine: Arc<dyn Engine> =
            Arc::new(SyncEngine::new_with_store(Arc::new(InMemory::new())));
        let mut txn = create_table(
            "memory:///row_tracking_logical_to_physical",
            schema_ref! { nullable "value": INTEGER },
            "test",
            true,
        )
        .with_table_properties([
            ("delta.columnMapping.mode", column_mapping_mode),
            ("delta.feature.rowTracking", "supported"),
        ])
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
        let mut metadata = txn.effective_table_config.metadata().clone();
        for (key, value) in row_tracking_state.properties() {
            metadata = metadata.with_configuration_entry(key, value);
        }
        let logical_schema = txn.effective_table_config.logical_schema();
        let table_config = TableConfiguration::try_new_with_schema(
            &txn.effective_table_config,
            metadata,
            logical_schema,
        )?;
        txn.replace_effective_table_config(table_config);
        let write_state = txn.write_state()?;
        let base_logical_field = write_state
            .base_logical_data_schema
            .fields()
            .next()
            .ok_or_else(|| KernelError::internal_error("base logical data field is missing"))?
            .clone();
        let base_physical_field = write_state
            .base_physical_data_schema
            .fields()
            .next()
            .ok_or_else(|| KernelError::internal_error("base physical data field is missing"))?
            .clone();
        let physical_row_id_name = write_state
            .materialized_row_id_column_name
            .clone()
            .ok_or_else(|| {
                KernelError::internal_error("materialized Row ID column name is missing")
            })?;
        let physical_row_commit_version_name = write_state
            .materialized_row_commit_version_column_name
            .clone()
            .ok_or_else(|| {
                KernelError::internal_error(
                    "materialized Row Commit Version column name is missing",
                )
            })?;

        // === When the connector specifies its row-tracking input columns ===
        let write_context = write_state
            .write_context_builder()
            .with_row_tracking_columns(RowTrackingMetadataColumns {
                row_id_col_name: Some("connector_row_id"),
                row_commit_version_col_name: Some("connector_row_commit_version"),
            })
            .build();

        // === Then ===
        // Supplying materialized row-tracking columns requires Row Tracking to be enabled.
        // Logical data schema, physical data schema, and logical_to_physical expression must be
        // generated correctly.
        if row_tracking_state != RowTrackingState::Enabled {
            assert_result_error_with_message(
                write_context,
                "Kernel does not allow writing materialized Row IDs or Row Commit Versions when \
                 Row Tracking is not enabled",
            );
            return Ok(());
        }
        let write_context = write_context?;
        assert_eq!(
            write_context.logical_data_schema(),
            &schema_ref! {
                (base_logical_field),
                nullable "connector_row_id": LONG,
                nullable "connector_row_commit_version": LONG,
            }
        );
        assert_eq!(
            write_context.physical_data_schema(),
            &schema_ref! {
                (base_physical_field),
                nullable (physical_row_id_name): LONG,
                nullable (physical_row_commit_version_name): LONG,
            }
        );
        assert_eq!(
            write_context.logical_to_physical(),
            Arc::new(Expression::struct_patch(
                ExpressionStructPatchBuilder::new()
            )?)
        );
        Ok(())
    }

    // =========================================================================
    // Stats validation tests for clustering columns
    // =========================================================================

    /// Per-file stats configuration for test add file helpers.
    enum TestFileStats {
        /// No stats (null stats struct)
        None,
        /// Normal stats with non-null min/max
        Present,
        /// All-null column: nullCount == numRecords, null min/max
        AllNull,
    }

    /// Creates test add file metadata with configurable stats for the "value" column.
    fn create_test_add_files(paths: Vec<&str>, stats: Vec<TestFileStats>) -> Box<dyn EngineData> {
        let value_schema = schema! { nullable "value": LONG };
        let value_fields = value_schema.fields().cloned().collect::<Vec<_>>();
        let value_struct_type = DataType::from(value_schema);
        let stats_schema = schema! {
            nullable NUM_RECORDS: LONG,
            nullable NULL_COUNT: (value_struct_type.clone()),
            nullable MIN_VALUES: (value_struct_type.clone()),
            nullable MAX_VALUES: (value_struct_type.clone()),
        };
        let stats_fields = stats_schema.fields().cloned().collect::<Vec<_>>();
        let stats_type = DataType::from(stats_schema);
        let schema = schema_ref! {
            not_null "path": STRING,
            not_null "partitionValues": { STRING => nullable STRING },
            not_null "size": LONG,
            not_null "modificationTime": LONG,
            nullable "stats": (stats_type.clone()),
        };

        let empty_map = Scalar::Map(
            MapData::try_new(
                MapType::new(DataType::STRING, DataType::STRING, true),
                Vec::<(&str, &str)>::new(),
            )
            .unwrap(),
        );

        let rows: Vec<Vec<Scalar>> = paths
            .iter()
            .zip(stats.iter())
            .map(|(path, stat)| {
                let stats_scalar = match stat {
                    TestFileStats::None => Scalar::Null(stats_type.clone()),
                    TestFileStats::Present | TestFileStats::AllNull => {
                        let value_struct = |v: Option<i64>| {
                            let scalar = v.map_or(Scalar::Null(DataType::LONG), |n| n.into());
                            Scalar::Struct(
                                StructData::try_new(value_fields.clone(), vec![scalar]).unwrap(),
                            )
                        };
                        let (null_count, min, max) = match stat {
                            TestFileStats::Present => (
                                value_struct(Some(0)),
                                value_struct(Some(1)),
                                value_struct(Some(100)),
                            ),
                            _ => (
                                value_struct(Some(100)),
                                value_struct(None),
                                value_struct(None),
                            ),
                        };
                        Scalar::Struct(
                            StructData::try_new(
                                stats_fields.clone(),
                                vec![100i64.into(), null_count, min, max],
                            )
                            .unwrap(),
                        )
                    }
                };
                vec![
                    (*path).into(),
                    empty_map.clone(),
                    1024i64.into(),
                    1000000i64.into(),
                    stats_scalar,
                ]
            })
            .collect();
        ArrowEvaluationHandler.create_many(schema, rows).unwrap()
    }

    #[test]
    fn test_stats_validation_allows_all_null_clustering_column() {
        let (engine, snapshot) = setup_non_dv_table();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))
            .unwrap()
            .with_clustering_columns_for_test(vec![column_name!("value")]);

        let add_files = create_test_add_files(vec!["file1.parquet"], vec![TestFileStats::AllNull]);

        let result = txn.validate_add_files_stats(&[add_files]);

        assert!(
            result.is_ok(),
            "Stats validation should pass for all-null clustering columns, got: {result:?}",
        );
    }

    #[test]
    fn test_stats_validation_when_clustering_cols_missing_stats() {
        let (engine, snapshot) = setup_non_dv_table();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))
            .unwrap()
            // Enable clustering columns for this test
            .with_clustering_columns_for_test(vec![column_name!("value")]);

        // Add files WITHOUT stats
        let add_files = create_test_add_files(vec!["file1.parquet"], vec![TestFileStats::None]);

        // Directly test the validation method instead of committing
        let result = txn.validate_add_files_stats(&[add_files]);

        assert!(
            result.is_err(),
            "Expected validation to fail when stats are missing for clustering columns"
        );

        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("Stats validation error") || err_msg.contains("no stats"),
            "Expected stats validation error, got: {err_msg}"
        );
    }

    #[test]
    fn test_stats_validation_when_clustering_stats_present() {
        let (engine, snapshot) = setup_non_dv_table();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))
            .unwrap()
            // Enable clustering columns for this test
            .with_clustering_columns_for_test(vec![column_name!("value")]);

        // Add files WITH stats
        let add_files = create_test_add_files(vec!["file1.parquet"], vec![TestFileStats::Present]);

        // Directly test the validation method
        let result = txn.validate_add_files_stats(&[add_files]);

        assert!(
            result.is_ok(),
            "Stats validation should pass when stats are present, got: {result:?}"
        );
    }

    #[test]
    fn test_stats_validation_skipped_without_clustering() {
        let (engine, snapshot) = setup_non_dv_table();
        let txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(FileSystemCommitter::new()))
            .unwrap();
        // No clustering columns set (default)

        // Add files WITHOUT stats
        let add_files = create_test_add_files(vec!["file1.parquet"], vec![TestFileStats::None]);

        // Directly test the validation method - should pass because no clustering
        let result = txn.validate_add_files_stats(&[add_files]);

        assert!(
            result.is_ok(),
            "Stats validation should be skipped without clustering, got: {result:?}"
        );
    }

    #[test]
    fn disallow_catalog_committer_for_non_catalog_managed_table() {
        let storage = Arc::new(InMemory::new());
        let table_root = url::Url::parse("memory:///").unwrap();
        let engine = crate::engine::sync::SyncEngine::new_with_store(storage.clone());

        // Create a non-catalog-managed table (no catalogManaged feature)
        let actions = [
            r#"{"commitInfo":{"timestamp":12345678900,"inCommitTimestamp":12345678900}}"#,
            r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":[],"writerFeatures":["inCommitTimestamp"]}}"#,
            r#"{"metaData":{"id":"test-id","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[]}","partitionColumns":[],"configuration":{"delta.enableInCommitTimestamps":"true"},"createdTime":1234567890}}"#,
        ].join("\n");

        let commit_path = Path::from("_delta_log/00000000000000000000.json");
        let rt = tokio::runtime::Runtime::new().unwrap();
        rt.block_on(storage.put(&commit_path, actions.into()))
            .unwrap();

        let snapshot = Snapshot::builder_for(table_root).build(&engine).unwrap();

        // Try to commit with a catalog committer to a non-catalog-managed table
        let committer = Box::new(MockCatalogCommitter);
        let err = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, committer)
            .unwrap()
            .commit(&engine)
            .unwrap_err();
        assert!(matches!(
            err,
            crate::KernelError::Generic(e) if e.contains("This table is path-based and cannot be committed to with a catalog committer")
        ));
    }

    #[test]
    fn disallow_catalog_committer_for_non_catalog_managed_create_table() {
        let storage = Arc::new(InMemory::new());
        let engine = crate::engine::sync::SyncEngine::new_with_store(storage);

        // Create a non-catalog-managed table using a catalog committer
        let schema = schema_ref! { nullable "id": INTEGER };
        let committer = Box::new(MockCatalogCommitter);
        let err = create_table("memory:///", schema, "test-engine", true)
            .build(&engine, committer)
            .unwrap()
            .commit(&engine)
            .unwrap_err();
        assert!(matches!(
            err,
            crate::KernelError::Generic(e) if e.contains("This table is path-based and cannot be committed to with a catalog committer")
        ));
    }

    struct CapturingCommitter {
        captured: Arc<Mutex<Option<i64>>>,
    }

    impl CapturingCommitter {
        fn new() -> (Self, Arc<Mutex<Option<i64>>>) {
            let captured = Arc::new(Mutex::new(None));
            (
                Self {
                    captured: captured.clone(),
                },
                captured,
            )
        }
    }

    impl Committer for CapturingCommitter {
        fn commit(
            &self,
            _engine: &dyn Engine,
            _actions: KernelResultIterator<'_, FilteredEngineData>,
            commit_metadata: CommitMetadata,
        ) -> KernelResult<CommitResponse> {
            *self.captured.lock().unwrap() = Some(commit_metadata.in_commit_timestamp());
            Ok(CommitResponse::Conflict {
                version: commit_metadata.version(),
            })
        }
        fn is_catalog_committer(&self) -> bool {
            false
        }
        fn publish(
            &self,
            _engine: &dyn Engine,
            _publish_metadata: PublishMetadata,
        ) -> KernelResult<()> {
            Ok(())
        }
    }

    #[test]
    fn test_commit_metadata_receives_ict_not_wall_time() -> KernelResult<()> {
        // Set up a table with ICT enabled and a very high previous ICT so that the
        // monotonicity rule (max(wall_time, prev_ict + 1)) produces a value strictly
        // greater than the current wall time. This lets us verify the computed ICT is
        // passed to CommitMetadata (not the wall-clock timestamp).
        let tempdir = tempfile::tempdir().unwrap();
        let log_dir = tempdir.path().join("_delta_log");
        std::fs::create_dir_all(&log_dir).unwrap();

        let future_ict: i64 = 9_999_999_999_999; // far-future timestamp in ms
        let commit_info = serde_json::json!({
            "commitInfo": {
                "timestamp": 1000,
                "operation": "WRITE",
                "inCommitTimestamp": future_ict
            }
        });
        let protocol = serde_json::json!({
            "protocol": {
                "minReaderVersion": 3,
                "minWriterVersion": 7,
                "readerFeatures": [],
                "writerFeatures": ["inCommitTimestamp"]
            }
        });
        let schema_json = serde_json::json!({
            "type": "struct",
            "fields": [{
                "name": "id",
                "type": "integer",
                "nullable": true,
                "metadata": {}
            }]
        });
        let metadata = serde_json::json!({
            "metaData": {
                "id": "test-id",
                "format": {"provider": "parquet", "options": {}},
                "schemaString": schema_json.to_string(),
                "partitionColumns": [],
                "configuration": {
                    "delta.enableInCommitTimestamps": "true"
                }
            }
        });
        let commit0 = format!("{commit_info}\n{protocol}\n{metadata}\n");
        std::fs::write(log_dir.join("00000000000000000000.json"), commit0).unwrap();

        let table_url = Url::from_directory_path(tempdir.path()).unwrap();
        let engine = SyncEngine::new();
        let snapshot = Snapshot::builder_for(table_url).build(&engine)?;

        let prev_ict = snapshot.get_in_commit_timestamp(&engine)?;
        assert_eq!(prev_ict, Some(future_ict));

        let (committer, captured_ts) = CapturingCommitter::new();
        let mut txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(&engine, Box::new(committer))?;
        add_dummy_file(&mut txn);

        let result = txn.commit(&engine)?;
        assert!(
            matches!(result, CommitResult::Conflicted(_)),
            "Expected Conflicted from capturing committer"
        );

        // The ICT in CommitMetadata must be prev_ict + 1 (monotonicity), NOT the wall time.
        let captured = captured_ts
            .lock()
            .unwrap()
            .expect("should have captured a timestamp");
        assert_eq!(
            captured,
            future_ict + 1,
            "CommitMetadata.in_commit_timestamp should be the computed ICT (prev_ict + 1), \
             not the wall-clock time"
        );
        Ok(())
    }

    // ===== Commit failure-metric tests =====

    fn commit_failure_event(reporter: &CapturingReporter) -> Option<TransactionCommitFailure> {
        reporter.events().into_iter().find_map(|event| match event {
            MetricEvent::TransactionCommitFailure(f) => Some(f),
            _ => None,
        })
    }

    #[test]
    fn test_commit_io_error_emits_retryable_io_failure_metric() -> KernelResult<()> {
        let (engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let reporter = Arc::new(CapturingReporter::default());
        let _guard = install_thread_local_metrics_reporter(reporter.clone());
        let mut txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(engine.as_ref(), Box::new(IoErrorCommitter))?;
        add_dummy_file(&mut txn);
        let result = txn.commit(engine.as_ref())?;
        assert!(matches!(result, CommitResult::Retryable(_)));
        let failure = commit_failure_event(&reporter).expect("commit failure event");
        assert_eq!(failure.reason, CommitFailureReason::RetryableIo);
        assert_eq!(failure.table_type, TableType::PathBased);
        Ok(())
    }

    #[test]
    fn test_commit_terminal_error_emits_error_failure_metric() -> KernelResult<()> {
        let (engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let reporter = Arc::new(CapturingReporter::default());
        let _guard = install_thread_local_metrics_reporter(reporter.clone());
        let mut txn = snapshot
            .update_table(UpdateTableOperation::Write, "test-engine", true)
            .build(engine.as_ref(), Box::new(GenericErrorCommitter))?;
        add_dummy_file(&mut txn);
        assert!(txn.commit(engine.as_ref()).is_err());
        let failure = commit_failure_event(&reporter).expect("commit failure event");
        assert_eq!(failure.reason, CommitFailureReason::Error);
        assert_eq!(failure.table_type, TableType::PathBased);
        Ok(())
    }
}