delta_kernel 0.30.1

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

use super::{Imperative, Transaction};
use crate::actions::{CommitInfo, COMMIT_INFO_NAME, LOG_COMMIT_INFO_SCHEMA};
use crate::engine_data::{GetData, MapItem, RowVisitor, TypedGetData as _};
use crate::expressions::{lit, null_lit, MapData, Scalar};
use crate::schema::{column_name, schema_ref, ColumnName, MapType, ToSchema};
use crate::struct_patch::ProjectionStructPatchBuilder;
use crate::utils::require;
use crate::{
    create_row, DataType, Engine, EngineData, Expression, ExpressionRef, KernelError, KernelResult,
};

pub(super) fn collect_operation_metadata<I, K, V>(entries: I) -> HashMap<String, Option<String>>
where
    I: IntoIterator<Item = (K, Option<V>)>,
    K: Into<String>,
    V: Into<String>,
{
    entries
        .into_iter()
        .map(|(key, value)| (key.into(), value.map(Into::into)))
        .collect()
}

/// Builds a list of `(field_name, literal_expression)` pairs covering every [`CommitInfo`]
/// field. Field names match the camelCase schema names produced by the `ToSchema` derive macro.
/// The returned vec preserves CommitInfo schema field order, which callers rely on when
/// inserting kernel-only fields after the last engine field.
fn commit_info_literal_exprs(
    commit_info: CommitInfo,
) -> KernelResult<Vec<(&'static str, ExpressionRef)>> {
    let string_map_type = MapType::new(DataType::STRING, DataType::STRING, true);
    #[cfg_attr(not(feature = "adaptive-metadata-in-dev"), allow(unused_mut))]
    let mut literal_exprs = vec![
        ("timestamp", Arc::new(lit(commit_info.timestamp))),
        (
            "inCommitTimestamp",
            Arc::new(lit(commit_info.in_commit_timestamp)),
        ),
        ("operation", Arc::new(lit(commit_info.operation))),
        (
            "operationParameters",
            string_map_literal_expr(commit_info.operation_parameters, &string_map_type)?,
        ),
        (
            "operationMetrics",
            string_map_literal_expr(commit_info.operation_metrics, &string_map_type)?,
        ),
        ("kernelVersion", Arc::new(lit(commit_info.kernel_version))),
        ("isBlindAppend", Arc::new(lit(commit_info.is_blind_append))),
        ("engineInfo", Arc::new(lit(commit_info.engine_info))),
        ("txnId", Arc::new(lit(commit_info.txn_id))),
        (
            "tags",
            string_map_literal_expr(commit_info.tags, &string_map_type)?,
        ),
    ];
    #[cfg(feature = "adaptive-metadata-in-dev")]
    literal_exprs.extend([
        (
            "dataChange",
            Arc::new(lit(commit_info.data_change)) as ExpressionRef,
        ),
        (
            "lastManifestCommit",
            Arc::new(lit(commit_info.last_manifest_commit)),
        ),
    ]);
    let expected_expr_len = CommitInfo::to_schema().fields().len();
    if literal_exprs.len() != expected_expr_len {
        return Err(KernelError::Generic(format!("expect the commit_info_literal_exprs return {expected_expr_len} expressions, but only get {} expressions. \
            If CommitInfo field was added/removed, please update Expression::Literal in this function and update the with_commit_info doc comment", literal_exprs.len())));
    }
    Ok(literal_exprs)
}

fn string_map_literal_expr(
    map: Option<HashMap<String, Option<String>>>,
    map_type: &MapType,
) -> KernelResult<ExpressionRef> {
    let expression = match map {
        Some(map) => lit(MapData::try_new(
            map_type.clone(),
            map.into_iter()
                .map(|(key, value)| (Scalar::String(key), value)),
        )?),
        None => null_lit(map_type.clone()),
    };
    Ok(Arc::new(expression))
}

impl<STATE> Transaction<STATE, Imperative> {
    pub(super) fn generate_commit_info(
        &self,
        engine: &dyn Engine,
        kernel_commit_info: CommitInfo,
    ) -> KernelResult<Box<dyn EngineData>> {
        match &self.engine_commit_info {
            Some((engine_commit_info, engine_commit_info_schema)) => {
                require!(
                    engine_commit_info.len() == 1,
                    KernelError::invalid_transaction_state(
                        "Connector commit info must contain exactly one row",
                    )
                );
                let kernel_schema = CommitInfo::to_schema();
                let mut commit_info = kernel_commit_info;
                if engine_commit_info_schema.contains("tags") {
                    let mut visitor = CommitInfoTagsVisitor::default();
                    visitor.visit_rows_of(engine_commit_info.as_ref())?;
                    commit_info.merge_tags(visitor.tags);
                }

                // Step 1: Build literal expressions for each CommitInfo field.
                let literal_exprs = commit_info_literal_exprs(commit_info)?;

                // Step 2: Build the output schema and expression patch together. Engine fields
                // pass through first, overlapping kernel fields are replaced in place, and
                // kernel-only fields are appended after the last engine field.
                let mut patch = ProjectionStructPatchBuilder::new(engine_commit_info_schema);
                for (field_name, expr_ref) in &literal_exprs {
                    let field = kernel_schema.field(*field_name).ok_or_else(|| {
                        KernelError::internal_error(format!(
                            "CommitInfo schema is missing field '{field_name}'"
                        ))
                    })?;
                    if engine_commit_info_schema.contains(*field_name) {
                        patch = patch.replace(*field_name, field.clone(), expr_ref.clone());
                    }
                }
                for (field_name, expr_ref) in &literal_exprs {
                    let field = kernel_schema.field(*field_name).ok_or_else(|| {
                        KernelError::internal_error(format!(
                            "CommitInfo schema is missing field '{field_name}'"
                        ))
                    })?;
                    if !engine_commit_info_schema.contains(*field_name) {
                        patch = patch.append(field.clone(), expr_ref.clone());
                    }
                }
                let (output_schema, patch) = patch.build()?;

                // Step 3: Wrap the patch in a struct expression so the output matches the
                // Delta log action format `{ "commitInfo": { merged fields... } }`, consistent
                // with the None branch which uses `LOG_COMMIT_INFO_SCHEMA`.
                let wrapped_expr = Expression::struct_from([patch]);
                let wrapped_schema = schema_ref! { nullable COMMIT_INFO_NAME: (output_schema) };
                let evaluator = engine.evaluation_handler().new_expression_evaluator(
                    engine_commit_info_schema.clone(),
                    Arc::new(wrapped_expr),
                    wrapped_schema.into(),
                )?;
                evaluator.evaluate(engine_commit_info.as_ref())
            }
            None => create_row(engine, LOG_COMMIT_INFO_SCHEMA.clone(), kernel_commit_info),
        }
    }
}

#[derive(Default)]
struct CommitInfoTagsVisitor {
    tags: Option<HashMap<String, Option<String>>>,
}

impl RowVisitor for CommitInfoTagsVisitor {
    fn selected_column_names_and_types(&self) -> (&'static [ColumnName], &'static [DataType]) {
        static NAMES: LazyLock<[ColumnName; 1]> = LazyLock::new(|| [column_name!("tags")]);
        static TYPES: LazyLock<[DataType; 1]> = LazyLock::new(|| {
            [DataType::from(MapType::new(
                DataType::STRING,
                DataType::STRING,
                true,
            ))]
        });
        (NAMES.as_slice(), TYPES.as_slice())
    }

    fn visit<'a>(
        &mut self,
        row_count: usize,
        getters: &[&'a dyn GetData<'a>],
    ) -> Result<(), KernelError> {
        require!(
            row_count == 1,
            KernelError::invalid_transaction_state(
                "Connector commit info must contain exactly one row"
            )
        );
        let [tags_getter] = getters else {
            return Err(KernelError::internal_error(format!(
                "CommitInfoTagsVisitor received {} getters instead of one",
                getters.len()
            )));
        };
        let tags: Option<MapItem<'_>> = tags_getter.get_opt(0, "tags")?;
        self.tags = tags.map(|tags| {
            tags.keys()
                .map(|key| (key.to_string(), tags.get(key).map(str::to_string)))
                .collect()
        });
        Ok(())
    }
}

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

    use super::CommitInfoTagsVisitor;
    use crate::actions::CommitInfo;
    use crate::arrow::array::{
        Array, ArrayRef, AsArray, BooleanArray, Int64Array, MapArray, MapBuilder,
        RecordBatchOptions, StringArray, StringBuilder, StructArray,
    };
    use crate::arrow::datatypes::{
        DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema,
    };
    use crate::arrow::record_batch::RecordBatch;
    use crate::committer::FileSystemCommitter;
    use crate::engine::arrow_conversion::TryIntoKernel;
    use crate::engine::arrow_data::ArrowEngineData;
    use crate::schema::{schema_ref, Schema, SchemaRef, ToSchema};
    use crate::transaction::Transaction;
    use crate::unit_test_utils::{assert_result_error_with_message, load_test_table};
    use crate::utils::FoldWithOption as _;
    use crate::{Engine, EngineData, KernelResult, RowVisitor};

    // ── build_commit_info tests ────────────────────────────────────────────────

    /// Helper: create a kernel `CommitInfo` that mirrors what `Transaction::commit` produces.
    fn make_kernel_commit_info() -> CommitInfo {
        CommitInfo::new(
            1_700_000_000_000i64,
            Some(134_000_000i64),
            Some("WRITE".to_string()),
            Some("test_engine/1.0".to_string()),
            false,
        )
    }

    /// Helper: build an Arrow RecordBatch + kernel SchemaRef for use as engine_commit_info.
    fn make_engine_commit_info(
        arrow_fields: Vec<ArrowField>,
        columns: Vec<ArrayRef>,
    ) -> (Box<dyn EngineData>, SchemaRef) {
        let arrow_schema = ArrowSchema::new(arrow_fields);
        let kernel_schema: Schema = arrow_schema.as_ref().try_into_kernel().unwrap();
        let batch =
            RecordBatch::try_new(Arc::new(arrow_schema), columns).expect("valid RecordBatch");
        (
            Box::new(ArrowEngineData::new(batch)),
            Arc::new(kernel_schema),
        )
    }

    /// Helper: extract the inner "commitInfo" StructArray from a top-level RecordBatch.
    /// Both branches of `build_commit_info` produce `{ "commitInfo": { ... } }`.
    fn commit_info_struct(result: &ArrowEngineData) -> &StructArray {
        let batch = result.record_batch();
        assert_eq!(
            batch.num_columns(),
            1,
            "expected single 'commitInfo' column"
        );
        assert_eq!(batch.schema().field(0).name(), "commitInfo");
        batch
            .column(0)
            .as_any()
            .downcast_ref::<StructArray>()
            .expect("commitInfo column should be a StructArray")
    }

    /// Helper: pull a non-null string value from a named column in a StructArray.
    fn get_str<'a>(s: &'a StructArray, col: &str) -> &'a str {
        s.column_by_name(col)
            .unwrap_or_else(|| panic!("field '{col}' not found"))
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap_or_else(|| panic!("field '{col}' is not a StringArray"))
            .value(0)
    }

    /// Helper: pull a non-null i64 value from a named column in a StructArray.
    fn get_i64(s: &StructArray, col: &str) -> i64 {
        s.column_by_name(col)
            .unwrap_or_else(|| panic!("field '{col}' not found"))
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap_or_else(|| panic!("field '{col}' is not an Int64Array"))
            .value(0)
    }

    /// Helper: pull the map value at row 0 from a named MapArray column in a StructArray.
    /// Returns the key-value pairs as a StructArray.
    fn get_map(s: &StructArray, col: &str) -> StructArray {
        s.column_by_name(col)
            .unwrap_or_else(|| panic!("field '{col}' not found"))
            .as_any()
            .downcast_ref::<MapArray>()
            .unwrap_or_else(|| panic!("field '{col}' is not a MapArray"))
            .value(0)
    }

    /// Helper: pull a non-null boolean value from a named column in a StructArray.
    fn get_bool(s: &StructArray, col: &str) -> bool {
        s.column_by_name(col)
            .unwrap_or_else(|| panic!("field '{col}' not found"))
            .as_any()
            .downcast_ref::<BooleanArray>()
            .unwrap_or_else(|| panic!("field '{col}' is not an Int64Array"))
            .value(0)
    }

    /// Create a transaction with the given engine_commit_info, using the shared test table.
    fn make_txn(
        engine_commit_info: Option<(Box<dyn EngineData>, SchemaRef)>,
    ) -> KernelResult<(Arc<dyn Engine>, Transaction)> {
        let (engine, snapshot, _tempdir) = load_test_table("table-without-dv-small")?;
        let txn = snapshot
            .update_table(
                crate::transaction::UpdateTableOperation::Write,
                "test-engine",
                true,
            )
            .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?
            .fold_with(engine_commit_info, |txn, (data, schema)| {
                txn.with_commit_info(data, schema)
            });
        Ok((engine, txn))
    }

    #[test]
    fn commit_info_tags_visitor_rejects_invalid_callback_shape() {
        let mut visitor = CommitInfoTagsVisitor::default();
        assert_result_error_with_message(
            visitor.visit(0, &[]),
            "Connector commit info must contain exactly one row",
        );
        assert_result_error_with_message(
            visitor.visit(1, &[]),
            "CommitInfoTagsVisitor received 0 getters instead of one",
        );
    }

    /// no engine_commit_info -- output is the kernel CommitInfo wrapped in a "commitInfo"
    /// outer struct, matching the Delta log action format produced by `LOG_COMMIT_INFO_SCHEMA`.
    #[test]
    fn test_build_commit_info_none_branch() -> KernelResult<()> {
        let (engine, txn) = make_txn(None)?;
        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), make_kernel_commit_info())?,
        )?;
        let ci = commit_info_struct(&result);

        let kernel_schema = CommitInfo::to_schema();
        assert_eq!(ci.num_columns(), kernel_schema.fields().count());
        assert_eq!(get_str(ci, "operation"), "WRITE");
        assert!(!get_str(ci, "kernelVersion").is_empty());
        assert!(!get_str(ci, "txnId").is_empty());
        Ok(())
    }

    /// engine schema has fields that are fully disjoint from CommitInfo -- all CommitInfo
    /// fields are appended after the engine-only fields, in CommitInfo schema order.
    #[test]
    fn test_build_commit_info_disjoint_schemas() -> KernelResult<()> {
        let (data, schema) = make_engine_commit_info(
            vec![
                ArrowField::new("customApp", ArrowDataType::Utf8, false),
                ArrowField::new("customVersion", ArrowDataType::Int64, false),
            ],
            vec![
                Arc::new(StringArray::from(vec!["myApp"])) as ArrayRef,
                Arc::new(Int64Array::from(vec![42i64])) as ArrayRef,
            ],
        );
        let (engine, txn) = make_txn(Some((data, schema)))?;

        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), make_kernel_commit_info())?,
        )?;
        let commit_info = commit_info_struct(&result);

        // All CommitInfo fields are appended after the two engine-only fields.
        assert_eq!(
            commit_info.num_columns(),
            2 + CommitInfo::to_schema().fields().count()
        );

        // Engine fields are first and their values pass through unchanged.
        assert_eq!(commit_info.fields()[0].name(), "customApp");
        assert_eq!(commit_info.fields()[1].name(), "customVersion");
        assert_eq!(get_str(commit_info, "customApp"), "myApp");
        assert_eq!(get_i64(commit_info, "customVersion"), 42);

        assert_eq!(get_str(commit_info, "operation"), "WRITE");
        assert!(!get_str(commit_info, "kernelVersion").is_empty());
        assert!(get_map(commit_info, "operationParameters").len() == 0);
        assert!(get_map(commit_info, "operationMetrics").is_empty());
        assert!(uuid::Uuid::parse_str(get_str(commit_info, "txnId")).is_ok());
        assert!(get_i64(commit_info, "timestamp") > 0);
        assert_eq!(get_i64(commit_info, "inCommitTimestamp"), 134_000_000);
        assert_eq!(get_str(commit_info, "engineInfo"), "test_engine/1.0");
        assert!(!get_bool(commit_info, "isBlindAppend"));

        Ok(())
    }

    /// engine schema contains every kernel's CommitInfo field.
    /// All overlapping fields must be replaced by kernel values, no new fields added.
    #[test]
    fn test_build_commit_info_full_overlap() -> KernelResult<()> {
        let mut map_builder = MapBuilder::new(None, StringBuilder::new(), StringBuilder::new());
        map_builder.keys().append_value("stale_key");
        map_builder.values().append_value("stale_value");
        map_builder.append(true).unwrap();
        let stale_op_params = Arc::new(map_builder.finish()) as ArrayRef;
        let connector_tags = stale_op_params.clone();
        let mut metrics_map_builder =
            MapBuilder::new(None, StringBuilder::new(), StringBuilder::new());
        metrics_map_builder.keys().append_value("stale_metric");
        metrics_map_builder.values().append_value("1");
        metrics_map_builder.append(true).unwrap();
        let stale_op_metrics = Arc::new(metrics_map_builder.finish()) as ArrayRef;

        let (data, schema) = make_engine_commit_info(
            vec![
                ArrowField::new("timestamp", ArrowDataType::Int64, true),
                ArrowField::new("inCommitTimestamp", ArrowDataType::Int64, true),
                ArrowField::new("operation", ArrowDataType::Utf8, true),
                ArrowField::new(
                    "operationParameters",
                    stale_op_params.data_type().clone(),
                    true,
                ),
                ArrowField::new(
                    "operationMetrics",
                    stale_op_metrics.data_type().clone(),
                    true,
                ),
                ArrowField::new("kernelVersion", ArrowDataType::Utf8, true),
                ArrowField::new("isBlindAppend", ArrowDataType::Boolean, true),
                ArrowField::new("engineInfo", ArrowDataType::Utf8, true),
                ArrowField::new("txnId", ArrowDataType::Utf8, true),
                ArrowField::new("tags", connector_tags.data_type().clone(), true),
            ],
            vec![
                Arc::new(Int64Array::from(vec![Some(0i64)])) as ArrayRef,
                Arc::new(Int64Array::from(vec![None::<i64>])) as ArrayRef,
                Arc::new(StringArray::from(vec!["STALE_OP"])) as ArrayRef,
                stale_op_params,
                stale_op_metrics,
                Arc::new(StringArray::from(vec!["v0.0.0"])) as ArrayRef,
                Arc::new(BooleanArray::from(vec![None::<bool>])) as ArrayRef,
                Arc::new(StringArray::from(vec!["stale_engine"])) as ArrayRef,
                Arc::new(StringArray::from(vec!["stale_txn"])) as ArrayRef,
                connector_tags,
            ],
        );
        let (engine, txn) = make_txn(Some((data, schema)))?;
        let mut kernel_commit_info = make_kernel_commit_info();
        kernel_commit_info.operation_parameters = Some(
            [(
                "current_parameter".to_string(),
                Some("current_value".to_string()),
            )]
            .into_iter()
            .collect(),
        );
        kernel_commit_info.operation_metrics = Some(
            [("current_metric".to_string(), Some("1".to_string()))]
                .into_iter()
                .collect(),
        );

        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), kernel_commit_info)?,
        )?;
        let commit_info = commit_info_struct(&result);

        // All CommitInfo fields are present in the engine schema -- no fields appended.
        assert_eq!(
            commit_info.num_columns(),
            CommitInfo::to_schema().fields().count()
        );

        assert_eq!(get_str(commit_info, "operation"), "WRITE");
        assert!(!get_str(commit_info, "kernelVersion").is_empty());
        let parameters = get_map(commit_info, "operationParameters");
        assert_eq!(parameters.len(), 1);
        assert_eq!(get_str(&parameters, "key"), "current_parameter");
        assert_eq!(get_str(&parameters, "value"), "current_value");
        let metrics = get_map(commit_info, "operationMetrics");
        assert_eq!(metrics.len(), 1);
        assert_eq!(get_str(&metrics, "key"), "current_metric");
        assert_eq!(get_str(&metrics, "value"), "1");
        assert!(uuid::Uuid::parse_str(get_str(commit_info, "txnId")).is_ok());
        assert!(get_i64(commit_info, "timestamp") > 0);
        assert_eq!(get_i64(commit_info, "inCommitTimestamp"), 134_000_000);
        assert_eq!(get_str(commit_info, "engineInfo"), "test_engine/1.0");
        assert!(!get_bool(commit_info, "isBlindAppend"));
        let tags = get_map(commit_info, "tags");
        let tag_keys = tags.column(0).as_string::<i32>();
        let tag_values = tags.column(1).as_string::<i32>();
        assert_eq!(tag_keys.value(0), "stale_key");
        assert_eq!(tag_values.value(0), "stale_value");

        Ok(())
    }

    /// engine schema has partial overlap -- overlapping fields are replaced, engine-only
    /// fields pass through, and remaining CommitInfo fields are appended after the last engine
    /// field.
    #[test]
    fn test_build_commit_info_partial_overlap() -> KernelResult<()> {
        let (data, schema) = make_engine_commit_info(
            vec![
                ArrowField::new("timestamp", ArrowDataType::Int64, true),
                ArrowField::new("operation", ArrowDataType::Utf8, true),
                ArrowField::new("myCustomField", ArrowDataType::Utf8, false),
            ],
            vec![
                Arc::new(Int64Array::from(vec![Some(0i64)])) as ArrayRef,
                Arc::new(StringArray::from(vec!["STALE_OP"])) as ArrayRef,
                Arc::new(StringArray::from(vec!["keep_me"])) as ArrayRef,
            ],
        );
        let (engine, txn) = make_txn(Some((data, schema)))?;

        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), make_kernel_commit_info())?,
        )?;
        let ci = commit_info_struct(&result);

        // Engine-only field passes through unchanged.
        assert_eq!(get_str(ci, "myCustomField"), "keep_me");

        // Overlapping fields are replaced with kernel values.
        assert_ne!(get_str(ci, "operation"), "STALE_OP");
        assert_eq!(get_str(ci, "operation"), "WRITE");

        // Engine fields keep their original schema positions (first 3 columns).
        assert_eq!(ci.fields()[0].name(), "timestamp");
        assert_eq!(ci.fields()[1].name(), "operation");
        assert_eq!(ci.fields()[2].name(), "myCustomField");

        // Remaining CommitInfo fields are appended after myCustomField.
        assert_eq!(
            ci.num_columns(),
            3 + CommitInfo::to_schema().fields().count() - 2
        );
        Ok(())
    }

    /// engine schema has overlapping fields with different DataTypes than kernel expects.
    /// Kernel replacement must win, so each output field has the kernel's type.
    #[test]
    fn test_build_commit_info_type_conflict_replaced_by_kernel() -> KernelResult<()> {
        let (data, schema) = make_engine_commit_info(
            vec![
                ArrowField::new("timestamp", ArrowDataType::Utf8, true),
                ArrowField::new("inCommitTimestamp", ArrowDataType::Utf8, true),
                ArrowField::new("operation", ArrowDataType::Int64, true),
                ArrowField::new("isBlindAppend", ArrowDataType::Utf8, true),
                ArrowField::new("myCustomField", ArrowDataType::Utf8, false),
            ],
            vec![
                Arc::new(StringArray::from(vec!["not-a-timestamp"])) as ArrayRef,
                Arc::new(StringArray::from(vec!["not-a-timestamp"])) as ArrayRef,
                Arc::new(Int64Array::from(vec![0i64])) as ArrayRef,
                Arc::new(StringArray::from(vec!["not-a-bool"])) as ArrayRef,
                Arc::new(StringArray::from(vec!["keep_me"])) as ArrayRef,
            ],
        );
        let (engine, txn) = make_txn(Some((data, schema)))?;

        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), make_kernel_commit_info())?,
        )?;
        let ci = commit_info_struct(&result);

        // Each kernel-owned field has the kernel's type, not the engine's.
        let field_type = |name: &str| {
            ci.fields()
                .iter()
                .find(|f| f.name() == name)
                .unwrap_or_else(|| panic!("field '{name}' must be present"))
                .data_type()
                .clone()
        };
        assert_eq!(field_type("timestamp"), ArrowDataType::Int64);
        assert_eq!(field_type("inCommitTimestamp"), ArrowDataType::Int64);
        assert_eq!(field_type("operation"), ArrowDataType::Utf8);
        assert_eq!(field_type("isBlindAppend"), ArrowDataType::Boolean);

        // Engine-only field passes through with its original type and value unchanged.
        assert_eq!(field_type("myCustomField"), ArrowDataType::Utf8);
        assert_eq!(get_str(ci, "myCustomField"), "keep_me");
        Ok(())
    }

    #[rstest::rstest]
    #[case::zero_rows(0, false)]
    #[case::one_row(1, true)]
    #[case::two_rows(2, false)]
    fn test_build_commit_info_empty_engine_schema(
        #[case] row_count: usize,
        #[case] succeeds: bool,
    ) -> KernelResult<()> {
        let empty_batch = RecordBatch::try_new_with_options(
            Arc::new(ArrowSchema::empty()),
            vec![],
            &RecordBatchOptions::new().with_row_count(Some(row_count)),
        )?;
        let empty_schema = schema_ref! {};
        let (engine, txn) = make_txn(Some((
            Box::new(ArrowEngineData::new(empty_batch)),
            empty_schema,
        )))?;

        let result = txn.generate_commit_info(engine.as_ref(), make_kernel_commit_info());
        if !succeeds {
            assert_result_error_with_message(
                result,
                "Connector commit info must contain exactly one row",
            );
            return Ok(());
        }

        let result = ArrowEngineData::try_from_engine_data(result?)?;
        let ci = commit_info_struct(&result);

        let kernel_schema = CommitInfo::to_schema();
        assert_eq!(ci.num_columns(), kernel_schema.fields().count());

        for (i, field) in kernel_schema.fields().enumerate() {
            assert_eq!(ci.fields()[i].name(), field.name());
        }
        Ok(())
    }

    /// Kernel's `dataChange` is emitted (null when unset) and overrides any engine-supplied value.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest::rstest]
    fn test_build_commit_info_data_change(
        #[values(None, Some(true), Some(false))] txn_data_change: Option<bool>,
        #[values(false, true)] engine_conflicting_data_change: bool,
    ) -> KernelResult<()> {
        // Always differs from the txn value so an override is observable.
        let engine_commit_info = engine_conflicting_data_change.then(|| {
            let conflicting_value = !txn_data_change.unwrap_or(false);
            make_engine_commit_info(
                vec![ArrowField::new("dataChange", ArrowDataType::Boolean, true)],
                vec![Arc::new(BooleanArray::from(vec![Some(conflicting_value)])) as ArrayRef],
            )
        });
        let (engine, txn) = make_txn(engine_commit_info)?;
        let mut commit_info = make_kernel_commit_info();
        if let Some(data_change) = txn_data_change {
            commit_info.set_data_change(data_change);
        }

        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), commit_info)?,
        )?;
        let ci = commit_info_struct(&result);
        let column = ci
            .column_by_name("dataChange")
            .expect("dataChange column should be present");
        match txn_data_change {
            None => assert!(column.is_null(0)),
            Some(data_change) => assert_eq!(get_bool(ci, "dataChange"), data_change),
        }
        Ok(())
    }

    /// A `None` `last_manifest_commit` emits a present-but-null `lastManifestCommit` struct column;
    /// a `Some` emits the struct values.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest::rstest]
    #[case::absent(None)]
    #[case::present(Some((5, 3)))]
    fn test_build_commit_info_last_manifest_commit(
        #[case] last_manifest_commit: Option<(i64, i64)>,
    ) {
        use crate::actions::LastManifestCommit;

        let (engine, txn) = make_txn(None).unwrap();
        let mut commit_info = make_kernel_commit_info();
        commit_info.last_manifest_commit =
            last_manifest_commit.map(|(version, content_root_version)| {
                LastManifestCommit::new(version, content_root_version).unwrap()
            });

        let result = ArrowEngineData::try_from_engine_data(
            txn.generate_commit_info(engine.as_ref(), commit_info)
                .unwrap(),
        )
        .unwrap();
        let ci = commit_info_struct(&result);
        let column = ci
            .column_by_name("lastManifestCommit")
            .expect("lastManifestCommit column should be present");

        match last_manifest_commit {
            None => assert!(column.is_null(0)),
            Some((version, content_root_version)) => {
                let lmc = column
                    .as_any()
                    .downcast_ref::<StructArray>()
                    .expect("lastManifestCommit should be a StructArray");
                assert!(lmc.is_valid(0));
                assert_eq!(get_i64(lmc, "version"), version);
                assert_eq!(get_i64(lmc, "contentRootVersion"), content_root_version);
            }
        }
    }
}