delta_kernel 0.30.0

//! Write-side translation for the Adaptive Metadata Tree (AMT).
//!
//! Translates Delta file write-metadata into content-tree entry [`EngineData`] -- the columnar
//! form of an AMT root manifest. This is the minimal blind-append path: it produces `Data`
//! entries only, with deletion vectors, tags, and leaf-manifest information left null, and
//! statistics and partition values omitted from the schema.

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

use crate::actions::NUM_RECORDS;
use crate::content_tree::{
    struct_expr_from_schema, ContentTreeNodeEntry, DataContentType, DataFileFormat, TrackingInfo,
    TrackingStatus, CONTENT_TYPE, FILE_FORMAT, FILE_SEQUENCE_NUMBER, FILE_SIZE_IN_BYTES,
    FIRST_ROW_ID, FORMAT_VERSION, LOCATION, PARTITION_SPEC_ID, RECORD_COUNT, SEQUENCE_NUMBER,
    TRACKING, TRACKING_SNAPSHOT_ID, TRACKING_STATUS,
};
use crate::engine_data::{EngineData, GetData, RowVisitor, TypedGetData as _};
use crate::expressions::{lit, ColumnName, Expression};
use crate::scan::log_replay::{
    BASE_ROW_ID_NAME, DEFAULT_ROW_COMMIT_VERSION_NAME, PATH_NAME, SIZE_NAME, STATS_NAME,
};
use crate::schema::{
    ColumnNamesAndTypes, DataType, SchemaRef, SchemaStructPatchBuilder, StructField, StructType,
    ToSchema as _,
};
use crate::transaction::augmented_write_metadata_schema;
use crate::{Engine, KernelError, KernelResult};

/// The Iceberg format version stamped onto each written entry: 4 (the V4 adaptive metadata tree).
const AMT_FORMAT_VERSION: i32 = 4;

/// Dotted path of the `stats.numRecords` leaf column, used for column selection and error messages.
const STATS_NUM_RECORDS: &str = "stats.numRecords";

/// The write-metadata input schema consumed by [`convert_append_metadata_to_root_entry_batch`]: the
/// [`augmented_write_metadata_schema`] projected to the columns this path reads and with `stats`
/// narrowed to `numRecords`. Derived from that single source so it cannot drift.
fn write_metadata_input_schema() -> KernelResult<SchemaRef> {
    let augmented = augmented_write_metadata_schema()?;
    let projected = augmented.project(&[
        PATH_NAME,
        SIZE_NAME,
        STATS_NAME,
        BASE_ROW_ID_NAME,
        DEFAULT_ROW_COMMIT_VERSION_NAME,
    ])?;
    let narrowed_stats = match projected.field(STATS_NAME).map(StructField::data_type) {
        Some(DataType::Struct(stats)) => stats.project_as_struct(&[NUM_RECORDS])?,
        _ => {
            return Err(KernelError::generic(
                "add-file write-metadata schema is missing the `stats` struct",
            ))
        }
    };
    let narrowed = SchemaStructPatchBuilder::new()
        .replace(
            STATS_NAME,
            StructField::nullable(STATS_NAME, narrowed_stats),
        )
        .build(&projected)?;
    Ok(Arc::new(narrowed))
}

/// The write-metadata leaf columns this path requires to be non-null on every row, in leaf order:
/// `stats.numRecords`, `baseRowId`, `defaultRowCommitVersion`. The evaluator does not enforce
/// input-schema non-nullability, so [`RequiredFieldsNonNullVisitor`] rejects nulls up front:
/// otherwise a null would land in a root-manifest entry field the manifest requires (`recordCount`,
/// `firstRowId`, `sequenceNumber`, `fileSequenceNumber`). `path`/`size` are exempt because the
/// upstream add-file writer always populates them.
static REQUIRED_NON_NULL_COLUMNS: LazyLock<ColumnNamesAndTypes> = LazyLock::new(|| {
    StructType::new_unchecked([
        StructField::not_null(
            STATS_NAME,
            StructType::new_unchecked([StructField::not_null(NUM_RECORDS, DataType::LONG)]),
        ),
        StructField::not_null(BASE_ROW_ID_NAME, DataType::LONG),
        StructField::not_null(DEFAULT_ROW_COMMIT_VERSION_NAME, DataType::LONG),
    ])
    .leaves(None)
});

/// Translates a Delta file write-metadata batch into a content-tree entry [`EngineData`] batch
/// (one `Data` entry per input row), suitable for serializing as an AMT root manifest.
///
/// Each input row is a newly added file, producing an [`TrackingStatus::Added`] `Data` entry
/// (`fileFormat` parquet, `formatVersion` 4, `specId` 0 -- unpartitioned tables only). Both
/// `sequenceNumber` and `fileSequenceNumber` take the input `defaultRowCommitVersion`; remaining
/// entry fields are null, and statistics and partition values are omitted from the output schema.
///
/// This path is only valid for AMT tables, which always have row tracking enabled, so every input
/// row carries an assigned `baseRowId` and `defaultRowCommitVersion`.
///
/// # Parameters
/// - `engine`: provides the [`crate::EvaluationHandler`] used to evaluate the transform.
/// - `write_metadata`: input batch matching [`write_metadata_input_schema`].
/// - `snapshot_id`: the AMT snapshot id the files are added in; stored in each entry's tracking.
/// - `partition_columns`: the table's partition columns; must be empty (partitioned tables are
///   rejected).
///
/// # Returns
/// An [`EngineData`] batch matching [`ContentTreeNodeEntry::to_schema`].
///
/// # Errors
/// Returns an error if `partition_columns` is non-empty, if a row's required `stats.numRecords`,
/// `baseRowId`, or `defaultRowCommitVersion` is null, or if the evaluator cannot be constructed or
/// fails to evaluate.
pub(crate) fn convert_append_metadata_to_root_entry_batch(
    engine: &dyn Engine,
    write_metadata: &dyn EngineData,
    snapshot_id: i64,
    partition_columns: &[String],
) -> KernelResult<Box<dyn EngineData>> {
    // TODO(#2866): This method doesn't allow for caching standard checks and expressions.
    // we should have some sort of struct wrapper to avoid reconstruction.

    // `specId` 0 below is only correct for an unpartitioned table (a partitioned table's spec 0 is
    // its real partition spec), so reject partitioned tables until this path supports them.
    if !partition_columns.is_empty() {
        return Err(KernelError::unsupported(format!(
            "AMT root writes do not yet support partitioned tables; partition columns: \
             {partition_columns:?}"
        )));
    }

    // Row tracking guarantees these are assigned, but the evaluator does not enforce the input
    // schema's non-nullability, so a missing assignment would otherwise emit a root entry with a
    // null sequence/firstRowId (invalid for a root manifest). Reject it up front.
    let mut validator = RequiredFieldsNonNullVisitor;
    validator.visit_rows_of(write_metadata)?;

    let output_schema = ContentTreeNodeEntry::to_schema();

    let projections = ContentTreeEntryProjections {
        status: TrackingStatus::Added,
        snapshot_id,
        // TODO(#3319): AMT `location` (Iceberg field id 100) is expected to be the
        // percent-decoded data-file path, but `path` is the raw RFC-2396-encoded `AddFile.path`.
        // A decode step is needed once kernel has an expression-level percent-decode op.
        location: Expression::column([PATH_NAME]),
        file_size_in_bytes: Expression::column([SIZE_NAME]),
        sequence_number: Expression::column([DEFAULT_ROW_COMMIT_VERSION_NAME]),
        record_count: Expression::column([STATS_NAME, NUM_RECORDS]),
        first_row_id: Expression::column([BASE_ROW_ID_NAME]),
    };

    let expr = build_content_tree_entry_expression(&output_schema, &projections)?;
    let evaluator = engine.evaluation_handler().new_expression_evaluator(
        write_metadata_input_schema()?,
        Arc::new(expr),
        DataType::from(output_schema),
    )?;
    evaluator.evaluate(write_metadata)
}

// === Helpers ===

/// Rejects any write-metadata row whose required row-tracking/statistic fields are null.
struct RequiredFieldsNonNullVisitor;

impl RowVisitor for RequiredFieldsNonNullVisitor {
    fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]) {
        REQUIRED_NON_NULL_COLUMNS.as_ref()
    }

    fn visit<'a>(&mut self, row_count: usize, getters: &[&'a dyn GetData<'a>]) -> KernelResult<()> {
        // Column order matches REQUIRED_NON_NULL_COLUMNS.
        for row in 0..row_count {
            require_non_null(getters[0], row, STATS_NUM_RECORDS)?;
            require_non_null(getters[1], row, BASE_ROW_ID_NAME)?;
            require_non_null(getters[2], row, DEFAULT_ROW_COMMIT_VERSION_NAME)?;
        }
        Ok(())
    }
}

/// Errors if `getter` holds a null at `row`, naming `field` in the message.
fn require_non_null<'a>(getter: &'a dyn GetData<'a>, row: usize, field: &str) -> KernelResult<()> {
    let value: Option<i64> = getter.get_opt(row, field)?;
    match value {
        Some(_) => Ok(()),
        None => Err(KernelError::missing_data(format!(
            "AMT content-tree write metadata has a null required field '{field}'"
        ))),
    }
}

/// Per-field expressions driving [`build_content_tree_entry_expression`].
struct ContentTreeEntryProjections {
    status: TrackingStatus,
    // TODO(#3321): root-manifest entries always carry an explicit `snapshotId` and
    // `sequenceNumber`. Leaf entries may instead inherit both from their manifest (a null value
    // meaning "inherit"), so this becomes optional once leaf writing is supported.
    snapshot_id: i64,
    location: Expression,
    file_size_in_bytes: Expression,
    /// The `defaultRowCommitVersion` of the added files, used for both `sequenceNumber` and
    /// `fileSequenceNumber`.
    sequence_number: Expression,
    record_count: Expression,
    /// The `baseRowId` of the added files, used for `firstRowId`.
    first_row_id: Expression,
}

/// Builds the expression mapping a write-metadata row to a [`ContentTreeNodeEntry`]-shaped struct.
fn build_content_tree_entry_expression(
    output_schema: &StructType,
    projections: &ContentTreeEntryProjections,
) -> KernelResult<Expression> {
    let tracking = build_tracking_expression(projections)?;
    struct_expr_from_schema(output_schema, |name| match name {
        CONTENT_TYPE => Some(lit(DataContentType::Data)),
        LOCATION => Some(projections.location.clone()),
        FILE_FORMAT => Some(lit(DataFileFormat::Parquet)),
        TRACKING => Some(tracking.clone()),
        // `specId` 0 is safe here because the caller rejects partitioned tables. TODO(#3320): carry
        // the table's actual spec id once the write path supports partitioned AMT tables.
        PARTITION_SPEC_ID => Some(lit(0i32)),
        RECORD_COUNT => Some(projections.record_count.clone()),
        FILE_SIZE_IN_BYTES => Some(projections.file_size_in_bytes.clone()),
        FORMAT_VERSION => Some(lit(AMT_FORMAT_VERSION)),
        _ => None,
    })
}

/// Builds the `tracking` sub-struct for an added `Data` entry.
fn build_tracking_expression(
    projections: &ContentTreeEntryProjections,
) -> KernelResult<Expression> {
    struct_expr_from_schema(&TrackingInfo::to_schema(), |name| match name {
        TRACKING_STATUS => Some(lit(projections.status)),
        TRACKING_SNAPSHOT_ID => Some(lit(projections.snapshot_id)),
        SEQUENCE_NUMBER | FILE_SEQUENCE_NUMBER => Some(projections.sequence_number.clone()),
        FIRST_ROW_ID => Some(projections.first_row_id.clone()),
        _ => None,
    })
}

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

    use super::*;
    use crate::engine::arrow_conversion::TryIntoArrow as _;
    use crate::engine::arrow_data::EngineDataArrowExt as _;
    use crate::engine::sync::SyncEngine;
    use crate::expressions::{Scalar, StructData};
    use crate::Engine;

    /// A row of write-metadata input, where any field may be null (to exercise null rejection).
    #[derive(Clone, Copy)]
    struct InputRow {
        path: Option<&'static str>,
        size: Option<i64>,
        num_records: Option<i64>,
        base_row_id: Option<i64>,
        commit_version: Option<i64>,
    }

    impl From<(&'static str, i64, i64, i64, i64)> for InputRow {
        fn from(
            (path, size, num_records, base_row_id, commit_version): (
                &'static str,
                i64,
                i64,
                i64,
                i64,
            ),
        ) -> Self {
            InputRow {
                path: Some(path),
                size: Some(size),
                num_records: Some(num_records),
                base_row_id: Some(base_row_id),
                commit_version: Some(commit_version),
            }
        }
    }

    fn opt_long(value: Option<i64>) -> Scalar {
        value
            .map(Scalar::Long)
            .unwrap_or(Scalar::Null(DataType::LONG))
    }

    /// Builds a write-metadata input batch from fully-populated `(path, size, num_records,
    /// base_row_id, commit_version)` tuples.
    fn write_metadata_input(
        engine: &dyn Engine,
        files: &[(&'static str, i64, i64, i64, i64)],
    ) -> Box<dyn EngineData> {
        let rows: Vec<InputRow> = files.iter().map(|&f| f.into()).collect();
        write_metadata_input_nullable(engine, &rows)
    }

    /// Builds a write-metadata input batch where any field may be null.
    fn write_metadata_input_nullable(
        engine: &dyn Engine,
        rows: &[InputRow],
    ) -> Box<dyn EngineData> {
        let scalars: Vec<Vec<Scalar>> = rows
            .iter()
            .map(
                |&InputRow {
                     path,
                     size,
                     num_records,
                     base_row_id,
                     commit_version,
                 }| {
                    let stats = StructData::try_new(
                        vec![StructField::nullable(NUM_RECORDS, DataType::LONG)],
                        vec![opt_long(num_records)],
                    )
                    .unwrap();
                    vec![
                        path.map(Scalar::from)
                            .unwrap_or(Scalar::Null(DataType::STRING)),
                        opt_long(size),
                        Scalar::Struct(stats),
                        opt_long(base_row_id),
                        opt_long(commit_version),
                    ]
                },
            )
            .collect();
        engine
            .evaluation_handler()
            .create_many(write_metadata_input_schema().unwrap(), scalars)
            .unwrap()
    }

    /// Builds the expected content-tree entry batch for `files` from explicit
    /// [`ContentTreeNodeEntry`] values.
    fn expected_entries(
        engine: &dyn Engine,
        files: &[(&'static str, i64, i64, i64, i64)],
        snapshot_id: i64,
    ) -> Box<dyn EngineData> {
        let entries: Vec<StructData> = files
            .iter()
            .map(|&(path, size, num_records, base_row_id, commit_version)| {
                expected_entry(
                    path,
                    size,
                    num_records,
                    base_row_id,
                    commit_version,
                    snapshot_id,
                )
                .into()
            })
            .collect();
        let rows: Vec<Vec<Scalar>> = entries.iter().map(|e| e.values().to_vec()).collect();
        engine
            .evaluation_handler()
            .create_many(Arc::new(ContentTreeNodeEntry::to_schema()), rows)
            .unwrap()
    }

    /// The expected `Added` `Data` entry for one input file.
    fn expected_entry(
        path: &str,
        size: i64,
        num_records: i64,
        base_row_id: i64,
        commit_version: i64,
        snapshot_id: i64,
    ) -> ContentTreeNodeEntry {
        ContentTreeNodeEntry {
            content_type: DataContentType::Data,
            location: path.to_string(),
            file_format: DataFileFormat::Parquet,
            tracking: TrackingInfo {
                status: TrackingStatus::Added,
                snapshot_id: Some(snapshot_id),
                dv_snapshot_id: None,
                sequence_number: Some(commit_version),
                file_sequence_number: Some(commit_version),
                first_row_id: Some(base_row_id),
                deleted_positions: None,
                replaced_positions: None,
            },
            deletion_vector: None,
            spec_id: Some(0),
            partition: None,
            sort_order_id: None,
            record_count: num_records,
            file_size_in_bytes: size,
            content_stats: None,
            manifest_info: None,
            key_metadata: None,
            split_offsets: None,
            equality_ids: None,
            format_version: AMT_FORMAT_VERSION,
            tags: None,
        }
    }

    #[test]
    fn convert_append_metadata_to_root_entry_batch_produces_added_data_entries() {
        let engine = SyncEngine::new();
        // (path, size, numRecords, baseRowId, defaultRowCommitVersion)
        let files = [("a.parquet", 100, 10, 0, 5), ("b.parquet", 200, 20, 10, 5)];
        let snapshot_id = 42;

        let out = convert_append_metadata_to_root_entry_batch(
            &engine,
            write_metadata_input(&engine, &files).as_ref(),
            snapshot_id,
            &[],
        )
        .unwrap();
        let expected = expected_entries(&engine, &files, snapshot_id);

        assert_eq!(
            out.try_into_record_batch().unwrap(),
            expected.try_into_record_batch().unwrap()
        );
    }

    #[rstest]
    #[case::spaced("a b.parquet")]
    #[case::percent_encoded("a%20b.parquet")]
    fn convert_append_metadata_to_root_entry_batch_location_is_verbatim(
        #[case] path: &'static str,
    ) {
        // Pins the current contract: `location` carries the raw path byte-for-byte.
        let engine = SyncEngine::new();
        let files = [(path, 1, 1, 0, 0)];
        let out = convert_append_metadata_to_root_entry_batch(
            &engine,
            write_metadata_input(&engine, &files).as_ref(),
            0,
            &[],
        )
        .unwrap();
        assert_eq!(
            out.try_into_record_batch().unwrap(),
            expected_entries(&engine, &files, 0)
                .try_into_record_batch()
                .unwrap()
        );
    }

    #[rstest]
    #[case::num_records(
        vec![InputRow { path: Some("a.parquet"), size: Some(1), num_records: None, base_row_id: Some(0), commit_version: Some(0) }],
        "stats.numRecords"
    )]
    #[case::base_row_id(
        vec![InputRow { path: Some("a.parquet"), size: Some(1), num_records: Some(1), base_row_id: None, commit_version: Some(0) }],
        BASE_ROW_ID_NAME
    )]
    #[case::commit_version(
        vec![InputRow { path: Some("a.parquet"), size: Some(1), num_records: Some(1), base_row_id: Some(0), commit_version: None }],
        DEFAULT_ROW_COMMIT_VERSION_NAME
    )]
    // A null in a non-zero row must still be rejected: guards against a validator that only
    // inspects row 0.
    #[case::null_in_second_row(
        vec![
            InputRow { path: Some("a.parquet"), size: Some(1), num_records: Some(1), base_row_id: Some(0), commit_version: Some(0) },
            InputRow { path: Some("b.parquet"), size: Some(1), num_records: Some(1), base_row_id: None, commit_version: Some(0) },
        ],
        BASE_ROW_ID_NAME
    )]
    fn convert_append_metadata_to_root_entry_batch_rejects_null_required_field(
        #[case] rows: Vec<InputRow>,
        #[case] field: &str,
    ) {
        let engine = SyncEngine::new();
        let input = write_metadata_input_nullable(&engine, &rows);
        let err = convert_append_metadata_to_root_entry_batch(&engine, input.as_ref(), 0, &[])
            .err()
            .expect("null required field should be rejected");
        assert!(
            err.to_string().contains(field),
            "expected error to name {field:?}, got: {err}"
        );
    }

    #[test]
    fn convert_append_metadata_to_root_entry_batch_rejects_partitioned_table() {
        let engine = SyncEngine::new();
        let input = write_metadata_input(&engine, &[("a.parquet", 1, 1, 0, 0)]);
        let partition_columns = ["part_col".to_string()];
        let err = convert_append_metadata_to_root_entry_batch(
            &engine,
            input.as_ref(),
            0,
            &partition_columns,
        )
        .err()
        .expect("a partitioned table should be rejected");
        assert!(
            err.to_string().contains("part_col"),
            "expected error to name the partition column, got: {err}"
        );
    }

    #[test]
    fn write_metadata_output_schema_matches_entry_schema() {
        let engine = SyncEngine::new();
        let input = write_metadata_input(&engine, &[("a.parquet", 1, 1, 0, 0)]);
        let out = convert_append_metadata_to_root_entry_batch(&engine, input.as_ref(), 0, &[])
            .unwrap()
            .try_into_record_batch()
            .unwrap();

        let expected = (&ContentTreeNodeEntry::to_schema())
            .try_into_arrow()
            .unwrap();
        assert_eq!(out.schema().as_ref(), &expected);
    }

    #[test]
    fn convert_append_metadata_to_root_entry_batch_empty_input_yields_empty_batch() {
        let engine = SyncEngine::new();
        let input = write_metadata_input(&engine, &[]);
        let out =
            convert_append_metadata_to_root_entry_batch(&engine, input.as_ref(), 0, &[]).unwrap();
        assert_eq!(out.len(), 0);
    }
}