delta_kernel_default_engine 0.28.0

//! Default Parquet handler implementation

use std::collections::HashMap;
use std::num::NonZero;
use std::ops::Range;
use std::sync::Arc;

use delta_kernel::arrow::array::builder::{MapBuilder, MapFieldNames, StringBuilder};
use delta_kernel::arrow::array::{Array, Int64Array, RecordBatch, StringArray, StructArray};
use delta_kernel::arrow::datatypes::{DataType, Field, Schema};
use delta_kernel::engine::arrow_conversion::{TryFromArrow as _, TryIntoArrow as _};
use delta_kernel::engine::arrow_data::ArrowEngineData;
use delta_kernel::engine::arrow_utils::{
    fixup_parquet_read, ordering_needs_row_indexes, parquet_read_plan, RowIndexBuilder,
};
use delta_kernel::engine::parquet_row_group_skipping::ParquetRowGroupSkipping;
use delta_kernel::engine::{reader_options, writer_options};
use delta_kernel::expressions::ColumnName;
use delta_kernel::object_store::path::Path;
use delta_kernel::object_store::{DynObjectStore, ObjectStoreExt as _};
use delta_kernel::parquet::arrow::arrow_reader::{
    ArrowReaderMetadata, ParquetRecordBatchReaderBuilder,
};
use delta_kernel::parquet::arrow::arrow_writer::ArrowWriter;
use delta_kernel::parquet::arrow::async_reader::{
    ParquetObjectReader, ParquetRecordBatchStreamBuilder,
};
use delta_kernel::parquet::arrow::async_writer::{AsyncArrowWriter, ParquetObjectWriter};
use delta_kernel::schema::{SchemaRef, StructType};
use delta_kernel::transaction::BoundWriteContext;
use delta_kernel::{
    CancellationTokenRef, DeltaResult, DeltaResultIteratorStatic, EngineData, Error,
    FileDataReadResultIterator, FileMeta, FoldWithOption as _, ParquetFooter, ParquetHandler,
    PredicateRef,
};
use futures::stream::{self, BoxStream};
use futures::{StreamExt, TryStreamExt};
use uuid::Uuid;

use crate::executor::TaskExecutor;
use crate::file_stream::{FileOpenFuture, FileOpener, FileStream};
use crate::stats::collect_stats;
use crate::UrlExt;

#[derive(Debug)]
pub struct DefaultParquetHandler<E: TaskExecutor> {
    store: Arc<DynObjectStore>,
    task_executor: Arc<E>,
    /// The maximum number of files to read concurrently in [`Self::read_parquet_files()`]. This is
    /// the number of futures buffered by `buffered`, i.e. the file-level I/O readahead depth.
    buffer_size: NonZero<usize>,
    /// The maximum number of rows per RecordBatch yielded by the read stream.
    batch_size: NonZero<usize>,
}

/// Metadata of a data file (typically a parquet file).
#[derive(Debug)]
pub struct DataFileMetadata {
    file_meta: FileMeta,
    /// Collected statistics for this file (includes numRecords, tightBounds, etc.).
    stats: StructArray,
}

impl DataFileMetadata {
    pub fn new(file_meta: FileMeta, stats: StructArray) -> Self {
        Self { file_meta, stats }
    }

    /// Returns the absolute URL of the written file.
    pub fn location(&self) -> &url::Url {
        &self.file_meta.location
    }

    /// Converts this `DataFileMetadata` into an [`EngineData`] record batch matching the schema
    /// returned by [`Transaction::add_files_schema`].
    ///
    /// The `partition_values` map uses physical column names as keys and protocol-serialized
    /// strings as values. `None` represents a null partition value. The serialization layer
    /// converts nulls and empty strings to `None` before reaching this method, so `Some("")`
    /// is not expected in normal usage.
    ///
    /// The `log_path` is the path string written to the Delta log's `add.path` field.
    ///
    /// [`Transaction::add_files_schema`]: delta_kernel::transaction::Transaction::add_files_schema
    pub(crate) fn as_record_batch(
        &self,
        partition_values: &HashMap<String, Option<String>>,
        log_path: &str,
    ) -> DeltaResult<Box<dyn EngineData>> {
        let path = Arc::new(StringArray::from(vec![log_path]));
        let key_builder = StringBuilder::new();
        let val_builder = StringBuilder::new();
        let names = MapFieldNames {
            entry: "key_value".to_string(),
            key: "key".to_string(),
            value: "value".to_string(),
        };
        let mut builder = MapBuilder::new(Some(names), key_builder, val_builder);
        for (k, v) in partition_values {
            builder.keys().append_value(k);
            match v.as_deref() {
                // The serialization layer already converts empty strings to None, so
                // Some("") should not occur. The empty check is purely defensive.
                Some(val) if !val.is_empty() => builder.values().append_value(val),
                _ => builder.values().append_null(),
            }
        }
        builder.append(true)?;
        let partitions = Arc::new(builder.finish());
        // this means max size we can write is i64::MAX (~8EB)
        let size: i64 = self
            .file_meta
            .size
            .try_into()
            .map_err(|_| Error::generic("Failed to convert parquet metadata 'size' to i64"))?;
        let size = Arc::new(Int64Array::from(vec![size]));
        let modification_time = Arc::new(Int64Array::from(vec![self.file_meta.last_modified]));

        let stats_array = Arc::new(self.stats.clone());

        // Build schema dynamically based on stats (stats schema varies based on collected
        // statistics)
        let key_value_struct = DataType::Struct(
            vec![
                Field::new("key", DataType::Utf8, false),
                Field::new("value", DataType::Utf8, true),
            ]
            .into(),
        );
        let schema = Schema::new(vec![
            Field::new("path", DataType::Utf8, false),
            Field::new(
                "partitionValues",
                DataType::Map(
                    Arc::new(Field::new("key_value", key_value_struct, false)),
                    false,
                ),
                false,
            ),
            Field::new("size", DataType::Int64, false),
            Field::new("modificationTime", DataType::Int64, false),
            Field::new("stats", stats_array.data_type().clone(), true),
        ]);

        Ok(Box::new(ArrowEngineData::new(RecordBatch::try_new(
            Arc::new(schema),
            vec![path, partitions, size, modification_time, stats_array],
        )?)))
    }
}

impl<E: TaskExecutor> DefaultParquetHandler<E> {
    pub fn new(store: Arc<DynObjectStore>, task_executor: Arc<E>) -> Self {
        Self {
            store,
            task_executor,
            buffer_size: super::DEFAULT_READ_BUFFER_SIZE,
            batch_size: super::DEFAULT_READ_BATCH_SIZE,
        }
    }

    /// Set the maximum number of files to read concurrently in [Self::read_parquet_files()].
    ///
    /// Defaults to `super::DEFAULT_READ_BUFFER_SIZE`.
    ///
    /// This setting applies only to object-store reads. When every file in the batch uses a
    /// presigned URL (`https://...`), reads bypass the object store and this value has no effect;
    /// use [`Self::with_batch_size`] to tune RecordBatch chunking in that path.
    ///
    /// Memory constraints can be imposed by constraining the buffer size and batch size. Note that
    /// overall memory usage is proportional to the product of these two values.
    /// 1. Batch size governs the size of RecordBatches yielded in each iteration of the stream.
    /// 2. Buffer size governs the number of concurrent file reads (which equals the size of the
    ///    readahead buffer).
    pub fn with_buffer_size(mut self, buffer_size: NonZero<usize>) -> Self {
        self.buffer_size = buffer_size;
        self
    }

    /// Set the maximum number of rows per RecordBatch yielded by [Self::read_parquet_files()].
    ///
    /// Defaults to `super::DEFAULT_READ_BATCH_SIZE` rows.
    pub fn with_batch_size(mut self, batch_size: NonZero<usize>) -> Self {
        self.batch_size = batch_size;
        self
    }

    // Write `data` to `{path}/<uuid>.parquet` as parquet using ArrowWriter and return the parquet
    // metadata (where `<uuid>` is a generated UUIDv4).
    //
    // Note: after encoding the data as parquet, this issues a PUT followed by a HEAD to storage in
    // order to obtain metadata about the object just written.
    async fn write_parquet(
        &self,
        path: &url::Url,
        data: Box<dyn EngineData>,
        stats_columns: &[ColumnName],
        physical_schema: &StructType,
    ) -> DeltaResult<DataFileMetadata> {
        let batch: Box<_> = ArrowEngineData::try_from_engine_data(data)?;
        let record_batch = batch.record_batch();

        // Collect statistics before writing (includes numRecords)
        let stats = collect_stats(record_batch, stats_columns, physical_schema)?;

        let mut buffer = vec![];
        let mut writer = ArrowWriter::try_new_with_options(
            &mut buffer,
            record_batch.schema(),
            writer_options(),
        )?;
        writer.write(record_batch)?;
        writer.close()?; // writer must be closed to write footer

        let size: u64 = buffer
            .len()
            .try_into()
            .map_err(|_| Error::generic("unable to convert usize to u64"))?;
        let name: String = format!("{}.parquet", Uuid::new_v4());
        // fail if path does not end with a trailing slash
        if !path.path().ends_with('/') {
            return Err(Error::generic(format!(
                "Path must end with a trailing slash: {path}"
            )));
        }
        let path = path.join(&name)?;

        self.store
            .put(&Path::from_url_path(path.path())?, buffer.into())
            .await?;

        let metadata = self.store.head(&Path::from_url_path(path.path())?).await?;
        let modification_time = metadata.last_modified.timestamp_millis();
        if size != metadata.size {
            return Err(Error::generic(format!(
                "Size mismatch after writing parquet file: expected {}, got {}",
                size, metadata.size
            )));
        }

        let file_meta = FileMeta::new(path, modification_time, size);
        Ok(DataFileMetadata::new(file_meta, stats))
    }

    /// Write `data` to a new parquet file under the [`BoundWriteContext::write_dir`] and return
    /// Add action metadata ready for [`Transaction::add_files`].
    ///
    /// Note that the schema does not contain the dataChange column. In order to set `data_change`
    /// flag, use [`delta_kernel::transaction::Transaction::with_data_change`].
    ///
    /// [`BoundWriteContext::write_dir`]: delta_kernel::transaction::BoundWriteContext::write_dir
    /// [`Transaction::add_files`]: delta_kernel::transaction::Transaction::add_files
    pub async fn write_parquet_file(
        &self,
        data: Box<dyn EngineData>,
        write_context: &BoundWriteContext,
    ) -> DeltaResult<Box<dyn EngineData>> {
        let file_metadata = self
            .write_parquet(
                &write_context.write_dir(),
                data,
                write_context.stats_columns(),
                write_context.physical_schema().as_ref(),
            )
            .await?;
        super::build_add_file_metadata(file_metadata, write_context)
    }
}

/// Internal async implementation of read_parquet_files
async fn read_parquet_files_impl(
    store: Arc<DynObjectStore>,
    files: Vec<FileMeta>,
    physical_schema: SchemaRef,
    predicate: Option<PredicateRef>,
    buffer_size: usize,
    batch_size: usize,
) -> DeltaResult<BoxStream<'static, DeltaResult<Box<dyn EngineData>>>> {
    if files.is_empty() {
        return Ok(Box::pin(stream::empty()));
    }

    let arrow_schema = Arc::new(physical_schema.as_ref().try_into_arrow()?);

    // get the first FileMeta to decide how to fetch the file.
    // NB: This means that every file in `FileMeta` _must_ have the same scheme or things will break
    // s3://    -> aws   (ParquetOpener)
    // nothing  -> local (ParquetOpener)
    // https:// -> assume presigned URL (and fetch without object_store)
    //   -> reqwest to get data
    //   -> parse to parquet
    // SAFETY: we did is_empty check above, this is ok.
    if files[0].location.is_presigned() {
        let file_opener = Box::new(PresignedUrlOpener::new(
            batch_size,
            physical_schema.clone(),
            predicate,
        ));
        let stream = FileStream::new(files, arrow_schema, file_opener)?.map_ok(
            |record_batch| -> Box<dyn EngineData> { Box::new(ArrowEngineData::new(record_batch)) },
        );
        return Ok(Box::pin(stream));
    }

    // an iterator of futures that open each file
    let file_futures = files.into_iter().map(move |file| {
        let store = store.clone();
        let schema = physical_schema.clone();
        let predicate = predicate.clone();
        async move { open_parquet_file(store, schema, predicate, None, batch_size, file).await }
    });
    // create a stream from that iterator which buffers up to `buffer_size` futures at a time
    let result_stream = stream::iter(file_futures)
        .buffered(buffer_size)
        .try_flatten()
        .map_ok(|record_batch| -> Box<dyn EngineData> {
            Box::new(ArrowEngineData::new(record_batch))
        });

    Ok(Box::pin(result_stream))
}

impl<E: TaskExecutor> ParquetHandler for DefaultParquetHandler<E> {
    fn read_parquet_files(
        &self,
        files: &[FileMeta],
        physical_schema: SchemaRef,
        predicate: Option<PredicateRef>,
    ) -> DeltaResult<FileDataReadResultIterator> {
        self.read_parquet_files_with_cancellation(files, physical_schema, predicate, None)
    }

    fn read_parquet_files_with_cancellation(
        &self,
        files: &[FileMeta],
        physical_schema: SchemaRef,
        predicate: Option<PredicateRef>,
        cancellation_token: Option<CancellationTokenRef>,
    ) -> DeltaResult<FileDataReadResultIterator> {
        let future = read_parquet_files_impl(
            self.store.clone(),
            files.to_vec(),
            physical_schema,
            predicate,
            self.buffer_size.get(),
            self.batch_size.get(),
        );
        super::stream_future_to_cancellable_iter(
            self.task_executor.clone(),
            future,
            cancellation_token,
        )
    }

    /// Writes engine data to a Parquet file at the specified location.
    ///
    /// This implementation uses asynchronous file I/O with object_store to write the Parquet file.
    /// If a file already exists at the given location, it will be overwritten.
    ///
    /// # Parameters
    ///
    /// - `location` - The full URL path where the Parquet file should be written (e.g., `s3://bucket/path/file.parquet`,
    ///   `file:///path/to/file.parquet`).
    /// - `data` - An iterator of engine data to be written to the Parquet file.
    ///
    /// # Returns
    ///
    /// A [`DeltaResult`] indicating success or failure.
    fn write_parquet_file(
        &self,
        location: url::Url,
        mut data: DeltaResultIteratorStatic<Box<dyn EngineData>>,
    ) -> DeltaResult<()> {
        let store = self.store.clone();

        self.task_executor.block_on(async move {
            let path = Path::from_url_path(location.path())?;

            // Get first batch to initialize writer with schema
            let first_batch = data.next().ok_or_else(|| {
                Error::generic("Cannot write parquet file with empty data iterator")
            })??;
            let first_arrow = ArrowEngineData::try_from_engine_data(first_batch)?;
            let first_record_batch: RecordBatch = (*first_arrow).into();

            let object_writer = ParquetObjectWriter::new(store, path);
            let schema = first_record_batch.schema();
            let mut writer =
                AsyncArrowWriter::try_new_with_options(object_writer, schema, writer_options())?;

            // Write the first batch
            writer.write(&first_record_batch).await?;

            // Write remaining batches
            for result in data {
                let engine_data = result?;
                let arrow_data = ArrowEngineData::try_from_engine_data(engine_data)?;
                let batch: RecordBatch = (*arrow_data).into();
                writer.write(&batch).await?;
            }

            writer.finish().await?;

            Ok(())
        })
    }

    fn read_parquet_footer(&self, file: &FileMeta) -> DeltaResult<ParquetFooter> {
        self.read_parquet_footer_with_cancellation(file, None)
    }

    fn read_parquet_footer_with_cancellation(
        &self,
        file: &FileMeta,
        cancellation_token: Option<CancellationTokenRef>,
    ) -> DeltaResult<ParquetFooter> {
        let store = self.store.clone();
        let location = file.location.clone();
        let file_size = file.size;

        let footer_future = async move {
            let metadata = if location.is_presigned() {
                let client = reqwest::Client::new();
                let response =
                    client.get(location.as_str()).send().await.map_err(|e| {
                        Error::generic(format!("Failed to fetch presigned URL: {e}"))
                    })?;
                let bytes = response
                    .bytes()
                    .await
                    .map_err(|e| Error::generic(format!("Failed to read response bytes: {e}")))?;
                ArrowReaderMetadata::load(&bytes, reader_options())?
            } else {
                let path = Path::from_url_path(location.path())?;
                let mut reader = ParquetObjectReader::new(store, path).with_file_size(file_size);
                ArrowReaderMetadata::load_async(&mut reader, reader_options()).await?
            };

            let schema = Arc::new(StructType::try_from_arrow(metadata.schema().as_ref())?);
            Ok(ParquetFooter { schema })
        };

        // Race the footer read against cancellation so a cancelled request stops promptly.
        match cancellation_token {
            Some(token) => super::block_on_or_cancelled(&self.task_executor, token, footer_future)
                .unwrap_or(Err(Error::Cancelled)),
            None => self.task_executor.block_on(footer_future),
        }
    }
}

/// Opens a Parquet file and returns a stream of record batches
async fn open_parquet_file(
    store: Arc<DynObjectStore>,
    table_schema: SchemaRef,
    predicate: Option<PredicateRef>,
    limit: Option<usize>,
    batch_size: usize,
    file_meta: FileMeta,
) -> DeltaResult<BoxStream<'static, DeltaResult<RecordBatch>>> {
    let file_location = file_meta.location.to_string();
    let path = Path::from_url_path(file_meta.location.path())?;

    let mut reader = {
        use delta_kernel::object_store::ObjectStoreScheme;
        // HACK: unfortunately, `ParquetObjectReader` under the hood does a suffix range
        // request which isn't supported by Azure. For now we just detect if the URL is
        // pointing to azure and if so, do a HEAD request so we can pass in file size to the
        // reader which will cause the reader to avoid a suffix range request.
        // see also: https://github.com/delta-io/delta-kernel-rs/issues/968

        // Since the `Remove` action's size value is optional as specified in the delta protocol
        // https://github.com/delta-io/delta/blob/master/PROTOCOL.md#add-file-and-remove-file,
        // the extracted size will be zero in this case. Thus, this function
        // need to handle the case of zero file_meta.size.
        if file_meta.size != 0 {
            ParquetObjectReader::new(store, path).with_file_size(file_meta.size)
        } else if let Ok((ObjectStoreScheme::MicrosoftAzure, _)) =
            ObjectStoreScheme::parse(&file_meta.location)
        {
            // also note doing HEAD then actual GET isn't atomic, and leaves us vulnerable
            // to file changing between the two calls.
            let meta = store.head(&path).await?;
            ParquetObjectReader::new(store, path).with_file_size(meta.size)
        } else {
            ParquetObjectReader::new(store, path)
        }
    };

    let metadata = ArrowReaderMetadata::load_async(&mut reader, reader_options()).await?;
    let (requested_ordering, mask) = parquet_read_plan(&table_schema, &metadata)?;

    let mut row_indexes = ordering_needs_row_indexes(&requested_ordering)
        .then(|| RowIndexBuilder::new(metadata.metadata().row_groups()));

    let builder = ParquetRecordBatchStreamBuilder::new_with_metadata(reader, metadata)
        .fold_with(mask, ParquetRecordBatchStreamBuilder::with_projection)
        .fold_with(predicate, |builder, predicate| {
            builder.with_row_group_filter(predicate.as_ref(), row_indexes.as_mut())
        })
        .fold_with(limit, ParquetRecordBatchStreamBuilder::with_limit)
        .with_batch_size(batch_size);

    let mut row_indexes = row_indexes.map(|rb| rb.build()).transpose()?;
    let stream = builder.build()?;

    let stream = stream.map(move |rbr| {
        fixup_parquet_read(
            rbr?,
            &requested_ordering,
            row_indexes.as_mut(),
            Some(&file_location),
            Some(&table_schema),
        )
        .map(Into::into)
    });
    Ok(stream.boxed())
}

/// Implements [`FileOpener`] for a opening a parquet file from a presigned URL
struct PresignedUrlOpener {
    batch_size: usize,
    predicate: Option<PredicateRef>,
    limit: Option<usize>,
    table_schema: SchemaRef,
    client: reqwest::Client,
}

impl PresignedUrlOpener {
    pub(crate) fn new(
        batch_size: usize,
        schema: SchemaRef,
        predicate: Option<PredicateRef>,
    ) -> Self {
        Self {
            batch_size,
            table_schema: schema,
            predicate,
            limit: None,
            client: reqwest::Client::new(),
        }
    }
}

impl FileOpener for PresignedUrlOpener {
    fn open(&self, file_meta: FileMeta, _range: Option<Range<i64>>) -> DeltaResult<FileOpenFuture> {
        let batch_size = self.batch_size;
        let table_schema = self.table_schema.clone();
        let predicate = self.predicate.clone();
        let limit = self.limit;
        let client = self.client.clone(); // uses Arc internally according to reqwest docs
        let file_location = file_meta.location.to_string();

        Ok(Box::pin(async move {
            // fetch the file from the interweb
            let reader = client.get(&file_location).send().await?.bytes().await?;
            let metadata = ArrowReaderMetadata::load(&reader, reader_options())?;
            let (requested_ordering, mask) = parquet_read_plan(&table_schema, &metadata)?;

            let mut row_indexes = ordering_needs_row_indexes(&requested_ordering)
                .then(|| RowIndexBuilder::new(metadata.metadata().row_groups()));

            let reader = ParquetRecordBatchReaderBuilder::new_with_metadata(reader, metadata)
                .fold_with(mask, ParquetRecordBatchReaderBuilder::with_projection)
                .fold_with(predicate, |builder, predicate| {
                    builder.with_row_group_filter(predicate.as_ref(), row_indexes.as_mut())
                })
                .fold_with(limit, ParquetRecordBatchReaderBuilder::with_limit)
                .with_batch_size(batch_size)
                .build()?;

            let mut row_indexes = row_indexes.map(|rb| rb.build()).transpose()?;
            let stream = futures::stream::iter(reader);
            let stream = stream.map(move |rbr| {
                fixup_parquet_read(
                    rbr?,
                    &requested_ordering,
                    row_indexes.as_mut(),
                    Some(&file_location),
                    Some(&table_schema),
                )
                .map(Into::into)
            });
            Ok(stream.boxed())
        }))
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::slice;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use bytes::Bytes;
    use delta_kernel::actions::{NUM_RECORDS, TIGHT_BOUNDS};
    use delta_kernel::arrow::array::{
        Array, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Float32Array, Float64Array,
        Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, StringArray,
        TimestampMicrosecondArray, TimestampMillisecondArray,
    };
    use delta_kernel::arrow::datatypes::{
        DataType as ArrowDataType, Field, Schema as ArrowSchema, TimeUnit,
    };
    use delta_kernel::engine::arrow_conversion::TryIntoKernel as _;
    use delta_kernel::engine::arrow_data::ArrowEngineData;
    use delta_kernel::object_store::local::LocalFileSystem;
    use delta_kernel::object_store::memory::InMemory;
    use delta_kernel::object_store::{
        CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
        PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
    };
    use delta_kernel::parquet::arrow::{ARROW_SCHEMA_META_KEY, PARQUET_FIELD_ID_META_KEY};
    use delta_kernel::schema::{
        schema, schema_ref, ColumnMetadataKey, DataType, MetadataValue, StructField, StructType,
    };
    use delta_kernel::EngineData;
    use delta_kernel_default_engine_test_utils::{
        assert_result_error_with_message, current_time_ms,
        try_into_record_batch as into_record_batch,
    };
    use itertools::Itertools;
    use test_utils::engine_contract::{
        test_parquet_handler_footer_errors_on_missing_file,
        test_parquet_handler_footer_preserves_field_ids,
        test_parquet_handler_reads_file_with_arrow_schema, test_parquet_handler_reads_footer,
        test_parquet_handler_write_always_overwrites,
        test_parquet_handler_write_omits_arrow_schema,
    };
    use url::Url;

    use super::*;
    use crate::executor::tokio::TokioBackgroundExecutor;
    use crate::DEFAULT_BATCH_SIZE;

    fn long_schema(name: &str) -> StructType {
        schema! { nullable (name): LONG }
    }

    /// Test `ObjectStore` that counts footer fetches. `get_opts` (footer range GETs) is counted;
    /// column-chunk data goes through `get_ranges` and is not counted, so the count isolates
    /// footer reads.
    #[derive(Debug)]
    struct GetOptsCountingStore<T: ObjectStore> {
        inner: T,
        get_opts_count: AtomicUsize,
    }

    impl<T: ObjectStore> GetOptsCountingStore<T> {
        fn new(inner: T) -> Self {
            Self {
                inner,
                get_opts_count: AtomicUsize::new(0),
            }
        }

        fn get_opts_count(&self) -> usize {
            self.get_opts_count.load(Ordering::SeqCst)
        }
    }

    impl<T: ObjectStore> std::fmt::Display for GetOptsCountingStore<T> {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "GetOptsCountingStore({})", self.get_opts_count())
        }
    }

    #[async_trait::async_trait]
    impl<T: ObjectStore> delta_kernel::object_store::ObjectStore for GetOptsCountingStore<T> {
        // ===== The method we instrument: count footer-fetch GETs =====
        async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
            self.get_opts_count.fetch_add(1, Ordering::SeqCst);
            self.inner.get_opts(location, options).await
        }

        // ===== Everything else: behavior unchanged, delegate to inner =====
        // Overridden (not inherited) so column-chunk data reads delegate straight to inner and
        // stay off the get_opts counter.
        async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
            self.inner.get_ranges(location, ranges).await
        }

        async fn put_opts(
            &self,
            location: &Path,
            payload: PutPayload,
            opts: PutOptions,
        ) -> Result<PutResult> {
            self.inner.put_opts(location, payload, opts).await
        }

        async fn put_multipart_opts(
            &self,
            location: &Path,
            opts: PutMultipartOptions,
        ) -> Result<Box<dyn MultipartUpload>> {
            self.inner.put_multipart_opts(location, opts).await
        }

        fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
            self.inner.list(prefix)
        }

        async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
            self.inner.list_with_delimiter(prefix).await
        }

        fn delete_stream(
            &self,
            locations: BoxStream<'static, Result<Path>>,
        ) -> BoxStream<'static, Result<Path>> {
            self.inner.delete_stream(locations)
        }

        async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
            self.inner.copy_opts(from, to, options).await
        }
    }

    async fn read_all_rows_helper(file_meta: FileMeta) -> DeltaResult<Vec<RecordBatch>> {
        let store = Arc::new(LocalFileSystem::new());
        let path = Path::from_url_path(file_meta.location.path()).unwrap();
        let reader = ParquetObjectReader::new(store.clone(), path);
        let physical_schema = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .schema()
            .clone();
        let stream = open_parquet_file(
            store,
            Arc::new(physical_schema.try_into_kernel().unwrap()),
            None,
            None,
            DEFAULT_BATCH_SIZE,
            file_meta,
        )
        .await
        .unwrap();

        let batches: Vec<RecordBatch> = stream.try_collect().await.unwrap();
        Ok(batches)
    }

    #[tokio::test]
    async fn test_open_parquet_file_with_size() {
        let path = std::fs::canonicalize(PathBuf::from(
            "../kernel/tests/data/table-with-dv-small/part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet"
        )).unwrap();
        let file_size = std::fs::metadata(&path).unwrap().len();
        let url = Url::from_file_path(path).unwrap();
        let file_meta = FileMeta {
            location: url,
            last_modified: 0,
            size: file_size,
        };
        let data = read_all_rows_helper(file_meta).await.unwrap();

        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 10);
    }

    #[tokio::test]
    async fn test_open_parquet_file_without_size() {
        let path = std::fs::canonicalize(PathBuf::from(
            "../kernel/tests/data/table-with-dv-small/part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet"
        )).unwrap();
        let url = Url::from_file_path(path).unwrap();
        let file_meta = FileMeta {
            location: url,
            last_modified: 0,
            size: 0,
        };
        let data = read_all_rows_helper(file_meta).await.unwrap();

        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 10);
    }

    #[tokio::test]
    async fn test_open_parquet_file_fetches_footer_once() {
        let path = std::fs::canonicalize(PathBuf::from(
            "../kernel/tests/data/table-with-dv-small/part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet"
        )).unwrap();
        let file_size = std::fs::metadata(&path).unwrap().len();
        let url = Url::from_file_path(&path).unwrap();
        let location = Path::from_url_path(url.path()).unwrap();

        // Baseline: how many `get_opts` calls a single footer load makes (one or more range GETs,
        // depending on parquet version). The reader builder must not exceed this.
        let baseline_store = Arc::new(GetOptsCountingStore::new(LocalFileSystem::new()));
        let mut reader =
            ParquetObjectReader::new(baseline_store.clone(), location).with_file_size(file_size);
        let metadata = ArrowReaderMetadata::load_async(&mut reader, reader_options())
            .await
            .unwrap();
        let footer_load_gets = baseline_store.get_opts_count();
        assert!(
            footer_load_gets > 0,
            "footer load should issue at least one GET"
        );
        let physical_schema = Arc::new(metadata.schema().clone().try_into_kernel().unwrap());

        // `open_parquet_file` must reuse the loaded footer rather than re-fetch it when building
        // the reader, so its footer GETs equal a single load, not double.
        let counting_store = Arc::new(GetOptsCountingStore::new(LocalFileSystem::new()));
        let file_meta = FileMeta {
            location: url,
            last_modified: 0,
            size: file_size,
        };
        let stream = open_parquet_file(
            counting_store.clone(),
            physical_schema,
            None,
            None,
            DEFAULT_BATCH_SIZE,
            file_meta,
        )
        .await
        .unwrap();
        let _batches: Vec<RecordBatch> = stream.try_collect().await.unwrap();

        assert_eq!(
            counting_store.get_opts_count(),
            footer_load_gets,
            "footer should be fetched once, not re-fetched when constructing the reader"
        );
    }

    #[tokio::test]
    async fn test_read_parquet_files() {
        let store = Arc::new(LocalFileSystem::new());

        let path = std::fs::canonicalize(PathBuf::from(
            "../kernel/tests/data/table-with-dv-small/part-00000-fae5310a-a37d-4e51-827b-c3d5516560ca-c000.snappy.parquet"
        )).unwrap();
        let url = url::Url::from_file_path(path).unwrap();
        let location = Path::from_url_path(url.path()).unwrap();
        let meta = store.head(&location).await.unwrap();

        let reader = ParquetObjectReader::new(store.clone(), location);
        let physical_schema = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .schema()
            .clone();

        let files = &[FileMeta {
            location: url.clone(),
            last_modified: meta.last_modified.timestamp(),
            size: meta.size,
        }];

        let handler = DefaultParquetHandler::new(store, Arc::new(TokioBackgroundExecutor::new()));
        let data: Vec<RecordBatch> = handler
            .read_parquet_files(
                files,
                Arc::new(physical_schema.try_into_kernel().unwrap()),
                None,
            )
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 10);
    }

    // A Parquet file can physically store TIMESTAMP(MILLIS) (for example, checkpoint stats
    // written by a non-kernel writer). End to end, the default engine must (a) convert the
    // millisecond footer schema to the kernel's microsecond TIMESTAMP / TIMESTAMP_NTZ, and
    // (b) rescale the values to microseconds (x1000) when reading them into that schema.
    #[tokio::test]
    async fn test_read_millisecond_timestamps() {
        let temp_dir = tempfile::tempdir().unwrap();
        let file_path = temp_dir.path().join("ms_timestamps.parquet");

        // 1700000000123 ms since epoch -> 1_700_000_000_123_000 us.
        let ms: i64 = 1_700_000_000_123;
        let arrow_schema = Arc::new(ArrowSchema::new(vec![
            Field::new(
                "ts_utc",
                ArrowDataType::Timestamp(TimeUnit::Millisecond, Some("UTC".into())),
                true,
            ),
            Field::new(
                "ts_ntz",
                ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
                true,
            ),
        ]));
        let batch = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![
                Arc::new(TimestampMillisecondArray::from(vec![ms]).with_timezone("UTC"))
                    as Arc<dyn Array>,
                Arc::new(TimestampMillisecondArray::from(vec![ms])) as Arc<dyn Array>,
            ],
        )
        .unwrap();

        let mut writer = ArrowWriter::try_new(
            std::fs::File::create(&file_path).unwrap(),
            arrow_schema,
            None,
        )
        .unwrap();
        writer.write(&batch).unwrap();
        writer.close().unwrap();

        let file_size = std::fs::metadata(&file_path).unwrap().len();
        let file_meta = FileMeta {
            location: Url::from_file_path(&file_path).unwrap(),
            last_modified: 0,
            size: file_size,
        };

        let handler = DefaultParquetHandler::new(
            Arc::new(LocalFileSystem::new()),
            Arc::new(TokioBackgroundExecutor::new()),
        );

        // (a) Footer schema: millisecond timestamps convert to the kernel's microsecond types.
        let footer = handler.read_parquet_footer(&file_meta).unwrap();
        assert_eq!(
            footer.schema.field("ts_utc").unwrap().data_type(),
            &DataType::TIMESTAMP
        );
        assert_eq!(
            footer.schema.field("ts_ntz").unwrap().data_type(),
            &DataType::TIMESTAMP_NTZ
        );

        // (b) Data path: values are rescaled ms -> us when read into the microsecond schema.
        let data: Vec<RecordBatch> = handler
            .read_parquet_files(slice::from_ref(&file_meta), footer.schema.clone(), None)
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 1);
        let expected_us = ms * 1_000;
        for col_idx in 0..2 {
            let col = data[0]
                .column(col_idx)
                .as_any()
                .downcast_ref::<TimestampMicrosecondArray>()
                .unwrap();
            assert_eq!(
                col.value(0),
                expected_us,
                "column {col_idx} should be rescaled to us"
            );
        }
    }

    #[rstest::rstest]
    fn test_as_record_batch(
        #[values(None, Some("a".to_string()))] partition_value: Option<String>,
    ) {
        let location = Url::parse("file:///test_url").unwrap();
        let size = 1_000_000;
        let last_modified = 10000000000;
        let num_records = 10;
        let file_metadata = FileMeta::new(location.clone(), last_modified, size);
        let stats = StructArray::try_new(
            vec![
                Field::new(NUM_RECORDS, ArrowDataType::Int64, true),
                Field::new(TIGHT_BOUNDS, ArrowDataType::Boolean, true),
            ]
            .into(),
            vec![
                Arc::new(Int64Array::from(vec![num_records as i64])),
                Arc::new(BooleanArray::from(vec![true])),
            ],
            None,
        )
        .unwrap();
        let data_file_metadata = DataFileMetadata::new(file_metadata, stats.clone());
        let partition_values = HashMap::from([("partition1".to_string(), partition_value.clone())]);
        let actual = data_file_metadata
            .as_record_batch(&partition_values, "test_url")
            .unwrap();
        let actual = ArrowEngineData::try_from_engine_data(actual).unwrap();

        let mut partition_values_builder = MapBuilder::new(
            Some(MapFieldNames {
                entry: "key_value".to_string(),
                key: "key".to_string(),
                value: "value".to_string(),
            }),
            StringBuilder::new(),
            StringBuilder::new(),
        );

        partition_values_builder.keys().append_value("partition1");
        match &partition_value {
            None => partition_values_builder.values().append_null(),
            Some(v) => partition_values_builder.values().append_value(v),
        }
        partition_values_builder.append(true).unwrap();
        let partition_values = partition_values_builder.finish();

        // Build expected schema dynamically based on stats
        let stats_field = Field::new("stats", stats.data_type().clone(), true);
        let schema = Arc::new(delta_kernel::arrow::datatypes::Schema::new(vec![
            Field::new("path", ArrowDataType::Utf8, false),
            Field::new(
                "partitionValues",
                ArrowDataType::Map(
                    Arc::new(Field::new(
                        "key_value",
                        ArrowDataType::Struct(
                            vec![
                                Field::new("key", ArrowDataType::Utf8, false),
                                Field::new("value", ArrowDataType::Utf8, true),
                            ]
                            .into(),
                        ),
                        false,
                    )),
                    false,
                ),
                false,
            ),
            Field::new("size", ArrowDataType::Int64, false),
            Field::new("modificationTime", ArrowDataType::Int64, false),
            stats_field,
        ]));

        let expected = RecordBatch::try_new(
            schema,
            vec![
                Arc::new(StringArray::from(vec!["test_url"])),
                Arc::new(partition_values),
                Arc::new(Int64Array::from(vec![size as i64])),
                Arc::new(Int64Array::from(vec![last_modified])),
                Arc::new(stats),
            ],
        )
        .unwrap();

        assert_eq!(actual.record_batch(), &expected);
    }

    #[tokio::test]
    async fn test_write_parquet() {
        let store = Arc::new(InMemory::new());
        let parquet_handler =
            DefaultParquetHandler::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));

        let data = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![(
                "a",
                Arc::new(Int64Array::from(vec![1, 2, 3])) as Arc<dyn Array>,
            )])
            .unwrap(),
        ));
        let physical_schema = long_schema("a");

        let write_metadata = parquet_handler
            .write_parquet(
                &Url::parse("memory:///data/").unwrap(),
                data,
                &[],
                &physical_schema,
            )
            .await
            .unwrap();

        let DataFileMetadata {
            file_meta:
                ref parquet_file @ FileMeta {
                    ref location,
                    last_modified,
                    size,
                },
            ref stats,
        } = write_metadata;
        let expected_location = Url::parse("memory:///data/").unwrap();

        // head the object to get metadata
        let meta = store
            .head(&Path::from_url_path(location.path()).unwrap())
            .await
            .unwrap();
        let expected_size = meta.size;

        // check that last_modified is within 10s of now
        let now: i64 = current_time_ms().unwrap();

        let filename = location.path().split('/').next_back().unwrap();
        assert_eq!(&expected_location.join(filename).unwrap(), location);
        assert_eq!(expected_size, size);
        assert!(now - last_modified < 10_000);

        // Check numRecords from stats
        let num_records = stats
            .column_by_name(NUM_RECORDS)
            .unwrap()
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap()
            .value(0);
        assert_eq!(num_records, 3);

        // check we can read back
        let path = Path::from_url_path(location.path()).unwrap();
        let reader = ParquetObjectReader::new(store.clone(), path);
        let physical_schema = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .schema()
            .clone();

        let data: Vec<RecordBatch> = parquet_handler
            .read_parquet_files(
                slice::from_ref(parquet_file),
                Arc::new(physical_schema.try_into_kernel().unwrap()),
                None,
            )
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 3);
    }

    #[tokio::test]
    async fn test_disallow_non_trailing_slash() {
        let store = Arc::new(InMemory::new());
        let parquet_handler =
            DefaultParquetHandler::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));

        let data = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![(
                "a",
                Arc::new(Int64Array::from(vec![1, 2, 3])) as Arc<dyn Array>,
            )])
            .unwrap(),
        ));
        let physical_schema = long_schema("a");

        assert_result_error_with_message(
            parquet_handler
                .write_parquet(
                    &Url::parse("memory:///data").unwrap(),
                    data,
                    &[],
                    &physical_schema,
                )
                .await,
            "Generic delta kernel error: Path must end with a trailing slash: memory:///data",
        );
    }

    #[tokio::test]
    async fn test_parquet_handler_trait_write() {
        let store = Arc::new(InMemory::new());
        let parquet_handler: Arc<dyn ParquetHandler> = Arc::new(DefaultParquetHandler::new(
            store.clone(),
            Arc::new(TokioBackgroundExecutor::new()),
        ));

        let engine_data: Box<dyn EngineData> = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![
                (
                    "x",
                    Arc::new(Int64Array::from(vec![10, 20, 30])) as Arc<dyn Array>,
                ),
                (
                    "y",
                    Arc::new(Int64Array::from(vec![100, 200, 300])) as Arc<dyn Array>,
                ),
            ])
            .unwrap(),
        ));

        // Create iterator with single batch
        let data_iter: DeltaResultIteratorStatic<Box<dyn EngineData>> =
            Box::new(std::iter::once(Ok(engine_data)));

        // Test writing through the trait method
        let file_url = Url::parse("memory:///test/data.parquet").unwrap();
        parquet_handler
            .write_parquet_file(file_url.clone(), data_iter)
            .unwrap();

        // Verify we can read the file back
        let path = Path::from_url_path(file_url.path()).unwrap();
        let metadata = store.head(&path).await.unwrap();
        let reader = ParquetObjectReader::new(store.clone(), path);
        let physical_schema = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .schema()
            .clone();

        let file_meta = FileMeta {
            location: file_url,
            last_modified: 0,
            size: metadata.size,
        };

        let data: Vec<RecordBatch> = parquet_handler
            .read_parquet_files(
                slice::from_ref(&file_meta),
                Arc::new(physical_schema.try_into_kernel().unwrap()),
                None,
            )
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 3);
        assert_eq!(data[0].num_columns(), 2);
    }

    #[tokio::test]
    async fn test_read_parquet_files_respects_batch_size() {
        let store = Arc::new(InMemory::new());
        let writer: Arc<dyn ParquetHandler> = Arc::new(DefaultParquetHandler::new(
            store.clone(),
            Arc::new(TokioBackgroundExecutor::new()),
        ));

        let engine_data: Box<dyn EngineData> = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![(
                "x",
                Arc::new(Int64Array::from((0..10).collect::<Vec<_>>())) as Arc<dyn Array>,
            )])
            .unwrap(),
        ));
        let file_url = Url::parse("memory:///test/batch_size.parquet").unwrap();
        writer
            .write_parquet_file(file_url.clone(), Box::new(std::iter::once(Ok(engine_data))))
            .unwrap();

        let path = Path::from_url_path(file_url.path()).unwrap();
        let metadata = store.head(&path).await.unwrap();
        let reader = ParquetObjectReader::new(store.clone(), path);
        let physical_schema = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .schema()
            .clone();
        let file_meta = FileMeta {
            location: file_url,
            last_modified: 0,
            size: metadata.size,
        };

        // With a batch size of 4, the 10-row file should be split into batches of 4, 4, 2.
        let handler =
            DefaultParquetHandler::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()))
                .with_batch_size(NonZero::new(4).unwrap());
        let data: Vec<RecordBatch> = handler
            .read_parquet_files(
                slice::from_ref(&file_meta),
                Arc::new(physical_schema.try_into_kernel().unwrap()),
                None,
            )
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        let row_counts: Vec<usize> = data.iter().map(|b| b.num_rows()).collect();
        assert_eq!(row_counts, vec![4, 4, 2]);
    }

    #[tokio::test]
    async fn test_parquet_handler_trait_write_and_read_roundtrip() {
        let store = Arc::new(InMemory::new());
        let parquet_handler: Arc<dyn ParquetHandler> = Arc::new(DefaultParquetHandler::new(
            store.clone(),
            Arc::new(TokioBackgroundExecutor::new()),
        ));

        // Create test data with all Delta-supported primitive types
        let engine_data: Box<dyn EngineData> = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![
                // Byte (i8)
                (
                    "byte_col",
                    Arc::new(Int8Array::from(vec![1i8, 2, 3, 4, 5])) as Arc<dyn Array>,
                ),
                // Short (i16)
                (
                    "short_col",
                    Arc::new(Int16Array::from(vec![100i16, 200, 300, 400, 500])) as Arc<dyn Array>,
                ),
                // Integer (i32)
                (
                    "int_col",
                    Arc::new(Int32Array::from(vec![1000i32, 2000, 3000, 4000, 5000]))
                        as Arc<dyn Array>,
                ),
                // Long (i64)
                (
                    "long_col",
                    Arc::new(Int64Array::from(vec![10000i64, 20000, 30000, 40000, 50000]))
                        as Arc<dyn Array>,
                ),
                // Float (f32)
                (
                    "float_col",
                    Arc::new(Float32Array::from(vec![1.1f32, 2.2, 3.3, 4.4, 5.5]))
                        as Arc<dyn Array>,
                ),
                // Double (f64)
                (
                    "double_col",
                    Arc::new(Float64Array::from(vec![1.11f64, 2.22, 3.33, 4.44, 5.55]))
                        as Arc<dyn Array>,
                ),
                // Boolean
                (
                    "bool_col",
                    Arc::new(BooleanArray::from(vec![true, false, true, false, true]))
                        as Arc<dyn Array>,
                ),
                // String
                (
                    "string_col",
                    Arc::new(StringArray::from(vec!["a", "b", "c", "d", "e"])) as Arc<dyn Array>,
                ),
                // Binary
                (
                    "binary_col",
                    Arc::new(BinaryArray::from_vec(vec![
                        b"bin1", b"bin2", b"bin3", b"bin4", b"bin5",
                    ])) as Arc<dyn Array>,
                ),
                // Date
                (
                    "date_col",
                    Arc::new(Date32Array::from(vec![18262, 18263, 18264, 18265, 18266]))
                        as Arc<dyn Array>, // Days since epoch (2020-01-01 onwards)
                ),
                // Timestamp (with UTC timezone)
                (
                    "timestamp_col",
                    Arc::new(
                        TimestampMicrosecondArray::from(vec![
                            1609459200000000i64, // 2021-01-01 00:00:00 UTC
                            1609545600000000i64,
                            1609632000000000i64,
                            1609718400000000i64,
                            1609804800000000i64,
                        ])
                        .with_timezone("UTC"),
                    ) as Arc<dyn Array>,
                ),
                // TimestampNtz (without timezone)
                (
                    "timestamp_ntz_col",
                    Arc::new(TimestampMicrosecondArray::from(vec![
                        1609459200000000i64, // 2021-01-01 00:00:00
                        1609545600000000i64,
                        1609632000000000i64,
                        1609718400000000i64,
                        1609804800000000i64,
                    ])) as Arc<dyn Array>,
                ),
                // Decimal (precision 10, scale 2)
                (
                    "decimal_col",
                    Arc::new(
                        Decimal128Array::from(vec![12345i128, 23456, 34567, 45678, 56789])
                            .with_precision_and_scale(10, 2)
                            .unwrap(),
                    ) as Arc<dyn Array>,
                ),
            ])
            .unwrap(),
        ));

        // Create iterator with single batch
        let data_iter: DeltaResultIteratorStatic<Box<dyn EngineData>> =
            Box::new(std::iter::once(Ok(engine_data)));

        // Write the data
        let file_url = Url::parse("memory:///roundtrip/test.parquet").unwrap();
        parquet_handler
            .write_parquet_file(file_url.clone(), data_iter)
            .unwrap();

        // Read it back
        let path = Path::from_url_path(file_url.path()).unwrap();
        let metadata = store.head(&path).await.unwrap();
        let reader = ParquetObjectReader::new(store.clone(), path);
        let physical_schema = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .schema()
            .clone();

        let file_meta = FileMeta {
            location: file_url.clone(),
            last_modified: 0,
            size: metadata.size,
        };

        let data: Vec<RecordBatch> = parquet_handler
            .read_parquet_files(
                slice::from_ref(&file_meta),
                Arc::new(physical_schema.try_into_kernel().unwrap()),
                None,
            )
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        // Verify the data
        assert_eq!(data.len(), 1);
        assert_eq!(data[0].num_rows(), 5);
        assert_eq!(data[0].num_columns(), 13);

        let mut col_idx = 0;

        // Verify byte column
        let byte_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Int8Array>()
            .unwrap();
        assert_eq!(byte_col.values(), &[1i8, 2, 3, 4, 5]);
        col_idx += 1;

        // Verify short column
        let short_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Int16Array>()
            .unwrap();
        assert_eq!(short_col.values(), &[100i16, 200, 300, 400, 500]);
        col_idx += 1;

        // Verify int column
        let int_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Int32Array>()
            .unwrap();
        assert_eq!(int_col.values(), &[1000i32, 2000, 3000, 4000, 5000]);
        col_idx += 1;

        // Verify long column
        let long_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        assert_eq!(long_col.values(), &[10000i64, 20000, 30000, 40000, 50000]);
        col_idx += 1;

        // Verify float column
        let float_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Float32Array>()
            .unwrap();
        assert_eq!(float_col.values(), &[1.1f32, 2.2, 3.3, 4.4, 5.5]);
        col_idx += 1;

        // Verify double column
        let double_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Float64Array>()
            .unwrap();
        assert_eq!(double_col.values(), &[1.11f64, 2.22, 3.33, 4.44, 5.55]);
        col_idx += 1;

        // Verify bool column
        let bool_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<BooleanArray>()
            .unwrap();
        assert!(bool_col.value(0));
        assert!(!bool_col.value(1));
        col_idx += 1;

        // Verify string column
        let string_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(string_col.value(0), "a");
        assert_eq!(string_col.value(4), "e");
        col_idx += 1;

        // Verify binary column
        let binary_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<BinaryArray>()
            .unwrap();
        assert_eq!(binary_col.value(0), b"bin1");
        assert_eq!(binary_col.value(4), b"bin5");
        col_idx += 1;

        // Verify date column
        let date_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Date32Array>()
            .unwrap();
        assert_eq!(date_col.values(), &[18262, 18263, 18264, 18265, 18266]);
        col_idx += 1;

        // Verify timestamp column (with UTC timezone)
        let timestamp_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(timestamp_col.value(0), 1609459200000000i64);
        assert_eq!(timestamp_col.value(4), 1609804800000000i64);
        assert!(timestamp_col
            .timezone()
            .is_some_and(|tz| tz.eq_ignore_ascii_case("utc")));
        col_idx += 1;

        // Verify timestamp_ntz column (without timezone)
        let timestamp_ntz_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(timestamp_ntz_col.value(0), 1609459200000000i64);
        assert_eq!(timestamp_ntz_col.value(4), 1609804800000000i64);
        assert!(timestamp_ntz_col.timezone().is_none());
        col_idx += 1;

        // Verify decimal column
        let decimal_col = data[0]
            .column(col_idx)
            .as_any()
            .downcast_ref::<Decimal128Array>()
            .unwrap();
        assert_eq!(decimal_col.value(0), 12345i128);
        assert_eq!(decimal_col.value(4), 56789i128);
        assert_eq!(decimal_col.precision(), 10);
        assert_eq!(decimal_col.scale(), 2);
    }

    /// Test that field IDs are accessible via ColumnMetadataKey::ParquetFieldId as documented.
    ///
    /// Per trait definitions in lib.rs, field IDs should be accessible via
    /// StructField::get_config_value with ColumnMetadataKey::ParquetFieldId.
    #[test]
    fn test_parquet_footer_read_with_field_id() {
        // Write parquet file with field ID
        let field = Field::new("value", ArrowDataType::Int64, false).with_metadata(HashMap::from(
            [(PARQUET_FIELD_ID_META_KEY.to_string(), "42".to_string())],
        ));
        let arrow_schema = Arc::new(ArrowSchema::new(vec![field]));

        let temp_dir = tempfile::tempdir().unwrap();
        let file_path = temp_dir.path().join("field_id_test.parquet");
        let batch = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![Arc::new(Int64Array::from(vec![1, 2, 3]))],
        )
        .unwrap();

        let file = std::fs::File::create(&file_path).unwrap();
        let mut writer = ArrowWriter::try_new(file, arrow_schema, None).unwrap();
        writer.write(&batch).unwrap();
        writer.close().unwrap();

        // Read footer and verify field ID accessibility
        let store = Arc::new(LocalFileSystem::new());
        let handler = DefaultParquetHandler::new(store, Arc::new(TokioBackgroundExecutor::new()));
        let file_size = std::fs::metadata(&file_path).unwrap().len();
        let file_meta = FileMeta {
            location: Url::from_file_path(&file_path).unwrap(),
            last_modified: 0,
            size: file_size,
        };

        let footer = handler.read_parquet_footer(&file_meta).unwrap();
        let field = footer
            .schema
            .fields()
            .find(|f| f.name() == "value")
            .unwrap();

        // Field ID is transformed to kernel key when reading. arrow->kernel parses the
        // `PARQUET:field_id` string back into kernel's canonical `MetadataValue::Number(i64)`.
        assert_eq!(
            field
                .metadata()
                .get(ColumnMetadataKey::ParquetFieldId.as_ref()),
            Some(&MetadataValue::Number(42))
        );

        // Field ID should be accessible via documented API
        let field_id = field.get_config_value(&ColumnMetadataKey::ParquetFieldId)
            .expect("Field ID should be accessible via ColumnMetadataKey::ParquetFieldId per lib.rs:836-837");

        match field_id {
            delta_kernel::schema::MetadataValue::String(id) => assert_eq!(id, "42"),
            delta_kernel::schema::MetadataValue::Number(id) => assert_eq!(*id, 42),
            other => panic!("Expected String or Number, got {other:?}"),
        }
    }

    /// Test that columns are matched by field ID when column names differ.
    ///
    /// Per lib.rs:676-680, field IDs (via [`ColumnMetadataKey::ParquetFieldId`]) should take
    /// precedence over field names for column matching.
    ///
    /// [`ColumnMetadataKey::ParquetFieldId`]: delta_kernel::schema::ColumnMetadataKey::ParquetFieldId
    #[test]
    fn test_read_parquet_with_field_id_matching() {
        // Write parquet with field IDs using PARQUET_FIELD_ID_META_KEY (Parquet's native key)
        // The kernel will transform these to parquet.field.id when reading
        let fields = vec![
            Field::new("id", ArrowDataType::Int64, false).with_metadata(HashMap::from([(
                PARQUET_FIELD_ID_META_KEY.to_string(),
                "1".to_string(),
            )])),
            Field::new("name", ArrowDataType::Utf8, false).with_metadata(HashMap::from([(
                PARQUET_FIELD_ID_META_KEY.to_string(),
                "2".to_string(),
            )])),
        ];
        let arrow_schema = Arc::new(ArrowSchema::new(fields));

        let temp_dir = tempfile::tempdir().unwrap();
        let file_path = temp_dir.path().join("field_id_matching.parquet");
        let batch = RecordBatch::try_new(
            arrow_schema.clone(),
            vec![
                Arc::new(Int64Array::from(vec![1, 2, 3])),
                Arc::new(StringArray::from(vec!["alice", "bob", "charlie"])),
            ],
        )
        .unwrap();

        let file = std::fs::File::create(&file_path).unwrap();
        let mut writer = ArrowWriter::try_new(file, arrow_schema, None).unwrap();
        writer.write(&batch).unwrap();
        writer.close().unwrap();

        // Create kernel schema with DIFFERENT names but SAME field IDs
        let kernel_schema = schema_ref! {
            (StructField::not_null("user_id", delta_kernel::schema::DataType::LONG)
                    .with_metadata([(
                        ColumnMetadataKey::ParquetFieldId.as_ref(),
                        MetadataValue::Number(1),
                    )])),
            (StructField::not_null("user_name", delta_kernel::schema::DataType::STRING)
                    .with_metadata([(
                        ColumnMetadataKey::ParquetFieldId.as_ref(),
                        MetadataValue::Number(2),
                    )])),
        };

        // Read using kernel schema with different column names
        let store = Arc::new(LocalFileSystem::new());
        let handler = DefaultParquetHandler::new(store, Arc::new(TokioBackgroundExecutor::new()));
        let file_meta = FileMeta {
            location: Url::from_file_path(&file_path).unwrap(),
            last_modified: 0,
            size: std::fs::metadata(&file_path).unwrap().len(),
        };

        // Should successfully match by field ID despite different names
        let data: Vec<RecordBatch> = handler
            .read_parquet_files(slice::from_ref(&file_meta), kernel_schema, None)
            .unwrap()
            .map(into_record_batch)
            .try_collect()
            .unwrap();

        // Verify data was correctly matched by field ID
        assert_eq!(data.len(), 1);
        let batch = &data[0];

        // Verify columns were renamed to match the kernel schema (the names from the parquet
        // file's schema are discarded; the matching agreed on field IDs only).
        let schema = batch.schema();
        assert_eq!(schema.field(0).name(), "user_id");
        assert_eq!(schema.field(1).name(), "user_name");

        let id_col = batch
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        assert_eq!(id_col.values(), &[1, 2, 3], "Should match by field ID 1");

        let name_col = batch
            .column(1)
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        assert_eq!(name_col.value(0), "alice", "Should match by field ID 2");
        assert_eq!(name_col.value(1), "bob");
        assert_eq!(name_col.value(2), "charlie");
    }

    // Verifies that write_parquet (the internal stats-collecting path) does not embed the Arrow
    // IPC schema in the Parquet file metadata.
    #[tokio::test]
    async fn write_parquet_omits_arrow_schema_metadata() {
        let store = Arc::new(InMemory::new());
        let parquet_handler =
            DefaultParquetHandler::new(store.clone(), Arc::new(TokioBackgroundExecutor::new()));

        let data = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![(
                "a",
                Arc::new(Int64Array::from(vec![1, 2, 3])) as Arc<dyn Array>,
            )])
            .unwrap(),
        ));
        let physical_schema = long_schema("a");
        let metadata = parquet_handler
            .write_parquet(
                &Url::parse("memory:///data/").unwrap(),
                data,
                &[],
                &physical_schema,
            )
            .await
            .unwrap();

        let path = Path::from_url_path(metadata.file_meta.location.path()).unwrap();
        let reader = ParquetObjectReader::new(store, path);
        let builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap();
        let kv = builder.metadata().file_metadata().key_value_metadata();
        let has = kv
            .map(|kv| kv.iter().any(|e| e.key == ARROW_SCHEMA_META_KEY))
            .unwrap_or(false);
        assert!(
            !has,
            "Parquet file should not contain embedded Arrow schema metadata"
        );
    }

    #[tokio::test]
    async fn write_parquet_file_creates_parent_directories() {
        // GIVEN a file path whose parent directories do not exist
        let temp_dir = tempfile::tempdir().unwrap();
        let nested_path = temp_dir.path().join("a/b/c/output.parquet");
        assert!(!nested_path.parent().unwrap().exists());

        let store = Arc::new(LocalFileSystem::new());
        let parquet_handler: Arc<dyn ParquetHandler> = Arc::new(DefaultParquetHandler::new(
            store.clone(),
            Arc::new(TokioBackgroundExecutor::new()),
        ));

        let engine_data: Box<dyn EngineData> = Box::new(ArrowEngineData::new(
            RecordBatch::try_from_iter(vec![(
                "x",
                Arc::new(Int64Array::from(vec![1, 2, 3])) as Arc<dyn Array>,
            )])
            .unwrap(),
        ));
        let data_iter: DeltaResultIteratorStatic<Box<dyn EngineData>> =
            Box::new(std::iter::once(Ok(engine_data)));

        // WHEN we write a parquet file to that path
        let file_url = Url::from_file_path(&nested_path).unwrap();
        parquet_handler
            .write_parquet_file(file_url.clone(), data_iter)
            .unwrap();

        // THEN the file is created and contains the expected data
        assert!(nested_path.exists());

        let path = Path::from_url_path(file_url.path()).unwrap();
        let reader = ParquetObjectReader::new(store.clone(), path);
        let batches: Vec<RecordBatch> = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .unwrap()
            .build()
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(batches.len(), 1);
        let col = batches[0]
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        assert_eq!(col.values(), &[1, 2, 3]);
    }

    // === ParquetHandler contract tests ===
    //
    // These call the shared contract helpers in `engine::tests` against `DefaultParquetHandler`
    // (the matching `SyncParquetHandler` invocations live in `engine/sync/parquet.rs`).

    fn default_parquet_handler() -> DefaultParquetHandler<TokioBackgroundExecutor> {
        DefaultParquetHandler::new(
            Arc::new(LocalFileSystem::new()),
            Arc::new(TokioBackgroundExecutor::new()),
        )
    }

    #[test]
    fn parquet_handler_reads_footer() {
        let checkpoint = PathBuf::from(
            "../kernel/tests/data/parsed-stats/_delta_log/00000000000000000003.checkpoint.parquet",
        );
        test_parquet_handler_reads_footer(&default_parquet_handler(), &checkpoint);
    }

    #[test]
    fn parquet_handler_footer_errors_on_missing_file() {
        test_parquet_handler_footer_errors_on_missing_file(&default_parquet_handler());
    }

    #[test]
    fn parquet_handler_footer_preserves_field_ids() {
        test_parquet_handler_footer_preserves_field_ids(&default_parquet_handler());
    }

    #[test]
    fn parquet_handler_write_always_overwrites() {
        test_parquet_handler_write_always_overwrites(&default_parquet_handler());
    }

    #[test]
    fn parquet_handler_write_omits_arrow_schema() {
        test_parquet_handler_write_omits_arrow_schema(&default_parquet_handler());
    }

    #[test]
    fn parquet_handler_reads_file_with_arrow_schema() {
        test_parquet_handler_reads_file_with_arrow_schema(&default_parquet_handler());
    }
}