delta_kernel 0.27.1

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use itertools::Itertools;
use serde::Serialize;
use tempfile::TempDir;
use test_utils::{
    copy_directory, delta_path_for_version, load_test_data, modify_add_file_partition_keys,
    replace_array_row, AddFilePartitionKeyModify,
};
use tracing::subscriber::DefaultGuard;
use tracing_subscriber::util::SubscriberInitExt as _;
use url::Url;

use crate::actions::{get_all_actions_schema, Add, Cdc, CommitInfo, Metadata, Protocol, Remove};
use crate::arrow::array::{
    new_empty_array, new_null_array, ArrayRef, Int64Array, MapArray, RecordBatch, StringArray,
    StructArray,
};
use crate::arrow::buffer::{OffsetBuffer, ScalarBuffer};
use crate::arrow::compute::concat_batches;
use crate::arrow::datatypes::{DataType, Field, Schema as ArrowSchema};
use crate::committer::FileSystemCommitter;
use crate::engine::arrow_conversion::{parquet_field_id_metadata, TryIntoArrow as _};
use crate::engine::arrow_data::ArrowEngineData;
use crate::engine::sync::SyncEngine;
use crate::metrics::{MetricEvent, MetricsReporter, WithMetricsReporterLayer as _};
use crate::object_store::local::LocalFileSystem;
use crate::object_store::memory::InMemory;
use crate::object_store::ObjectStoreExt as _;
use crate::parquet::arrow::PARQUET_FIELD_ID_META_KEY;
use crate::path::ParsedLogPath;
use crate::table_features::ColumnMappingMode;
use crate::transaction::create_table::create_table;
use crate::transaction::{CreateTable, Transaction, BASE_ADD_FILES_SCHEMA};
use crate::{DeltaResult, Engine, EngineData, Error, FileMeta, Snapshot, SnapshotRef};

/// Parses `path` (a full URL string) into a [`ParsedLogPath`] with zero size, for building
/// synthetic log-file listings in tests.
pub(crate) fn create_log_path(path: &str) -> ParsedLogPath<FileMeta> {
    create_log_path_with_size(path, 0)
}

/// [`create_log_path`] with an explicit file size.
pub(crate) fn create_log_path_with_size(path: &str, size: u64) -> ParsedLogPath<FileMeta> {
    ParsedLogPath::try_from(FileMeta {
        location: Url::parse(path).expect("Invalid file URL"),
        last_modified: 0,
        size,
    })
    .unwrap()
    .unwrap()
}

/// A metrics reporter that captures all events for test assertions.
#[derive(Debug, Default)]
pub(crate) struct CapturingReporter {
    events: Mutex<Vec<MetricEvent>>,
}

impl MetricsReporter for CapturingReporter {
    fn report(&self, event: MetricEvent) {
        self.events.lock().unwrap().push(event);
    }
}

impl CapturingReporter {
    /// Returns a copy of all captured events.
    pub(crate) fn events(&self) -> Vec<MetricEvent> {
        self.events.lock().unwrap().clone()
    }
}

/// Kernel-internal twin of [`test_utils::install_thread_local_metrics_reporter`].
///
/// Internal tests need their own helper because the trait identity of `MetricsReporter`
/// differs across the test_utils <-> kernel path-dep boundary. Both helpers wrap
/// [`test_utils::ensure_metrics_compatible_global_subscriber`] + a thread-local
/// `set_default` and serve the same purpose: install a metrics-collecting subscriber
/// in a way that is robust against tracing callsite-cache poisoning.
pub(crate) fn install_thread_local_metrics_reporter(
    reporter: Arc<dyn MetricsReporter>,
) -> DefaultGuard {
    test_utils::ensure_metrics_compatible_global_subscriber();
    tracing_subscriber::registry()
        .with_metrics_reporter_layer(reporter)
        .set_default()
}

#[derive(Serialize)]
pub(crate) enum Action {
    #[serde(rename = "add")]
    Add(Add),
    #[serde(rename = "remove")]
    Remove(Remove),
    #[serde(rename = "cdc")]
    Cdc(Cdc),
    #[serde(rename = "metaData")]
    Metadata(Metadata),
    #[serde(rename = "protocol")]
    Protocol(Protocol),
    #[allow(unused)]
    #[serde(rename = "commitInfo")]
    CommitInfo(CommitInfo),
}

use crate::schema::{
    schema, schema_ref, ArrayType, ColumnMetadataKey, DataType as KernelDataType, MapType,
    MetadataValue, SchemaRef, StructField, StructType,
};
#[cfg(feature = "geo-type-in-dev")]
use crate::schema::{EdgeInterpolationAlgorithm, GeographyType, GeometryType, PrimitiveType};

/// A mock table that writes commits to a local temporary delta log. This can be used to
/// construct a delta log used for testing.
pub(crate) struct LocalMockTable {
    commit_num: u64,
    store: Arc<LocalFileSystem>,
    dir: TempDir,
}

impl LocalMockTable {
    pub(crate) fn new() -> Self {
        let dir = tempfile::tempdir().unwrap();
        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path()).unwrap());
        Self {
            commit_num: 0,
            store,
            dir,
        }
    }
    /// Writes all `actions` to a new commit in the log
    pub(crate) async fn commit(&mut self, actions: impl IntoIterator<Item = Action>) {
        let data = actions
            .into_iter()
            .map(|action| serde_json::to_string(&action).unwrap())
            .join("\n");

        let path = delta_path_for_version(self.commit_num, "json");
        self.commit_num += 1;

        self.store
            .put(&path, data.into())
            .await
            .expect("put log file in store");
    }

    /// Get the path to the root of the table.
    pub(crate) fn table_root(&self) -> &Path {
        self.dir.path()
    }
}

/// Try to convert an `EngineData` into a `RecordBatch`. Panics if not using `ArrowEngineData`
/// from the default module
fn into_record_batch(engine_data: Box<dyn EngineData>) -> RecordBatch {
    ArrowEngineData::try_from_engine_data(engine_data)
        .unwrap()
        .into()
}

/// Checks that two `EngineData` objects are equal by converting them to `RecordBatch` and
/// comparing
pub(crate) fn assert_batch_matches(actual: Box<dyn EngineData>, expected: Box<dyn EngineData>) {
    assert_eq!(into_record_batch(actual), into_record_batch(expected));
}

/// Helper for building valid add-file batches.
///
/// Tests can generate add-file metadata without writing a Parquet file using this helper. All
/// required fields are populated. When `all_nullable` is true, every field in the batch
/// schema is nullable.
pub(crate) fn create_valid_add_file_batch(all_nullable: bool) -> RecordBatch {
    let schema = if all_nullable {
        let fields = BASE_ADD_FILES_SCHEMA
            .fields()
            .map(|f| StructField::nullable(f.name().clone(), f.data_type().clone()));
        StructType::try_new(fields).expect("nullable add-file schema")
    } else {
        BASE_ADD_FILES_SCHEMA.as_ref().clone()
    };
    let arrow_schema: ArrowSchema = (&schema).try_into_arrow().expect("arrow schema");
    let columns: Vec<ArrayRef> = arrow_schema
        .fields()
        .iter()
        .map(|field| match field.name().as_str() {
            "path" => Arc::new(StringArray::from(vec!["dummy"])) as ArrayRef,
            "size" | "modificationTime" => Arc::new(Int64Array::from(vec![1i64])) as ArrayRef,
            "partitionValues" => {
                let DataType::Map(entries_field, sorted) = field.data_type() else {
                    panic!("partitionValues must be a map type");
                };
                let entries = new_empty_array(entries_field.data_type())
                    .as_any()
                    .downcast_ref::<StructArray>()
                    .expect("map entries struct")
                    .clone();
                // One row, empty partition map.
                let offsets = OffsetBuffer::new(ScalarBuffer::from(vec![0i32, 0]));
                Arc::new(
                    MapArray::try_new(entries_field.clone(), offsets, entries, None, *sorted)
                        .expect("empty map array"),
                )
            }
            // Non-mandatory fields (e.g. `stats`) are left null.
            _ => new_null_array(field.data_type(), 1),
        })
        .collect();
    RecordBatch::try_new(Arc::new(arrow_schema), columns).expect("valid add-file batch")
}

/// Builds one valid add-file row with a fully nullable schema.
pub(crate) fn nullable_add_file() -> RecordBatch {
    create_valid_add_file_batch(true /* all_nullable */)
}

/// Builds `row_count` valid add-file rows with a fully nullable schema.
pub(crate) fn nullable_add_files(row_count: usize) -> RecordBatch {
    let batch = nullable_add_file();
    concat_batches(&batch.schema(), &vec![batch; row_count])
        .expect("failed to concatenate rows into a multi-row add-file batch")
}

pub(crate) fn replace_column(batch: &RecordBatch, field: &str, column: ArrayRef) -> RecordBatch {
    let schema = batch.schema();
    let index = schema.index_of(field).expect("field in schema");
    let mut columns = batch.columns().to_vec();
    columns[index] = column;
    RecordBatch::try_new(schema, columns).expect("failed to rebuild batch after replacing a column")
}

pub(crate) fn set_field_as_null(batch: &RecordBatch, field: &str, row: usize) -> RecordBatch {
    let schema = batch.schema();
    let index = schema.index_of(field).expect("field in schema");
    let mut columns = batch.columns().to_vec();
    let null = new_null_array(schema.field(index).data_type(), 1);
    columns[index] = replace_array_row(&columns[index], null, row);
    RecordBatch::try_new(schema, columns)
        .expect("failed to rebuild batch after replacing a field value with null")
}

/// Returns nullable add-file rows with `partitionValues` replaced by `partition_values`.
pub(crate) fn add_files_with_partition_values(
    partition_values: &[&[(&str, Option<&str>)]],
) -> RecordBatch {
    let batches: Vec<_> = partition_values
        .iter()
        .map(|entries| {
            let modifications: Vec<_> = entries
                .iter()
                .map(|(key, value)| AddFilePartitionKeyModify::Insert { key, value: *value })
                .collect();
            modify_add_file_partition_keys(nullable_add_file(), &modifications)
        })
        .collect();
    concat_batches(&batches[0].schema(), &batches)
        .expect("failed to concatenate rows with partition values")
}

pub(crate) fn string_array_to_engine_data(string_array: StringArray) -> Box<dyn EngineData> {
    let string_field = Arc::new(Field::new("a", DataType::Utf8, true));
    let schema = Arc::new(ArrowSchema::new(vec![string_field]));
    let batch = RecordBatch::try_new(schema, vec![Arc::new(string_array)])
        .expect("Can't convert to record batch");
    Box::new(ArrowEngineData::new(batch))
}

pub(crate) fn parse_json_batch(json_strings: StringArray) -> Box<dyn EngineData> {
    let engine = SyncEngine::new();
    let json_handler = engine.json_handler();
    let output_schema = get_all_actions_schema().clone();
    json_handler
        .parse_json(string_array_to_engine_data(json_strings), output_schema)
        .unwrap()
}

pub(crate) fn action_batch() -> Box<dyn EngineData> {
    let json_strings: StringArray = vec![
        r#"{"add":{"path":"part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet","partitionValues":{},"size":635,"modificationTime":1677811178336,"dataChange":true,"stats":"{\"numRecords\":10,\"minValues\":{\"value\":0},\"maxValues\":{\"value\":9},\"nullCount\":{\"value\":0},\"tightBounds\":true}","tags":{"INSERTION_TIME":"1677811178336000","MIN_INSERTION_TIME":"1677811178336000","MAX_INSERTION_TIME":"1677811178336000","OPTIMIZE_TARGET_SIZE":"268435456"}}}"#,
        r#"{"remove":{"path":"part-00003-f525f459-34f9-46f5-82d6-d42121d883fd.c000.snappy.parquet","deletionTimestamp":1670892998135,"dataChange":true,"partitionValues":{"c1":"4","c2":"c"},"size":452}}"#,
        r#"{"commitInfo":{"timestamp":1677811178585,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isolationLevel":"WriteSerializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"10","numOutputBytes":"635"},"engineInfo":"Databricks-Runtime/<unknown>","txnId":"a6a94671-55ef-450e-9546-b8465b9147de"}}"#,
        r#"{"protocol":{"minReaderVersion":3,"minWriterVersion":7,"readerFeatures":["deletionVectors"],"writerFeatures":["deletionVectors"]}}"#,
        r#"{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"value\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.enableDeletionVectors":"true","delta.columnMapping.mode":"none", "delta.enableChangeDataFeed":"true"},"createdTime":1677811175819}}"#,
        r#"{"cdc":{"path":"_change_data/age=21/cdc-00000-93f7fceb-281a-446a-b221-07b88132d203.c000.snappy.parquet","partitionValues":{"age":"21"},"size":1033,"dataChange":false}}"#,
        r#"{"sidecar":{"path":"016ae953-37a9-438e-8683-9a9a4a79a395.parquet","sizeInBytes":9268,"modificationTime":1714496113961,"tags":{"tag_foo":"tag_bar"}}}"#,
        r#"{"txn":{"appId":"myApp","version": 3}}"#,
        r#"{"checkpointMetadata":{"version":2, "tags":{"tag_foo":"tag_bar"}}}"#,
    ]
    .into();
    parse_json_batch(json_strings)
}

// TODO: allow tests to pass in context (issue#1133)
#[track_caller]
pub(crate) fn assert_result_error_with_message<T, E: ToString>(res: Result<T, E>, message: &str) {
    match res {
        Ok(_) => panic!("Expected error with message {message}, but got Ok result"),
        Err(error) => {
            let error_str = error.to_string();
            assert!(
                error_str.contains(message),
                "Error message does not contain the expected message.\nExpected message:\t{message}\nActual message:\t\t{error_str}"
            );
        }
    }
}

/// Asserts the 2x2 matrix of (schema_has_feature, protocol_supports_feature) outcomes
/// for schema-level feature validators. The expected pattern is:
/// - schema + protocol => Ok
/// - no schema + no protocol => Ok
/// - no schema + protocol => Ok
/// - schema + no protocol => Err (orphaned schema presence)
///
/// Additional error schemas (e.g. nested) are also tested against `protocol_without`.
#[track_caller]
pub(crate) fn assert_schema_feature_validation(
    schema_with: &StructType,
    schema_without: &StructType,
    protocol_with: &Protocol,
    protocol_without: &Protocol,
    extra_err_schemas: &[&StructType],
    err_msg: &str,
) {
    make_test_tc(schema_with.clone(), protocol_with.clone(), [])
        .expect("feature present + supported");
    make_test_tc(schema_without.clone(), protocol_without.clone(), [])
        .expect("feature absent + unsupported");
    make_test_tc(schema_without.clone(), protocol_with.clone(), [])
        .expect("feature absent + supported");
    assert_result_error_with_message(
        make_test_tc(schema_with.clone(), protocol_without.clone(), []),
        err_msg,
    );
    for schema in extra_err_schemas {
        assert_result_error_with_message(
            make_test_tc((*schema).clone(), protocol_without.clone(), []),
            err_msg,
        );
    }
}

/// Creates a [`TableConfiguration`] from a schema, protocol, and table properties.
/// Useful for testing validators that need a TC.
pub(crate) fn make_test_tc(
    schema: StructType,
    protocol: Protocol,
    props: impl IntoIterator<Item = (String, String)>,
) -> crate::DeltaResult<crate::table_configuration::TableConfiguration> {
    let schema = std::sync::Arc::new(schema);
    let metadata =
        Metadata::try_new(None, None, schema, vec![], 0, props.into_iter().collect()).unwrap();
    let table_root = Url::try_from("file:///").unwrap();
    crate::table_configuration::TableConfiguration::try_new(metadata, protocol, table_root, 0)
}

// ==================== Test schema helpers ====================
//
// Reusable test schemas
// Each variant exists with and without column mapping metadata.

/// Builds a nullable [`StructField`] carrying column-mapping id + physical name metadata.
fn cm_field(
    name: &str,
    id: i64,
    physical_name: &str,
    ty: impl Into<KernelDataType>,
) -> StructField {
    StructField::nullable(name, ty).with_metadata([
        (
            ColumnMetadataKey::ColumnMappingId.as_ref(),
            MetadataValue::Number(id),
        ),
        (
            ColumnMetadataKey::ColumnMappingPhysicalName.as_ref(),
            MetadataValue::String(physical_name.into()),
        ),
    ])
}

/// Shared fixture for nested field-id propagation tests.
pub(crate) struct NestedFieldIdFixture {
    pub(crate) kernel_schema: StructType,
    pub(crate) input_arrow_data: StructArray,
    pub(crate) expected_arrow_schema: ArrowSchema,
}

/// Recursively collect `(field_name, metadata_value)` pairs for the given metadata key
/// across all (nested) Arrow fields in `schema`.
pub(crate) fn collect_arrow_field_metadata(
    schema: &ArrowSchema,
    metadata_key: &str,
) -> Vec<(String, String)> {
    fn collect_from_fields(
        fields: &[Arc<Field>],
        metadata_key: &str,
        out: &mut Vec<(String, String)>,
    ) {
        for field in fields {
            collect_from_field(field, metadata_key, out);
        }
    }

    fn collect_from_field(field: &Field, metadata_key: &str, out: &mut Vec<(String, String)>) {
        if let Some(value) = field.metadata().get(metadata_key) {
            out.push((field.name().clone(), value.clone()));
        }

        match field.data_type() {
            DataType::Struct(fields) => collect_from_fields(fields, metadata_key, out),
            DataType::List(entry) | DataType::Map(entry, _) => {
                collect_from_field(entry, metadata_key, out)
            }
            _ => {}
        }
    }

    let mut out = Vec::new();
    collect_from_fields(schema.fields(), metadata_key, &mut out);
    out
}

/// Build the kernel schema for `array_in_map: map<int, array<int>>` with caller-provided
/// top-level field metadata.
pub(crate) fn array_in_map_kernel_schema(
    metadata: impl IntoIterator<Item = (String, MetadataValue)>,
) -> StructType {
    let array_in_map = StructField::nullable(
        "array_in_map",
        MapType::new(
            KernelDataType::INTEGER,
            ArrayType::new(KernelDataType::INTEGER, true),
            true,
        ),
    )
    .with_metadata(metadata);
    StructType::try_new(vec![array_in_map]).unwrap()
}

/// Build an [`array_in_map_kernel_schema`] with `parquet.field.id` on the top-level field
/// and a nested-ids JSON map (key/value/element) under `nested_ids_meta_key`.
pub(crate) fn array_in_map_with_field_ids(nested_ids_meta_key: &str) -> StructType {
    let nested_ids = MetadataValue::Other(test_utils::nested_ids_json(&[
        ("array_in_map.key", 100),
        ("array_in_map.value", 101),
        ("array_in_map.value.element", 102),
    ]));
    array_in_map_kernel_schema([
        (
            ColumnMetadataKey::ParquetFieldId.as_ref().to_string(),
            MetadataValue::from(1i64),
        ),
        (nested_ids_meta_key.to_string(), nested_ids),
    ])
}

/// Build an empty Arrow `StructArray` matching [`array_in_map_kernel_schema`] with no field-id
/// metadata.
pub(crate) fn array_in_map_arrow_data_without_field_ids() -> StructArray {
    let kernel_schema = array_in_map_kernel_schema(std::iter::empty::<(String, MetadataValue)>());
    let arrow_schema: ArrowSchema = (&kernel_schema).try_into_arrow().unwrap();
    let batch = RecordBatch::new_empty(Arc::new(arrow_schema));
    StructArray::try_new(
        batch.schema().fields.clone(),
        batch.columns().to_vec(),
        None,
    )
    .unwrap()
}

/// Build a [`NestedFieldIdFixture`] covering Array+Map nested-id propagation through a
/// Struct boundary.
///
/// ## 1. Kernel schema
///
/// Two [`StructField`]s `top` and `inner` carry `parquet.field.id` (rewritten to
/// `PARQUET:field_id` in the Arrow output) plus a `<nested_ids_meta_key>` JSON map rooted
/// at each field's name. Each carries an array inside a map.
///
/// ```json
/// {
///   "type": "struct",
///   "fields": [{
///     "name": "top",
///     "type": {
///       "type": "map",
///       "keyType":   {"type": "array", "elementType": "integer"},
///       "valueType": {"type": "struct", "fields": [{
///         "name": "inner",
///         "type": {"type": "map", "keyType": "integer",
///                  "valueType": {"type": "array", "elementType": "integer"}},
///         "metadata": {
///           "parquet.field.id": 2,
///           "<nested_ids_meta_key>": {
///             "inner.key": 200, "inner.value": 201, "inner.value.element": 202
///           }
///         }
///       }]}
///     },
///     "metadata": {
///       "parquet.field.id": 1,
///       "<nested_ids_meta_key>": {
///         "top.key": 100, "top.key.element": 101, "top.value": 102
///       }
///     }
///   }]
/// }
/// ```
///
/// ## 2. Input Arrow schema
///
/// Same shape as kernel schema with no metadata anywhere, except a *stale*
/// `PARQUET:field_id=999` on the synthesized `top.key.element` field.
///
/// ## 3. Expected output Arrow schema
///
/// What `try_into_arrow(kernel schema)` and
/// `apply_schema(input arrow schema, kernel schema)` should both produce:
/// - `top` and `inner` carry `PARQUET:field_id` (rewritten from `parquet.field.id`).
/// - Synthesized list/map `key`/`value`/`element` fields each carry `PARQUET:field_id` pulled from
///   the corresponding nested-ids JSON entry. `top.key.element` has `101` (kernel's), not the stale
///   `999` from the input.
///
/// ```json
/// {
///   "fields": [{
///     "name": "top", "type": "map",
///     "metadata": {"PARQUET:field_id": "1"},
///     "entries": { "name": "key_value", "type": "struct", "fields": [
///       {
///         "name": "key", "type": "list",
///         "metadata": {"PARQUET:field_id": "100"},
///         "element": {
///           "name": "element", "type": "int32",
///           "metadata": {"PARQUET:field_id": "101"}
///         }
///       },
///       {
///         "name": "value", "type": "struct",
///         "metadata": {"PARQUET:field_id": "102"},
///         "fields": [{
///           "name": "inner", "type": "map",
///           "metadata": {"PARQUET:field_id": "2"},
///           "entries": { "name": "key_value", "type": "struct", "fields": [
///             {
///               "name": "key", "type": "int32",
///               "metadata": {"PARQUET:field_id": "200"}
///             },
///             {
///               "name": "value", "type": "list",
///               "metadata": {"PARQUET:field_id": "201"},
///               "element": {
///                 "name": "element", "type": "int32",
///                 "metadata": {"PARQUET:field_id": "202"}
///               }
///             }
///           ]}
///         }]
///       }
///     ]}
///   }]
/// }
/// ```
pub(crate) fn complex_nested_with_field_ids(nested_ids_meta_key: &str) -> NestedFieldIdFixture {
    NestedFieldIdFixture {
        kernel_schema: build_complex_nested_kernel_schema(nested_ids_meta_key),
        input_arrow_data: build_arrow_input_with_stale_element_id(),
        expected_arrow_schema: expected_complex_nested_arrow_schema(),
    }
}

/// Build the input Arrow data for [`complex_nested_with_field_ids`] by
/// striping the metadata from [`build_complex_nested_kernel_schema`], and
/// add one stale `PARQUET:field_id` to the `top.key.element` field.
fn build_arrow_input_with_stale_element_id() -> StructArray {
    // Get the no-meta Arrow shape from the kernel schema.
    let plain_inner = StructField::nullable("inner", complex_nested_inner_map_type());
    let plain_top = StructField::nullable(
        "top",
        complex_nested_outer_map_type(StructType::try_new(vec![plain_inner]).unwrap()),
    );
    let plain_kernel_schema = StructType::try_new(vec![plain_top]).unwrap();
    let plain_arrow_schema: ArrowSchema = (&plain_kernel_schema).try_into_arrow().unwrap();

    // Add stale `PARQUET:field_id` to the `top.key.element` field.
    let top = plain_arrow_schema.field(0);
    let DataType::Map(entries, sorted) = top.data_type() else {
        unreachable!("top is a Map by construction");
    };
    let DataType::Struct(entries_fields) = entries.data_type() else {
        unreachable!("map entries is a Struct by construction");
    };
    let outer_key = &entries_fields[0];
    let DataType::List(outer_element) = outer_key.data_type() else {
        unreachable!("outer map key is a List by construction");
    };
    let stale_element = outer_element
        .as_ref()
        .clone()
        .with_metadata([(PARQUET_FIELD_ID_META_KEY.to_string(), "999".to_string())].into());
    let new_outer_key = Field::new(
        outer_key.name(),
        DataType::List(Arc::new(stale_element)),
        outer_key.is_nullable(),
    );
    let new_entries = Field::new(
        entries.name(),
        DataType::Struct(vec![new_outer_key, entries_fields[1].as_ref().clone()].into()),
        entries.is_nullable(),
    );
    let new_top = Field::new(
        top.name(),
        DataType::Map(Arc::new(new_entries), *sorted),
        top.is_nullable(),
    );
    let arrow_input_schema = ArrowSchema::new(vec![new_top]);
    let batch = RecordBatch::new_empty(Arc::new(arrow_input_schema));
    StructArray::try_new(
        batch.schema().fields.clone(),
        batch.columns().to_vec(),
        None,
    )
    .unwrap()
}

fn complex_nested_inner_map_type() -> KernelDataType {
    KernelDataType::from(MapType::new(
        KernelDataType::INTEGER,
        ArrayType::new(KernelDataType::INTEGER, true),
        true,
    ))
}

fn complex_nested_outer_map_type(struct_value: StructType) -> KernelDataType {
    KernelDataType::from(MapType::new(
        ArrayType::new(KernelDataType::INTEGER, true),
        struct_value,
        true,
    ))
}

/// Build the kernel schema described by [`complex_nested_with_field_ids`].
pub(crate) fn build_complex_nested_kernel_schema(nested_ids_meta_key: &str) -> StructType {
    let top_nested_ids = test_utils::nested_ids_json(&[
        ("top.key", 100),
        ("top.key.element", 101),
        ("top.value", 102),
    ]);
    let inner_nested_ids = test_utils::nested_ids_json(&[
        ("inner.key", 200),
        ("inner.value", 201),
        ("inner.value.element", 202),
    ]);
    // Each `StructField` carries `parquet.field.id` plus the nested-ids JSON.
    let inner_field = StructField::nullable("inner", complex_nested_inner_map_type())
        .with_metadata([
            (
                ColumnMetadataKey::ParquetFieldId.as_ref().to_string(),
                MetadataValue::from(2i64),
            ),
            (
                nested_ids_meta_key.to_string(),
                MetadataValue::Other(inner_nested_ids),
            ),
        ]);
    let top_field = StructField::nullable(
        "top",
        complex_nested_outer_map_type(StructType::try_new(vec![inner_field]).unwrap()),
    )
    .with_metadata([
        (
            ColumnMetadataKey::ParquetFieldId.as_ref().to_string(),
            MetadataValue::from(1i64),
        ),
        (
            nested_ids_meta_key.to_string(),
            MetadataValue::Other(top_nested_ids),
        ),
    ]);
    StructType::try_new(vec![top_field]).unwrap()
}

/// Build the expected output Arrow schema for [`complex_nested_with_field_ids`].
fn expected_complex_nested_arrow_schema() -> ArrowSchema {
    // top.value.inner.value.element: int (PARQUET:field_id=202).
    let inner_list_element = Field::new("element", DataType::Int32, true)
        .with_metadata(parquet_field_id_metadata(Some(202)));
    // top.value.inner.value: list<int> (PARQUET:field_id=201).
    let inner_value = Field::new("value", DataType::List(Arc::new(inner_list_element)), true)
        .with_metadata(parquet_field_id_metadata(Some(201)));
    // top.value.inner.key: int (PARQUET:field_id=200).
    let inner_key = Field::new("key", DataType::Int32, false)
        .with_metadata(parquet_field_id_metadata(Some(200)));
    // top.value.inner.key_value: synthesized map-entries struct (no field id).
    let inner_entries = Field::new(
        "key_value",
        DataType::Struct(vec![inner_key, inner_value].into()),
        false,
    );
    // top.value.inner: map<int, list<int>> (PARQUET:field_id=2).
    let inner_field = Field::new("inner", DataType::Map(Arc::new(inner_entries), false), true)
        .with_metadata(parquet_field_id_metadata(Some(2)));
    // top.value: struct<inner: ...> (PARQUET:field_id=102).
    let struct_value_field = Field::new("value", DataType::Struct(vec![inner_field].into()), true)
        .with_metadata(parquet_field_id_metadata(Some(102)));
    // top.key.element: int (PARQUET:field_id=101).
    let outer_key_element = Field::new("element", DataType::Int32, true)
        .with_metadata(parquet_field_id_metadata(Some(101)));
    // top.key: list<int> (PARQUET:field_id=100).
    let outer_key = Field::new("key", DataType::List(Arc::new(outer_key_element)), false)
        .with_metadata(parquet_field_id_metadata(Some(100)));
    // top.key_value: synthesized map-entries struct (no field id).
    let outer_entries = Field::new(
        "key_value",
        DataType::Struct(vec![outer_key, struct_value_field].into()),
        false,
    );
    // top: map<list<int>, struct<...>> (PARQUET:field_id=1).
    let top_field = Field::new("top", DataType::Map(Arc::new(outer_entries), false), true)
        .with_metadata(parquet_field_id_metadata(Some(1)));
    ArrowSchema::new(vec![top_field])
}

/// Flat schema: `[id: long, name: string]`
pub(crate) fn test_schema_flat() -> SchemaRef {
    schema_ref! {
        nullable "id": LONG,
        nullable "name": STRING,
    }
}

/// Flat schema with column mapping metadata.
pub(crate) fn test_schema_flat_with_column_mapping() -> SchemaRef {
    schema_ref! {
        (cm_field("id", 1, "phys_id", KernelDataType::LONG)),
        (cm_field("name", 2, "phys_name", KernelDataType::STRING)),
    }
}

/// Nested struct schema with array and map inside the struct
pub(crate) fn test_schema_nested() -> SchemaRef {
    schema_ref! {
        nullable "id": LONG,
        nullable "info": {
            nullable "name": STRING,
            nullable "age": INTEGER,
            nullable "tags": { STRING => nullable STRING },
            nullable "scores": [ nullable INTEGER ],
        },
    }
}

/// Nested struct schema with column mapping metadata.
pub(crate) fn test_schema_nested_with_column_mapping() -> SchemaRef {
    schema_ref! {
        (cm_field("id", 1, "phys_id", KernelDataType::LONG)),
        (cm_field("info", 2, "phys_info", schema! {
            (cm_field("name", 3, "phys_name", KernelDataType::STRING)),
            (cm_field("age", 4, "phys_age", KernelDataType::INTEGER)),
            (cm_field("tags", 5, "phys_tags",
                MapType::new(KernelDataType::STRING, KernelDataType::STRING, true))),
            (cm_field("scores", 6, "phys_scores",
                ArrayType::new(KernelDataType::INTEGER, true))),
        })),
    }
}

/// Schema with a map
pub(crate) fn test_schema_with_map() -> SchemaRef {
    schema_ref! {
        nullable "id": LONG,
        nullable "entries": { STRING => nullable {
            nullable "key": STRING,
            nullable "value": INTEGER,
        } },
        nullable "name": STRING,
    }
}

/// Schema with a map and column mapping metadata.
pub(crate) fn test_schema_with_map_and_column_mapping() -> SchemaRef {
    let value_struct = schema! {
        (cm_field("key", 4, "phys_key", KernelDataType::STRING)),
        (cm_field("value", 5, "phys_value", KernelDataType::INTEGER)),
    };
    schema_ref! {
        (cm_field("id", 1, "phys_id", KernelDataType::LONG)),
        (cm_field("entries", 2, "phys_entries",
            MapType::new(KernelDataType::STRING, value_struct, true))),
        (cm_field("name", 3, "phys_name", KernelDataType::STRING)),
    }
}

/// Schema with an array
pub(crate) fn test_schema_with_array() -> SchemaRef {
    schema_ref! {
        nullable "id": LONG,
        nullable "items": [ nullable {
            nullable "label": STRING,
            nullable "count": INTEGER,
        } ],
        nullable "name": STRING,
    }
}

/// Schema with an array and column mapping metadata.
pub(crate) fn test_schema_with_array_and_column_mapping() -> SchemaRef {
    let item_struct = schema! {
        (cm_field("label", 4, "phys_label", KernelDataType::STRING)),
        (cm_field("count", 5, "phys_count", KernelDataType::INTEGER)),
    };
    schema_ref! {
        (cm_field("id", 1, "phys_id", KernelDataType::LONG)),
        (cm_field("items", 2, "phys_items", ArrayType::new(item_struct, true))),
        (cm_field("name", 3, "phys_name", KernelDataType::STRING)),
    }
}

/// Deeply nested schema: struct -> array -> struct -> map(value) -> struct.
///
/// The leaf struct field is intentionally **not** annotated with column mapping metadata,
/// so this schema can be used to test error paths when column mapping is enabled.
pub(crate) fn test_deep_nested_schema_missing_leaf_cm() -> StructType {
    let map_type = MapType::new(
        KernelDataType::STRING,
        schema! { not_null "leaf": INTEGER },
        true,
    );
    let array_type = ArrayType::new(
        schema! { (cm_field("mid_field", 2, "phys_mid_field", map_type)) },
        true,
    );
    schema! {
        (cm_field("top", 1, "phys_top", array_type)),
    }
}

/// Build a create-table transaction with the given schema and column mapping mode.
/// Returns the engine and uncommitted transaction.
pub(crate) fn setup_column_mapping_txn(
    schema: SchemaRef,
    mode: ColumnMappingMode,
) -> DeltaResult<(Arc<dyn Engine>, Transaction<CreateTable>)> {
    let mode_str = match mode {
        ColumnMappingMode::Name => "name",
        ColumnMappingMode::Id => "id",
        ColumnMappingMode::None => "none",
    };
    let store = Arc::new(InMemory::new());
    let engine: Arc<dyn Engine> = Arc::new(SyncEngine::new_with_store(store));

    let txn = create_table("memory:///test_table", schema, "DefaultEngine")
        .with_table_properties([("delta.columnMapping.mode", mode_str)])
        .build(engine.as_ref(), Box::new(FileSystemCommitter::new()))?;
    Ok((engine, txn))
}

/// Validate that a physical schema matches the logical schema's column mapping metadata.
/// For Name/Id modes, checks physicalName, columnMapping.id, and parquet.field.id on
/// each field. For None mode, only checks field names match.
pub(crate) fn validate_physical_schema_column_mapping(
    logical_schema: &StructType,
    physical_schema: &StructType,
    mode: ColumnMappingMode,
) {
    assert_eq!(
        physical_schema.fields().count(),
        logical_schema.fields().count()
    );

    // Collect expected (physical_name, field_id) from logical schema
    let expected: Vec<_> = logical_schema
        .fields()
        .map(|f| {
            let physical_name =
                match f.get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName) {
                    Some(MetadataValue::String(name)) => name.clone(),
                    _ if mode == ColumnMappingMode::None => f.name().to_string(),
                    _ => panic!("Logical field '{}' missing physicalName metadata", f.name()),
                };
            let field_id = match f.get_config_value(&ColumnMetadataKey::ColumnMappingId) {
                Some(MetadataValue::Number(id)) => *id,
                _ if mode == ColumnMappingMode::None => -1,
                _ => panic!(
                    "Logical field '{}' missing columnMapping.id metadata",
                    f.name()
                ),
            };
            (physical_name, field_id)
        })
        .collect();

    // Validate each physical field against expected values
    for (physical_field, (expected_name, expected_id)) in
        physical_schema.fields().zip(expected.iter())
    {
        assert_eq!(
            physical_field.name(),
            expected_name,
            "Physical field name mismatch"
        );

        if mode == ColumnMappingMode::None {
            continue;
        }

        assert_eq!(
            physical_field.get_config_value(&ColumnMetadataKey::ColumnMappingPhysicalName),
            Some(&MetadataValue::String(expected_name.clone())),
            "columnMapping.physicalName mismatch for '{}'",
            physical_field.name()
        );

        assert_eq!(
            physical_field.get_config_value(&ColumnMetadataKey::ColumnMappingId),
            Some(&MetadataValue::Number(*expected_id)),
            "columnMapping.id mismatch for '{}'",
            physical_field.name()
        );

        assert_eq!(
            physical_field.get_config_value(&ColumnMetadataKey::ParquetFieldId),
            Some(&MetadataValue::Number(*expected_id)),
            "parquet.field.id mismatch for '{}'",
            physical_field.name()
        );
    }
}

fn resolve_test_table_path(table_name: &str) -> DeltaResult<(PathBuf, Option<TempDir>)> {
    match load_test_data("tests/data", table_name) {
        Ok(test_dir) => {
            let test_path = test_dir.path().join(table_name);
            Ok((test_path, Some(test_dir)))
        }
        Err(_) => {
            let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                .join("tests/data")
                .join(table_name);
            let path = std::fs::canonicalize(path)
                .map_err(|e| Error::generic(format!("Failed to canonicalize path: {e}")))?;
            Ok((path, None))
        }
    }
}

/// Copies a test-table fixture into a writable temporary directory.
pub(crate) fn copy_test_table(table_name: &str) -> DeltaResult<(Url, TempDir)> {
    let (source, _source_tempdir) = resolve_test_table_path(table_name)?;
    let tempdir = tempfile::tempdir()?;
    let table_path = tempdir.path().join(table_name);
    copy_directory(&source, &table_path)
        .map_err(|e| Error::generic(format!("Failed to copy test table: {e}")))?;
    let url = Url::from_directory_path(&table_path)
        .map_err(|_| Error::generic("Failed to create URL from path"))?;
    Ok((url, tempdir))
}

/// Load a test table from tests/data directory.
/// Tries compressed (tar.zst) first, falls back to extracted.
/// Returns (engine, snapshot, optional tempdir). The TempDir must be kept alive
/// for the duration of the test to prevent premature cleanup of extracted files.
pub(crate) fn load_test_table(
    table_name: &str,
) -> DeltaResult<(Arc<dyn Engine>, SnapshotRef, Option<TempDir>)> {
    let (path, tempdir) = resolve_test_table_path(table_name)?;

    let url = Url::from_directory_path(&path)
        .map_err(|_| Error::generic("Failed to create URL from path"))?;

    let engine = Arc::new(SyncEngine::new());
    let snapshot = Snapshot::builder_for(url).build(engine.as_ref())?;
    Ok((engine, snapshot, tempdir))
}

pub(crate) mod column_mapping_physical_name_dedup_fixtures {
    use crate::schema::{
        schema, ArrayType, ColumnMetadataKey, DataType, MapType, MetadataValue, StructField,
        StructType,
    };

    /// Two fields with the same physical name at different physical paths should be accepted.
    pub(crate) fn same_phy_name_different_paths() -> StructType {
        let nested = StructType::new_unchecked([cm_field("id", 3, "id", DataType::INTEGER)]);
        StructType::new_unchecked([
            cm_field("id", 1, "id", DataType::INTEGER),
            cm_field("nested", 2, "nested", nested),
        ])
    }

    /// Two nested fields with same physical path should be rejected.
    pub(crate) fn deeply_nested_repeat_physical_paths() -> StructType {
        let inner = StructType::new_unchecked([
            cm_field("a", 2, "x", DataType::INTEGER),
            cm_field("b", 3, "x", DataType::INTEGER),
        ]);
        let arr_of_struct = ArrayType::new(inner, true);
        let map_to_arr = MapType::new(DataType::STRING, arr_of_struct, true);
        StructType::new_unchecked([cm_field("outer", 1, "outer", map_to_arr)])
    }

    /// Full logical paths of the two colliding fields in
    /// [`deeply_nested_repeat_physical_paths`], in the order the walker visits them.
    pub(crate) fn deeply_nested_collider_paths() -> (&'static str, &'static str) {
        (
            "outer.`<map value>`.`<array element>`.a",
            "outer.`<map value>`.`<array element>`.b",
        )
    }

    /// Two collision sites in the same schema:
    /// - **shallower** (visited first by DFS): top-level siblings `a` and `b`, both have physical
    ///   name "p".
    /// - **deeper** (never reached): inside `nested` struct, siblings `x` and `y`, both have
    ///   physical name "q".
    ///
    /// Dedup must error at the shallower site and never report the deeper one.
    pub(crate) fn multiple_physical_name_collisions() -> StructType {
        schema! {
            (cm_field("a", 1, "p", schema! { (cm_field("aa", 6, "aa", DataType::INTEGER)) })),
            (cm_field("b", 2, "p", schema! { (cm_field("bb", 7, "bb", DataType::INTEGER)) })),
            (cm_field("nested", 3, "nested", schema! {
                (cm_field("x", 4, "q", DataType::INTEGER)),
                (cm_field("y", 5, "q", DataType::INTEGER)),
            })),
        }
    }

    fn cm_field(name: &str, id: i64, phys: &str, ty: impl Into<DataType>) -> StructField {
        StructField::new(name, ty, true).with_metadata([
            (
                ColumnMetadataKey::ColumnMappingId.as_ref(),
                MetadataValue::Number(id),
            ),
            (
                ColumnMetadataKey::ColumnMappingPhysicalName.as_ref(),
                MetadataValue::String(phys.to_string()),
            ),
        ])
    }
}

#[cfg(feature = "geo-type-in-dev")]
pub(crate) fn geometry_type(crs: &str) -> KernelDataType {
    PrimitiveType::Geometry(Box::new(GeometryType::try_new(crs).unwrap())).into()
}

#[cfg(feature = "geo-type-in-dev")]
pub(crate) fn geography_type(crs: &str, algorithm: EdgeInterpolationAlgorithm) -> KernelDataType {
    PrimitiveType::Geography(Box::new(GeographyType::try_new(crs, algorithm).unwrap())).into()
}