delta_kernel 0.30.0

//! Update table transaction methods.
//!
//! This module contains staging APIs and deletion vector update logic for update-table
//! transactions. Construction lives in [`builder::update_table`](super::builder::update_table).
//! See [`mod.rs`](super) for shared commit logic.

use std::collections::HashMap;
use std::sync::{Arc, LazyLock};

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

#[cfg(feature = "adaptive-metadata-in-dev")]
use super::manifest_commit_state::ManifestCommitState;
#[cfg(feature = "adaptive-metadata-in-dev")]
use super::root_manifest_file::RootManifestFile;
#[cfg(feature = "adaptive-metadata-in-dev")]
use super::ManifestWrite;
use super::{ExecutionMode, Imperative, Transaction, UpdateTable};
use crate::actions::deletion_vector::DeletionVectorDescriptor;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::actions::BackReference;
use crate::actions::{LOG_ADD_SCHEMA, NUM_RECORDS, TIGHT_BOUNDS};
use crate::engine_data::{
    FilteredEngineData, FilteredRowVisitor, GetData, RowIndexIterator, TypedGetData,
};
use crate::error::KernelError;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::expressions::null_lit;
use crate::expressions::{
    col, column_name, lit, ArrayData, ColumnName, ExpressionStructPatchBuilder, Scalar, StructData,
};
use crate::scan::data_skipping::stats_schema::schema_with_all_fields_nullable;
use crate::scan::log_replay::get_scan_metadata_transform_expr;
use crate::scan::{restored_add_schema, scan_row_schema};
use crate::schema::{lazy_schema_ref, ArrayType, SchemaRef, StructField, ToSchema};
use crate::table_features::TableFeature;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::utils::require;
#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::FileMeta;
use crate::{DataType, Engine, Expression, KernelResult};

// =============================================================================
// Update table transactions only
// =============================================================================
impl<MODE: ExecutionMode> Transaction<UpdateTable, MODE> {
    // -------------------------------------------------------------------------
    // Public API
    // -------------------------------------------------------------------------

    /// Remove domain metadata from the Delta log.
    /// If the domain exists in the Delta log, this creates a tombstone to logically delete
    /// the domain. The tombstone preserves the previous configuration value.
    /// If the domain does not exist in the Delta log, this is a no-op.
    /// Note that each domain can only appear once per transaction. That is, multiple operations
    /// on 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).
    /// Removing metadata for multiple distinct domains is allowed.
    pub fn with_domain_metadata_removed(mut self, domain: String) -> Self {
        self.user_domain_removals.push(domain);
        self
    }

    /// Acknowledges that the connector correctly preserves Stable Row IDs and Stable Row Commit
    /// Versions. That is:
    ///
    /// - Copied or updated rows retain their Stable Row IDs.
    /// - Copied rows retain their Stable Row Commit Versions.
    /// - The connector preserves these values in the materialized Row ID and Row Commit Version
    ///   columns.
    /// - The connector also satisfies all protocol MUST requirements for those columns.
    ///
    /// See [Row Tracking] in the Delta protocol for more details.
    ///
    /// Kernel does not validate rewritten files or materialized values. Calling this method asserts
    /// that the connector has satisfied these requirements.
    ///
    /// The Delta protocol specifies this preservation as a SHOULD requirement. Kernel requires it
    /// for compatibility.
    ///
    /// This acknowledgment is required before committing Remove actions or deletion-vector updates
    /// on tables with Row Tracking enabled.
    ///
    /// [Row Tracking]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#row-tracking
    pub fn ack_row_tracking_preservation(&mut self) {
        self.row_tracking_preservation_acknowledged = true;
    }

    /// Set an explicit row-tracking high-water mark for this transaction.
    ///
    /// Use this when row IDs must also be coordinated with another system. Kernel still assigns
    /// row-tracking fields to added files and rejects this value if it is less than the high-water
    /// mark calculated from those files. Callers cannot use [`Self::with_domain_metadata`] to
    /// modify `delta.rowTracking` or other `delta.*` domains. Table-feature and table-state
    /// validation occurs during commit.
    #[internal_api]
    #[allow(dead_code)] // used in FFI
    pub(crate) fn with_row_tracking_high_water_mark(
        mut self,
        high_water_mark: i64,
    ) -> KernelResult<Self> {
        if self.provided_row_tracking_high_water_mark.is_some() {
            return Err(KernelError::generic(
                "Row-tracking high-water mark already specified in this transaction",
            ));
        }
        self.provided_row_tracking_high_water_mark = Some(high_water_mark);
        Ok(self)
    }

    /// Stages `file` to be committed as the table's root manifest.
    ///
    /// Mutually exclusive with [`with_manifest_commit`](Transaction::with_manifest_commit), which
    /// has kernel build the tree instead.
    ///
    /// # Errors
    ///
    /// Returns an error if the table does not support the `adaptiveMetadata-preview` feature, or if
    /// a manifest (content-tree) commit was already staged.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    pub fn with_root_manifest_file(mut self, file: FileMeta) -> KernelResult<Self> {
        require!(
            !matches!(self.manifest_write, Some(ManifestWrite::Commit(_))),
            KernelError::invalid_transaction_state(
                "explicit root manifest and manifest commit are mutually exclusive"
            )
        );
        require!(
            self.effective_table_config
                .is_feature_supported(&TableFeature::AdaptiveMetadataPreview),
            KernelError::unsupported(
                "root manifest file commit requires the adaptiveMetadata-preview feature"
            )
        );
        let read_snapshot = self.read_snapshot_opt.clone().ok_or_else(|| {
            KernelError::internal_error("existing-table transaction unexpectedly has no snapshot")
        })?;
        self.manifest_write = Some(ManifestWrite::RootFile(RootManifestFile::new(
            file,
            read_snapshot,
        )));
        Ok(self)
    }
}

impl Transaction<UpdateTable, Imperative> {
    /// Enables a manifest (content-tree) commit for this transaction, returning the
    /// [`ManifestCommitState`] that hands out leaf writers accepting file changes.
    ///
    /// Mutually exclusive with [`with_root_manifest_file`](Self::with_root_manifest_file), which
    /// commits a caller-supplied root manifest instead of having kernel build the tree. Repeated
    /// calls return the state initialized by the first call. Configure schema changes on the
    /// transaction builder before building the transaction and staging a manifest commit.
    ///
    /// # Errors
    ///
    /// Returns an error if the table does not support the `adaptiveMetadata-preview` feature, if a
    /// root manifest file was already staged, or if delta log commits exist after the last manifest
    /// commit (not yet supported).
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[internal_api]
    pub(crate) fn with_manifest_commit(
        &mut self,
        engine: &dyn Engine,
    ) -> KernelResult<&mut ManifestCommitState> {
        match &self.manifest_write {
            Some(ManifestWrite::RootFile(_)) => {
                return Err(KernelError::invalid_transaction_state(
                    "explicit root manifest and manifest commit are mutually exclusive",
                ))
            }
            // Repeated calls reuse the state from the first call.
            Some(ManifestWrite::Commit(_)) => {}
            None => {
                let read_snapshot = self.read_snapshot_opt.clone().ok_or_else(|| {
                    KernelError::internal_error(
                        "existing-table transaction unexpectedly has no snapshot",
                    )
                })?;
                let state = ManifestCommitState::try_new(
                    engine,
                    read_snapshot,
                    self.get_commit_version(),
                    &self.effective_table_config,
                )?;
                self.manifest_write = Some(ManifestWrite::Commit(state));
            }
        }
        match &mut self.manifest_write {
            Some(ManifestWrite::Commit(state)) => Ok(state),
            _ => Err(KernelError::internal_error(
                "manifest commit state missing after initialization",
            )),
        }
    }

    /// Remove files from the table in this transaction. This API generally enables the engine to
    /// delete data (at file-level granularity) from the table. Note that this API can be called
    /// multiple times to remove multiple batches.
    ///
    /// The expected schema for `remove_metadata` is given by [`scan_row_schema`]. It is expected
    /// this will be the result of passing [`FilteredEngineData`] returned from a scan
    /// with the selection vector modified to select rows for removal (selected rows in the
    /// selection vector are the ones to be removed).
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use std::sync::Arc;
    /// # use delta_kernel::Engine;
    /// # use delta_kernel::snapshot::Snapshot;
    /// # use delta_kernel::committer::FileSystemCommitter;
    /// # fn example(
    /// #     engine: Arc<dyn Engine>,
    /// #     table_url: url::Url,
    /// # ) -> delta_kernel::KernelResult<()> {
    /// // Create a snapshot and transaction
    /// let snapshot = Snapshot::builder_for(table_url).build(engine.as_ref())?;
    /// let mut txn = snapshot
    ///     .clone()
    ///     .update_table(
    ///         delta_kernel::transaction::UpdateTableOperation::Write,
    ///         "my-app/1.0",
    ///         /* data_change */ true,
    ///     )
    ///     .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
    ///
    /// // Get file metadata from a scan
    /// let scan = snapshot.scan_builder().build()?;
    /// let scan_metadata = scan.scan_metadata(engine.as_ref())?;
    ///
    /// // Remove specific files based on scan metadata
    /// for metadata in scan_metadata {
    ///     let metadata = metadata?;
    ///     // In practice, you would modify the selection vector to choose which files to remove
    ///     let files_to_remove = metadata.scan_files;
    ///     txn.remove_files(files_to_remove);
    /// }
    ///
    /// // Commit the transaction
    /// txn.commit(engine.as_ref())?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn remove_files(&mut self, remove_metadata: FilteredEngineData) {
        self.staged_data_changes
            .remove_files_metadata
            .push(remove_metadata);
    }

    // -------------------------------------------------------------------------
    // Deletion vector updates
    // -------------------------------------------------------------------------

    /// Helper function to convert scan metadata iterator to filtered engine data iterator.
    ///
    /// This adapter extracts the `scan_files` field from each [`crate::scan::ScanMetadata`] item,
    /// making it easy to pass scan results directly to `update_deletion_vectors`.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let scan = snapshot.scan_builder().build()?;
    /// let metadata = scan.scan_metadata(engine)?;
    /// let mut dv_map = HashMap::new();
    /// // ... populate dv_map ...
    /// let files_iter = Transaction::scan_metadata_to_engine_data(metadata);
    /// txn.update_deletion_vectors(dv_map, files_iter)?;
    /// ```
    pub fn scan_metadata_to_engine_data(
        scan_metadata: impl Iterator<Item = KernelResult<crate::scan::ScanMetadata>>,
    ) -> impl Iterator<Item = KernelResult<FilteredEngineData>> {
        scan_metadata.map(|result| result.map(|metadata| metadata.scan_files))
    }

    /// Update deletion vectors for files in the table.
    ///
    /// This method can be called multiple times to update deletion vectors for different sets of
    /// files.
    ///
    /// This method takes a map of file paths to new deletion vector descriptors and an iterator
    /// of scan file data. It joins the two together internally and will generate appropriate
    /// remove/add actions on commit to update the deletion vectors.
    ///
    /// Required AddFile fields on matched rows are validated at commit. Staging can therefore
    /// succeed for metadata that commit later rejects, including an empty path, negative size,
    /// or incorrect physical partition keys.
    ///
    /// On commit, each matched file's add action carries `stats.tightBounds: false`.
    ///
    /// # Arguments
    ///
    /// * `new_dv_descriptors` - A map from data file path (as provided in scan operations) to the
    ///   new deletion vector descriptor for that file.
    /// * `existing_data_files` - An iterator over FilteredEngineData from scan metadata using
    ///   [`scan_row_schema`]. Selected rows must preserve the scan-file values and cover every path
    ///   in `new_dv_descriptors`. Files with deletion vectors must have an accurate
    ///   `stats.numRecords` value.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The transaction targets table creation instead of an existing table
    /// - The table does not have deletion vectors enabled via protocol support and the
    ///   `delta.enableDeletionVectors=true` table property
    /// - A file path in `new_dv_descriptors` is not found in `existing_data_files`
    /// - A matched file's scan metadata is missing or has invalid `stats.numRecords`
    /// - A matched file's `stats` is not valid JSON or is not a JSON object
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// let mut txn = snapshot
    ///     .clone()
    ///     .update_table(UpdateTableOperation::Update, "my-app/1.0", /* data_change */ true)
    ///     .build(engine, Box::new(FileSystemCommitter::new()))?;
    ///
    /// let scan = snapshot.scan_builder().build()?;
    /// let files: Vec<FilteredEngineData> = scan.scan_metadata(engine)?
    ///     .collect::<Result<Vec<_>, _>>()?
    ///     .into_iter()
    ///     .map(|sm| sm.scan_files)
    ///     .collect();
    ///
    /// // Create map of file paths to new deletion vector descriptors
    /// let mut dv_map = HashMap::new();
    /// // ... populate dv_map with file paths and their new DV descriptors ...
    ///
    /// txn.update_deletion_vectors(dv_map, files.into_iter())?;
    /// txn.commit(engine)?;
    /// ```
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    #[instrument(
        name = "txn.update_dvs",
        skip_all,
        fields(num_dv_updates = new_dv_descriptors.len()),
        err
    )]
    pub(crate) fn update_deletion_vectors(
        &mut self,
        new_dv_descriptors: HashMap<String, DeletionVectorDescriptor>,
        existing_data_files: impl Iterator<Item = KernelResult<FilteredEngineData>>,
    ) -> KernelResult<()> {
        if self.is_create_table() {
            return Err(KernelError::generic(
                "Deletion vector operations require an existing table",
            ));
        }
        self.ensure_deletion_vectors_enabled()?;

        let mut matched_dv_files = 0;
        let mut matched_files = Vec::new();
        let mut visitor = DvMatchVisitor::new(&new_dv_descriptors);

        // Process each batch of scan file metadata to prepare for DV updates:
        // 1. Visit rows to match file paths against the DV descriptor map
        // 2. Append new DV descriptors and stats as a temporary column to matched files
        // 3. Update selection vector to only keep files that need DV updates
        // 4. Cache the result in dv_matched_files for generating remove/add actions during commit
        for scan_file_result in existing_data_files {
            let scan_file = scan_file_result?;
            visitor.new_entries.clear();
            visitor.matched_file_indexes.clear();
            visitor.visit_rows_of(&scan_file)?;
            let (data, _) = scan_file.into_parts();

            // Update selection vector to keep only files that matched DV descriptors.
            // This ensures we only generate remove/add actions for files being updated.
            let selection_vector =
                selection_vector_for_matches(data.len(), &visitor.matched_file_indexes);
            matched_dv_files += visitor.matched_file_indexes.len();

            // Append two temporary columns to the scan data: the new DV descriptor and the
            // rewritten stats (with `tightBounds: false`).
            let (dv_entries, stats_entries): (Vec<Scalar>, Vec<Scalar>) = visitor
                .new_entries
                .drain(..)
                .map(|entry| (entry.deletion_vector, entry.stats))
                .unzip();
            let new_columns = vec![
                ArrayData::try_new(struct_deletion_vector_schema().clone(), dv_entries)?,
                ArrayData::try_new(ArrayType::new(DataType::STRING, true), stats_entries)?,
            ];
            matched_files.push(FilteredEngineData::try_new(
                data.append_columns(new_dv_column_schema().clone(), new_columns)?,
                selection_vector,
            )?);
        }

        if matched_dv_files != new_dv_descriptors.len() {
            return Err(KernelError::generic(format!(
                "Number of matched DV files does not match number of new DV descriptors: {} != {}",
                matched_dv_files,
                new_dv_descriptors.len()
            )));
        }

        self.staged_data_changes
            .dv_matched_files
            .extend(matched_files);
        self.staged_data_changes.num_dv_updates += matched_dv_files;
        Ok(())
    }

    /// Verify the table has deletion vectors *enabled* (feature supported in both reader and
    /// writer features AND the `delta.enableDeletionVectors` table property set to `true`).
    fn ensure_deletion_vectors_enabled(&self) -> KernelResult<()> {
        if !self
            .effective_table_config
            .is_feature_enabled(&TableFeature::DeletionVectors)
        {
            return Err(KernelError::unsupported(
                "Deletion vector writes require reader version 3, writer version 7, the \
                 'deletionVectors' feature in both reader and writer features, and the \
                 `delta.enableDeletionVectors` table property set to `true`",
            ));
        }
        Ok(())
    }
}

fn selection_vector_for_matches(num_rows: usize, matched_file_indexes: &[usize]) -> Vec<bool> {
    let mut next_match = 0;
    (0..num_rows)
        .map(|row_index| {
            if matched_file_indexes.get(next_match) == Some(&row_index) {
                next_match += 1;
                true
            } else {
                false
            }
        })
        .collect()
}

// =============================================================================
// Deletion vector schemas and commit helpers
// =============================================================================

/// Column name for temporary column used during deletion vector updates.
/// This column holds new DV descriptors appended to scan file metadata before transforming to final
/// add actions.
static NEW_DELETION_VECTOR_NAME: &str = "newDeletionVector";

/// Column name for the temporary column holding the rewritten `stats` string for a DV update.
static NEW_STATS_NAME: &str = "newStats";

/// Schema for scan row data with an additional column for new deletion vector descriptors.
/// This is an intermediate schema used during deletion vector updates before transforming to final
/// add actions.
static INTERMEDIATE_DV_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    ..(scan_row_schema().fields()),
    nullable NEW_DELETION_VECTOR_NAME: (DeletionVectorDescriptor::to_schema()),
    nullable NEW_STATS_NAME: STRING,
};

/// Returns the intermediate schema with deletion vector column appended to scan row schema.
pub(super) fn intermediate_dv_schema() -> &'static SchemaRef {
    &INTERMEDIATE_DV_SCHEMA
}

/// Schema for scan row data with nullable statistics fields.
/// Used when generating remove actions to ensure statistics can be null if missing.
static NULLABLE_SCAN_ROWS_SCHEMA: LazyLock<SchemaRef> =
    LazyLock::new(|| schema_with_all_fields_nullable(scan_row_schema().as_ref()).into());

/// Returns the nullable scan row schema.
fn nullable_scan_rows_schema() -> &'static SchemaRef {
    &NULLABLE_SCAN_ROWS_SCHEMA
}

/// Schema for restored add actions with nullable statistics fields.
/// Used when transforming scan data back to add actions with potentially missing statistics.
static NULLABLE_RESTORED_ADD_SCHEMA: LazyLock<SchemaRef> =
    LazyLock::new(|| schema_with_all_fields_nullable(restored_add_schema()).into());

/// Returns the nullable restored add action schema.
fn nullable_restored_add_schema() -> &'static SchemaRef {
    &NULLABLE_RESTORED_ADD_SCHEMA
}

/// Schema for add actions that is nullable for use in transforms as as a workaround to avoid issues
/// with null values in required fields that aren't selected.
static NULLABLE_ADD_LOG_SCHEMA: LazyLock<SchemaRef> =
    LazyLock::new(|| schema_with_all_fields_nullable(&LOG_ADD_SCHEMA).into());

/// Returns the schema for nullable restored add actions with dataChange field.
/// This schema extends the nullable restored add schema with a dataChange boolean field
/// that indicates whether the add action represents a logical data change.
fn nullable_add_log_schema() -> &'static SchemaRef {
    &NULLABLE_ADD_LOG_SCHEMA
}

/// Schema for an array of deletion vector descriptors.
/// Used when appending DV columns to scan file data.
#[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
static STRUCT_DELETION_VECTOR_SCHEMA: LazyLock<ArrayType> =
    LazyLock::new(|| ArrayType::new(DeletionVectorDescriptor::to_schema(), true));

/// Returns the schema for an array of deletion vector descriptors.
#[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
fn struct_deletion_vector_schema() -> &'static ArrayType {
    &STRUCT_DELETION_VECTOR_SCHEMA
}

/// Schema for the intermediate column holding new DV descriptors.
/// This temporary column is dropped during transformation to final add actions.
static NEW_DV_COLUMN_SCHEMA: LazyLock<SchemaRef> = lazy_schema_ref! {
    nullable NEW_DELETION_VECTOR_NAME: (DeletionVectorDescriptor::to_schema()),
    nullable NEW_STATS_NAME: STRING,
};

/// Returns the schema for the intermediate column holding new DV descriptors.
pub(super) fn new_dv_column_schema() -> &'static SchemaRef {
    &NEW_DV_COLUMN_SCHEMA
}

// DV updates can only be staged on update-table transactions. The create-table guard protects
// action generation, which is shared by all imperative transaction states.
impl<STATE> Transaction<STATE, Imperative> {
    /// Generate remove/add action pairs for files with DV updates.
    ///
    /// This method processes the cached matched files, generating the necessary Remove and Add
    /// actions. For each file:
    /// 1. A Remove action is generated for the old file
    /// 2. An Add action is generated with the new DV descriptor
    pub(super) fn generate_dv_update_actions<'a>(
        &'a self,
        engine: &'a dyn Engine,
    ) -> KernelResult<impl Iterator<Item = KernelResult<FilteredEngineData>> + Send + 'a> {
        // Create-table transactions should not have any DV update actions
        if self.is_create_table() && !self.staged_data_changes.dv_matched_files.is_empty() {
            return Err(crate::error::KernelError::internal_error(
                "CREATE TABLE transaction cannot have DV update actions",
            ));
        }

        let remove_actions = self.generate_remove_actions(
            engine,
            self.staged_data_changes.dv_matched_files.iter(),
            true, /* has_dv_update_columns */
        )?;
        let add_actions = self.generate_adds_for_dv_update(
            engine,
            self.staged_data_changes.dv_matched_files.iter(),
        )?;
        Ok(remove_actions.chain(add_actions))
    }

    /// Generates Add actions for files with updated deletion vectors.
    ///
    /// This transforms scan file metadata with new DV descriptors (appended as a temporary column)
    /// into Add actions for the Delta log.
    fn generate_adds_for_dv_update<'a>(
        &'a self,
        engine: &'a dyn Engine,
        file_metadata_batch: impl Iterator<Item = &'a FilteredEngineData> + Send + 'a,
    ) -> KernelResult<impl Iterator<Item = KernelResult<FilteredEngineData>> + Send + 'a> {
        let evaluation_handler = engine.evaluation_handler();
        // Struct patch to replace the deletionVector field with the new DV/stats from
        // NEW_DELETION_VECTOR_NAME/NEW_STATS_NAME, then drop the
        // NEW_DELETION_VECTOR_NAME/NEW_STATS_NAME columns.
        let with_new_dv_expr = Expression::struct_patch(
            ExpressionStructPatchBuilder::new()
                .replace("deletionVector", col!(NEW_DELETION_VECTOR_NAME))
                .replace("stats", col!(NEW_STATS_NAME))
                .drop(NEW_DELETION_VECTOR_NAME)
                .drop(NEW_STATS_NAME),
        )?;
        // TODO(#3263): `file_metadata_batch` may contain `stats_parsed` and
        // `partitionValues_parsed`; provide its full schema to both evaluators.
        let with_new_dv_eval = evaluation_handler.new_expression_evaluator(
            intermediate_dv_schema().clone(),
            Arc::new(with_new_dv_expr),
            nullable_scan_rows_schema().clone().into(),
        )?;
        let restored_add_eval = evaluation_handler.new_expression_evaluator(
            nullable_scan_rows_schema().clone(),
            get_scan_metadata_transform_expr(),
            nullable_restored_add_schema().clone().into(),
        )?;
        #[cfg_attr(not(feature = "adaptive-metadata-in-dev"), allow(unused_mut))]
        let mut add_patch = ExpressionStructPatchBuilder::new_nested(["add"])
            .insert_after("modificationTime", lit(self.data_change));
        // 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_ADD_SCHEMA.
        #[cfg(feature = "adaptive-metadata-in-dev")]
        {
            add_patch = add_patch.append(null_lit(BackReference::to_schema()));
        }
        let with_data_change_patch = Expression::struct_patch(add_patch)?;
        let with_data_change_expr = Arc::new(Expression::struct_from([with_data_change_patch]));
        let with_data_change_eval = evaluation_handler.new_expression_evaluator(
            nullable_restored_add_schema().clone(),
            with_data_change_expr,
            nullable_add_log_schema().clone().into(),
        )?;
        Ok(file_metadata_batch.map(
            move |file_metadata_batch| -> KernelResult<FilteredEngineData> {
                let with_new_dv_data = with_new_dv_eval.evaluate(file_metadata_batch.data())?;

                let as_partial_add_data = restored_add_eval.evaluate(with_new_dv_data.as_ref())?;

                let with_data_change_data =
                    with_data_change_eval.evaluate(as_partial_add_data.as_ref())?;

                FilteredEngineData::try_new(
                    with_data_change_data,
                    file_metadata_batch.selection_vector().to_vec(),
                )
            },
        ))
    }
}

// =============================================================================
// DvMatchVisitor: matches file paths from scan data against new DV descriptors
// =============================================================================

/// One visited row's new deletion vector descriptor paired with its rewritten stats.
struct DvUpdateEntry {
    deletion_vector: Scalar,
    stats: Scalar,
}

impl DvUpdateEntry {
    /// The (null DV, null stats) entry for rows that are not getting a DV update.
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    fn null() -> Self {
        static NULL_DV: LazyLock<Scalar> =
            LazyLock::new(|| Scalar::null(DeletionVectorDescriptor::to_schema()));
        static NULL_STATS: LazyLock<Scalar> = LazyLock::new(|| Scalar::Null(DataType::STRING));
        Self {
            deletion_vector: NULL_DV.clone(),
            stats: NULL_STATS.clone(),
        }
    }
}

/// Visitor that matches file paths from scan data against new deletion vector descriptors.
/// Used by update_deletion_vectors() to attach new DV descriptors to scan file metadata.
#[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
struct DvMatchVisitor<'a> {
    /// Map from file path to the new deletion vector descriptor for that file
    dv_updates: &'a HashMap<String, DeletionVectorDescriptor>,
    /// Accumulated (DV descriptor, rewritten stats) pairs for each visited row, in visit order.
    /// Matched rows carry the new DV and their original stats with `tightBounds` forced to
    /// `false`.
    new_entries: Vec<DvUpdateEntry>,
    /// Indexes of rows that matched a file path in dv_update. These must be in
    /// ascending order
    matched_file_indexes: Vec<usize>,
}

impl<'a> DvMatchVisitor<'a> {
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    const PATH_INDEX: usize = 0;
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    const STATS_INDEX: usize = 1;

    /// Creates a new DvMatchVisitor that will match file paths against the provided DV updates map.
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    fn new(dv_updates: &'a HashMap<String, DeletionVectorDescriptor>) -> Self {
        Self {
            dv_updates,
            new_entries: Vec::new(),
            matched_file_indexes: Vec::new(),
        }
    }
}

/// A `FilteredRowVisitor` that matches file paths against the provided DV updates map.
impl FilteredRowVisitor for DvMatchVisitor<'_> {
    fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]) {
        static NAMES_AND_TYPES: LazyLock<(Vec<ColumnName>, Vec<DataType>)> = LazyLock::new(|| {
            let names = vec![column_name!("path"), column_name!("stats")];
            let types = vec![DataType::STRING, DataType::STRING];
            (names, types)
        });
        (&NAMES_AND_TYPES.0, &NAMES_AND_TYPES.1)
    }

    /// For each selected row checks if the path is in the hash-map and if so, extracts DV
    /// details that can be appended back to the EngineData. Also tracks matched row indexes
    /// so the selection vector can be updated to contain only DV-matched files.
    fn visit_filtered<'a>(
        &mut self,
        getters: &[&'a dyn GetData<'a>],
        rows: RowIndexIterator<'_>,
    ) -> KernelResult<()> {
        static DV_SCHEMA_FIELDS: LazyLock<Vec<StructField>> = LazyLock::new(|| {
            DeletionVectorDescriptor::to_schema()
                .into_fields()
                .collect()
        });
        let num_rows = rows.num_rows();
        self.new_entries.reserve(num_rows);
        for row_index in rows {
            // Fill in null entries for any deselected rows before this one.
            self.new_entries.resize_with(row_index, DvUpdateEntry::null);
            let path_opt: Option<String> = getters[Self::PATH_INDEX].get_opt(row_index, "path")?;
            let Some(path) = path_opt else {
                // Null path means a non-add action row (remove, metadata, etc.)
                self.new_entries.push(DvUpdateEntry::null());
                continue;
            };
            if let Some(dv_result) = self.dv_updates.get(&path) {
                // DVs require an accurate numRecords stat per the Delta protocol.
                let stats: Option<String> =
                    getters[Self::STATS_INDEX].get_opt(row_index, "stats")?;
                let stats = stats.ok_or_else(|| {
                    KernelError::generic(format!(
                        "update_deletion_vectors: file {path} has no stats; \
                         deletion vectors require an accurate {NUM_RECORDS}"
                    ))
                })?;
                let mut parsed: serde_json::Value = serde_json::from_str(&stats).map_err(|e| {
                    KernelError::generic(format!(
                        "update_deletion_vectors: stats for {path} is not valid JSON: {e}"
                    ))
                })?;
                let stats_obj = parsed.as_object_mut().ok_or_else(|| {
                    KernelError::generic(format!(
                        "update_deletion_vectors: stats for {path} is not a JSON object"
                    ))
                })?;
                if stats_obj
                    .get(NUM_RECORDS)
                    .and_then(serde_json::Value::as_u64)
                    .is_none()
                {
                    return Err(KernelError::generic(format!(
                        "update_deletion_vectors: stats for {path} is missing {NUM_RECORDS} \
                         or it is not a non-negative integer"
                    )));
                }

                // Widen tightBounds to false (unless already false) instead of recomputing the
                // stats.
                let already_wide = stats_obj
                    .get(TIGHT_BOUNDS)
                    .and_then(serde_json::Value::as_bool)
                    == Some(false);
                let new_stats = if already_wide {
                    stats
                } else {
                    stats_obj.insert(TIGHT_BOUNDS.to_string(), serde_json::Value::Bool(false));
                    serde_json::to_string(&parsed).map_err(|e| {
                        KernelError::generic(format!(
                            "update_deletion_vectors: failed to re-serialize stats for {path}: {e}"
                        ))
                    })?
                };
                let deletion_vector = Scalar::Struct(StructData::try_new(
                    DV_SCHEMA_FIELDS.clone(),
                    vec![
                        Scalar::from(dv_result.storage_type.to_string()),
                        Scalar::from(dv_result.path_or_inline_dv.clone()),
                        Scalar::from(dv_result.offset),
                        Scalar::from(dv_result.size_in_bytes),
                        Scalar::from(dv_result.cardinality),
                    ],
                )?);
                self.new_entries.push(DvUpdateEntry {
                    deletion_vector,
                    stats: Scalar::from(new_stats),
                });
                self.matched_file_indexes.push(row_index);
            } else {
                self.new_entries.push(DvUpdateEntry::null());
            }
        }
        // Pad with trailing null entries for any deselected rows at the end.
        self.new_entries.resize_with(num_rows, DvUpdateEntry::null);
        Ok(())
    }
}

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

    use super::*;
    use crate::actions::deletion_vector::DeletionVectorStorageType;
    use crate::arrow::array::{ArrayRef, StringArray};
    use crate::arrow::datatypes::{DataType as ArrowDataType, Field as ArrowField, Schema};
    use crate::arrow::record_batch::RecordBatch;
    use crate::engine::arrow_data::ArrowEngineData;

    const TEST_PATH: &str = "data/file.parquet";

    /// Build a single-row scan-metadata-like batch. `path` and `stats` are addressed by name
    /// (matching `DvMatchVisitor::selected_column_names_and_types`), so the surrounding
    /// `extra_columns` and their positions deliberately exercise name-based getter binding.
    fn make_scan_metadata_row(
        path: &str,
        stats: Option<&str>,
        extra_columns: &[&str],
    ) -> FilteredEngineData {
        let mut fields = vec![
            ArrowField::new("path", ArrowDataType::Utf8, true),
            ArrowField::new("stats", ArrowDataType::Utf8, true),
        ];
        let mut columns: Vec<ArrayRef> = vec![
            Arc::new(StringArray::from(vec![Some(path)])),
            Arc::new(StringArray::from(vec![stats])),
        ];
        for name in extra_columns {
            fields.push(ArrowField::new(*name, ArrowDataType::Utf8, true));
            columns.push(Arc::new(StringArray::from(vec![Some("ignored")])));
        }
        let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
        FilteredEngineData::with_all_rows_selected(Box::new(ArrowEngineData::new(batch)))
    }

    fn dv_updates_for(path: &str) -> HashMap<String, DeletionVectorDescriptor> {
        let descriptor = DeletionVectorDescriptor::try_new(
            DeletionVectorStorageType::Inline,
            "AAAAAAAA".to_string(),
            None,
            8,
            0,
        )
        .unwrap();
        HashMap::from([(path.to_string(), descriptor)])
    }

    #[rstest]
    #[case::null_stats(None, &[], Some(NUM_RECORDS))]
    #[case::stats_missing_num_records(Some("{}"), &[], Some(NUM_RECORDS))]
    #[case::stats_not_json_object(Some("5"), &[], Some("not a JSON object"))]
    #[case::stats_invalid_json(Some("{not json"), &[], Some("not valid JSON"))]
    #[case::stats_with_num_records(Some(r#"{"numRecords":10}"#), &[], None)]
    #[case::extra_columns_dont_shift_indexes(
        Some(r#"{"numRecords":10}"#),
        &["size", "modificationTime"],
        None,
    )]
    fn dv_match_visitor_validates_matched_row_stats(
        #[case] stats: Option<&str>,
        #[case] extra_columns: &[&str],
        #[case] expected_err: Option<&str>,
    ) {
        let dv_updates = dv_updates_for(TEST_PATH);
        let mut visitor = DvMatchVisitor::new(&dv_updates);
        let data = make_scan_metadata_row(TEST_PATH, stats, extra_columns);

        let result = visitor.visit_rows_of(&data);
        match expected_err {
            Some(sub_str) => {
                let msg = result.expect_err("expected error").to_string();
                assert!(msg.contains(sub_str), "message was: {msg}");
                assert!(msg.contains(TEST_PATH), "message was: {msg}");
            }
            None => {
                result.expect("validation should pass");
                assert_eq!(visitor.matched_file_indexes, vec![0]);
            }
        }
    }

    #[test]
    fn dv_match_visitor_skips_validation_for_unmatched_rows() {
        let dv_updates = dv_updates_for("other/file.parquet");
        let mut visitor = DvMatchVisitor::new(&dv_updates);
        let data = make_scan_metadata_row(TEST_PATH, None, &[]);

        visitor
            .visit_rows_of(&data)
            .expect("unmatched rows must not trigger stats validation");
        assert!(visitor.matched_file_indexes.is_empty());
    }
}