datafusion-ducklake 0.3.1

//! DuckLake table provider implementation

use std::any::Any;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use crate::Result;
use crate::column_rename::ColumnRenameExec;
use crate::delete_filter::DeleteFilterExec;
use crate::metadata_provider::{
    DuckLakeFileData, DuckLakeTableColumn, DuckLakeTableFile, MetadataProvider,
};
use crate::path_resolver::resolve_path;
use crate::positional_source::PositionalFileSource;
use crate::row_id::{
    FileRowNumberExec, ROW_ID_PARQUET_FIELD_ID, ROWID_COLUMN_NAME, RowIdExec, rowid_field,
};
use crate::types::{
    build_arrow_schema, build_read_schema_with_field_id_mapping, extract_parquet_field_ids,
};

#[cfg(feature = "write")]
use crate::insert_exec::DuckLakeInsertExec;
#[cfg(feature = "write")]
use crate::metadata_writer::{MetadataWriter, WriteMode};

#[cfg(feature = "encryption")]
use crate::encryption::EncryptionFactoryBuilder;
use arrow::array::{Array, Int64Array};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use async_trait::async_trait;
use datafusion::catalog::{Session, TableProvider};
use datafusion::common::Statistics;
use datafusion::common::stats::Precision;
use datafusion::datasource::listing::PartitionedFile;
use datafusion::datasource::physical_plan::parquet::{ParquetAccessPlan, RowGroupAccess};
use datafusion::datasource::physical_plan::{FileGroup, FileScanConfigBuilder, ParquetSource};
use datafusion::datasource::source::DataSourceExec;
use datafusion::error::{DataFusionError, Result as DataFusionResult};
use datafusion::execution::object_store::ObjectStoreUrl;
#[cfg(feature = "write")]
use datafusion::logical_expr::dml::InsertOp;
use datafusion::logical_expr::{Expr, TableProviderFilterPushDown, TableType};
use datafusion::physical_plan::ExecutionPlan;
use futures::StreamExt;
use object_store::path::Path as ObjectPath;
use parquet::arrow::ParquetRecordBatchStreamBuilder;
use parquet::arrow::async_reader::ParquetObjectReader;

#[cfg(feature = "encryption")]
use datafusion::execution::parquet_encryption::EncryptionFactory;

// Delete file schema constants (public for testing)
pub const DELETE_FILE_PATH_COL: &str = "file_path";
pub const DELETE_POS_COL: &str = "pos";

/// Validate and convert file_size_bytes from i64 (as stored in DuckLake metadata) to u64.
///
/// DuckLake stores file sizes as signed integers in SQL. A negative value indicates
/// corrupt or invalid metadata. Without this check, a negative i64 cast to u64 would
/// wrap to a huge value (e.g., -1 becomes u64::MAX), causing confusing downstream errors.
pub(crate) fn validated_file_size(file_size_bytes: i64, file_path: &str) -> DataFusionResult<u64> {
    u64::try_from(file_size_bytes).map_err(|_| {
        DataFusionError::Execution(format!(
            "Invalid file_size_bytes ({}) for file '{}': value must be non-negative",
            file_size_bytes, file_path
        ))
    })
}

/// Validate and convert record_count from i64 (as stored in DuckLake metadata) to u64.
///
/// DuckLake stores record counts as signed integers in SQL. A negative value indicates
/// corrupt or invalid metadata. Without this check, a negative record_count would cause
/// incorrect behavior (e.g., empty ranges in full-file deletes, or incorrect row filtering).
pub(crate) fn validated_record_count(record_count: i64, file_path: &str) -> DataFusionResult<u64> {
    u64::try_from(record_count).map_err(|_| {
        DataFusionError::Execution(format!(
            "Invalid record_count ({}) for file '{}': value must be non-negative",
            record_count, file_path
        ))
    })
}

/// Returns the expected schema for DuckLake delete files
///
/// Delete files have a standard schema: (file_path: VARCHAR, pos: INT64)
/// The file_path column is metadata/documentation only (for Iceberg compatibility).
/// The pos column contains the row positions to delete.
pub fn delete_file_schema() -> SchemaRef {
    Arc::new(Schema::new(vec![
        Field::new(DELETE_FILE_PATH_COL, DataType::Utf8, false),
        Field::new(DELETE_POS_COL, DataType::Int64, false),
    ]))
}

/// Cached schema mapping for renamed columns
type SchemaMapping = (SchemaRef, HashMap<String, String>);

/// Per-file read configuration computed for the row-lineage scan path.
///
/// Encapsulates the decision made by `DuckLakeMultiFileReader::GetVirtualColumnExpression`
/// in the C++ extension: either the parquet file embeds a row-id column
/// (UPDATE/compaction case — surviving rowids preserved across file rewrite),
/// or it doesn't (INSERT-only case — synthesize from `row_id_start + position`).
#[derive(Debug, Clone)]
struct FileReadConfig {
    /// Schema we pass to `ParquetSource::new` for this file. When
    /// `embedded_rowid_parquet_name` is `Some`, this schema has the embedded
    /// rowid column appended at the end (under its parquet name).
    read_schema: SchemaRef,
    /// Parquet-name → user-facing-name renames. Includes the rowid rename
    /// (parquet column → `"rowid"`) when the file has an embedded column with
    /// a different name.
    name_mapping: HashMap<String, String>,
    /// `Some(parquet_column_name)` if the file embeds the rowid column
    /// (tagged with [`ROW_ID_PARQUET_FIELD_ID`]); `None` otherwise.
    embedded_rowid_parquet_name: Option<String>,
    /// Per-row-group starting physical row position (prefix sums of
    /// `row_groups[i].num_rows()`). `row_group_starts[i]` is the 0-based file
    /// position of the first row of row group `i`. Used to build row-group-
    /// aligned scan partitions whose starting position is known at plan time,
    /// so `FileRowNumberExec` can synthesize true physical positions instead of
    /// counting stream arrivals. The Parquet footer is the source of truth; the
    /// catalog does not store per-row-group counts.
    row_group_starts: Vec<i64>,
    /// Number of row groups in the file (`row_group_starts.len()`). Required to
    /// build a `ParquetAccessPlan` of the correct length.
    row_group_count: usize,
}

/// DuckLake table provider
///
/// Represents a table within a DuckLake schema and provides access to data via Parquet files.
/// Caches snapshot_id and uses it to load all metadata atomically.
pub struct DuckLakeTable {
    #[allow(dead_code)]
    table_id: i64,
    table_name: String,
    #[allow(dead_code)]
    provider: Arc<dyn MetadataProvider>,
    /// Object store URL for resolving file paths (e.g., s3://bucket/ or file:///)
    object_store_url: Arc<ObjectStoreUrl>,
    /// Table path for resolving relative file paths
    table_path: String,
    /// User-facing schema. Equals `physical_schema` when row lineage is off, or
    /// `physical_schema` with a `rowid` BIGINT appended at the end when on.
    schema: SchemaRef,
    /// Schema of the physical (parquet-backed) columns only — no rowid.
    physical_schema: SchemaRef,
    /// When true, `schema` includes a trailing `rowid` column and `scan()`
    /// injects it per-file via [`RowIdExec`].
    row_lineage: bool,
    /// Column metadata from DuckLake (needed for field_id mapping)
    columns: Vec<DuckLakeTableColumn>,
    /// Table files with paths as stored in metadata (resolved on-the-fly when needed)
    table_files: Vec<DuckLakeTableFile>,
    /// Per-file row-lineage read config, populated lazily on the rowid scan
    /// path. Each file requires its own parquet metadata read to detect an
    /// embedded `_ducklake_internal_row_id` column; we memoize so repeated
    /// scans don't re-fetch.
    file_read_config_cache: std::sync::Mutex<HashMap<String, Arc<FileReadConfig>>>,
    /// Encryption factory for decrypting encrypted Parquet files (when encryption feature is enabled)
    #[cfg(feature = "encryption")]
    encryption_factory: Option<Arc<dyn EncryptionFactory>>,
    /// Schema name (needed for write operations)
    #[cfg(feature = "write")]
    schema_name: Option<String>,
    /// Metadata writer for write operations (when write feature is enabled)
    #[cfg(feature = "write")]
    writer: Option<Arc<dyn MetadataWriter>>,
}

impl std::fmt::Debug for DuckLakeTable {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DuckLakeTable")
            .field("table_id", &self.table_id)
            .field("table_name", &self.table_name)
            .field("table_path", &self.table_path)
            .field("schema", &self.schema)
            .field("columns", &self.columns)
            .field("table_files", &self.table_files)
            .finish_non_exhaustive()
    }
}

impl DuckLakeTable {
    /// Create a new DuckLake table
    pub fn new(
        table_id: i64,
        table_name: impl Into<String>,
        provider: Arc<dyn MetadataProvider>,
        snapshot_id: i64, // Received from schema
        object_store_url: Arc<ObjectStoreUrl>,
        table_path: String,
    ) -> Result<Self> {
        // Load ALL metadata with this snapshot_id
        let columns = provider.get_table_structure(table_id)?;
        let physical_schema = Arc::new(build_arrow_schema(&columns)?);
        let schema = physical_schema.clone();
        let table_files = provider.get_table_files_for_select(table_id, snapshot_id)?;

        // Build encryption factory from file encryption keys (when encryption feature is enabled)
        #[cfg(feature = "encryption")]
        let encryption_factory = {
            let mut builder = EncryptionFactoryBuilder::new();
            for table_file in &table_files {
                // Resolve the file path for the mapping
                let resolved_path = resolve_path(
                    &table_path,
                    &table_file.file.path,
                    table_file.file.path_is_relative,
                )?;
                builder.add_file(&resolved_path, table_file.file.encryption_key.as_deref());

                // Also add delete file encryption key if present
                if let Some(ref delete_file) = table_file.delete_file {
                    let resolved_delete_path =
                        resolve_path(&table_path, &delete_file.path, delete_file.path_is_relative)?;
                    builder.add_file(&resolved_delete_path, delete_file.encryption_key.as_deref());
                }
            }
            let factory = builder.build();
            if factory.has_encrypted_files() {
                Some(Arc::new(factory) as Arc<dyn EncryptionFactory>)
            } else {
                None
            }
        };

        Ok(Self {
            table_id,
            table_name: table_name.into(),
            provider,
            object_store_url,
            table_path,
            schema,
            physical_schema,
            row_lineage: false,
            columns,
            table_files,
            #[cfg(feature = "encryption")]
            encryption_factory,
            file_read_config_cache: std::sync::Mutex::new(HashMap::new()),
            #[cfg(feature = "write")]
            schema_name: None,
            #[cfg(feature = "write")]
            writer: None,
        })
    }

    /// Enable / disable the row-lineage feature. When enabled, the table's
    /// public schema includes a trailing `rowid` BIGINT column synthesized
    /// from each row's catalog-recorded `row_id_start + position_in_file`.
    pub fn with_row_lineage(mut self, enabled: bool) -> Self {
        self.row_lineage = enabled;
        self.schema = if enabled {
            let mut fields: Vec<Arc<Field>> =
                self.physical_schema.fields().iter().cloned().collect();
            fields.push(Arc::new(rowid_field()));
            Arc::new(Schema::new(fields))
        } else {
            self.physical_schema.clone()
        };
        self
    }

    /// Index of the synthetic `rowid` column in `self.schema`, when enabled.
    fn rowid_index(&self) -> Option<usize> {
        self.row_lineage
            .then(|| self.physical_schema.fields().len())
    }

    /// Resolve a file path (data or delete file) to its absolute path
    fn resolve_file_path(&self, file: &DuckLakeFileData) -> DataFusionResult<String> {
        resolve_path(&self.table_path, &file.path, file.path_is_relative)
            .map_err(|e| DataFusionError::External(Box::new(e)))
    }

    /// Create a ParquetSource with encryption support if enabled and needed
    fn create_parquet_source(&self, schema: SchemaRef) -> ParquetSource {
        #[cfg(feature = "encryption")]
        if let Some(ref factory) = self.encryption_factory {
            return ParquetSource::new(schema).with_encryption_factory(Arc::clone(factory));
        }
        ParquetSource::new(schema)
    }

    /// Compute the field_id -> physical-name read schema and rename mapping for a
    /// SINGLE file. Physical column names can differ across files (e.g. a column
    /// renamed after some files were written), so this is resolved per file.
    async fn file_schema_mapping(
        &self,
        state: &dyn Session,
        file: &DuckLakeFileData,
    ) -> DataFusionResult<SchemaMapping> {
        let resolved_path = self.resolve_file_path(file)?;
        let object_store = state
            .runtime_env()
            .object_store(self.object_store_url.as_ref())?;
        let object_path = ObjectPath::from(resolved_path.as_str());

        let reader = ParquetObjectReader::new(object_store, object_path);

        // Build the ParquetRecordBatchStreamBuilder with decryption if needed
        #[cfg(feature = "encryption")]
        let builder = {
            use parquet::arrow::arrow_reader::ArrowReaderOptions;

            // Check if file has encryption key
            let options = if let Some(ref key) = file.encryption_key {
                if !key.is_empty() {
                    let key_bytes = crate::encryption::DuckLakeEncryptionFactory::decode_key(key)?;
                    let decryption_props =
                        parquet::encryption::decrypt::FileDecryptionProperties::builder(key_bytes)
                            .build()
                            .map_err(|e| {
                                DataFusionError::Execution(format!(
                                    "Failed to create decryption properties: {}",
                                    e
                                ))
                            })?;
                    ArrowReaderOptions::new().with_file_decryption_properties(decryption_props)
                } else {
                    ArrowReaderOptions::new()
                }
            } else {
                ArrowReaderOptions::new()
            };

            ParquetRecordBatchStreamBuilder::new_with_options(reader, options)
                .await
                .map_err(|e| DataFusionError::External(Box::new(e)))?
        };

        #[cfg(not(feature = "encryption"))]
        let builder = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .map_err(|e| DataFusionError::External(Box::new(e)))?;

        let field_id_map = extract_parquet_field_ids(builder.metadata());

        // No field_ids means external file - use current schema directly
        if field_id_map.is_empty() {
            return Ok((self.schema.clone(), HashMap::new()));
        }

        let (read_schema, name_mapping) = build_read_schema_with_field_id_mapping(
            &self.columns,
            &field_id_map,
            Some(builder.schema().as_ref()),
        )
        .map_err(|e| DataFusionError::External(Box::new(e)))?;

        Ok((Arc::new(read_schema), name_mapping))
    }

    /// Read a delete file and extract all deleted row positions
    ///
    /// The delete file is already associated with a specific data file via metadata.
    /// We only need to extract the "pos" column - the "file_path" column is
    /// metadata/documentation only (for Iceberg compatibility).
    async fn read_delete_file_positions(
        &self,
        state: &dyn Session,
        delete_file: &DuckLakeFileData,
    ) -> DataFusionResult<HashSet<i64>> {
        // Get the standard delete file schema
        let delete_schema = delete_file_schema();

        // Resolve the delete file path
        let resolved_delete_path = self.resolve_file_path(delete_file)?;

        // Create PartitionedFile with footer size hint if available
        let mut pf = PartitionedFile::new(
            &resolved_delete_path,
            validated_file_size(delete_file.file_size_bytes, &resolved_delete_path)?,
        );
        if let Some(footer_size) = delete_file.footer_size
            && footer_size > 0
            && let Ok(hint) = usize::try_from(footer_size)
        {
            pf = pf.with_metadata_size_hint(hint);
        }

        // Create file scan config for the delete file
        let file_scan_config = FileScanConfigBuilder::new(
            self.object_store_url.as_ref().clone(),
            Arc::new(self.create_parquet_source(delete_schema)),
        )
        .with_file_group(FileGroup::new(vec![pf]))
        .build();

        // Use DataSourceExec directly to preserve our ParquetSource with encryption factory
        let exec = DataSourceExec::from_data_source(file_scan_config);

        // Execute and collect all batches
        let task_ctx = state.task_ctx();
        let stream = exec.execute(0, task_ctx)?;

        let batches: Vec<RecordBatch> = stream
            .collect::<Vec<_>>()
            .await
            .into_iter()
            .collect::<DataFusionResult<Vec<_>>>()
            .map_err(|e| {
                if is_object_store_not_found(&e) {
                    DataFusionError::Execution(format!(
                        "Delete file '{}' referenced in catalog metadata was not found. This may indicate catalog corruption or that the file was deleted outside of DuckLake.",
                        resolved_delete_path
                    ))
                } else {
                    e
                }
            })?;

        // Extract all positions from all batches
        let mut positions = HashSet::new();
        for batch in batches {
            extract_deleted_positions_from_batch(&batch, &mut positions)?;
        }

        Ok(positions)
    }

    /// Build a single execution plan for all files without delete files
    ///
    /// Groups multiple files into a single efficient execution plan since they don't
    /// need delete filtering.
    async fn build_exec_for_files_without_deletes(
        &self,
        state: &dyn Session,
        files: &[&DuckLakeTableFile],
        projection: Option<&Vec<usize>>,
        limit: Option<usize>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        // Physical column names can differ across files (e.g. a column renamed
        // after some files were written), so the field_id -> physical-name read
        // schema must be resolved PER FILE. Group files that share the same
        // physical schema into one ParquetSource and union the groups; the common
        // case (no schema evolution) stays a single group / single scan.
        let mut groups: Vec<(SchemaMapping, Vec<PartitionedFile>)> = Vec::new();
        let mut group_index: HashMap<String, usize> = HashMap::new();

        for table_file in files {
            let mapping = self.file_schema_mapping(state, &table_file.file).await?;

            let resolved_path = self.resolve_file_path(&table_file.file)?;
            let mut pf = PartitionedFile::new(
                &resolved_path,
                validated_file_size(table_file.file.file_size_bytes, &resolved_path)?,
            );
            // Footer size hint cuts I/O from 2 reads to 1 per file (helps S3/MinIO).
            if let Some(footer_size) = table_file.file.footer_size
                && footer_size > 0
                && let Ok(hint) = usize::try_from(footer_size)
            {
                pf = pf.with_metadata_size_hint(hint);
            }

            // Group key: physical field names + types, then the rename mapping.
            let (read_schema, name_mapping) = &mapping;
            let mut key = String::new();
            for f in read_schema.fields() {
                key.push_str(f.name());
                key.push('\u{1}');
                key.push_str(&format!("{:?}", f.data_type()));
                key.push('\u{2}');
            }
            let mut pairs: Vec<(&String, &String)> = name_mapping.iter().collect();
            pairs.sort();
            for (k, v) in pairs {
                key.push_str(k);
                key.push('\u{3}');
                key.push_str(v);
                key.push('\u{4}');
            }

            match group_index.get(&key) {
                Some(&gi) => groups[gi].1.push(pf),
                None => {
                    group_index.insert(key, groups.len());
                    groups.push((mapping, vec![pf]));
                },
            }
        }

        let output_schema = match projection {
            Some(indices) => Arc::new(self.schema.project(indices)?),
            None => self.schema.clone(),
        };

        // Build one scan per physical-schema group; ColumnRenameExec coerces each
        // group to the catalog schema (renamed columns or a differing Arrow type).
        let mut execs: Vec<Arc<dyn ExecutionPlan>> = Vec::with_capacity(groups.len());
        for ((read_schema, name_mapping), partitioned_files) in groups {
            let mut builder = FileScanConfigBuilder::new(
                self.object_store_url.as_ref().clone(),
                Arc::new(self.create_parquet_source(read_schema.clone())),
            )
            .with_limit(limit)
            .with_file_group(FileGroup::new(partitioned_files));

            if let Some(proj) = projection {
                builder = builder.with_projection_indices(Some(proj.clone()))?;
            }

            let parquet_exec: Arc<dyn ExecutionPlan> =
                DataSourceExec::from_data_source(builder.build());

            let exec = if !name_mapping.is_empty() || parquet_exec.schema() != output_schema {
                Arc::new(ColumnRenameExec::new(
                    parquet_exec,
                    output_schema.clone(),
                    name_mapping,
                )) as Arc<dyn ExecutionPlan>
            } else {
                parquet_exec
            };
            execs.push(exec);
        }

        combine_execution_plans(execs)
    }

    /// Configure this table for write operations.
    ///
    /// This method enables write support by attaching a metadata writer and data path.
    /// Once configured, the table can handle INSERT INTO operations.
    ///
    /// # Arguments
    /// * `schema_name` - Name of the schema this table belongs to
    /// * `writer` - Metadata writer for catalog operations
    #[cfg(feature = "write")]
    pub fn with_writer(mut self, schema_name: String, writer: Arc<dyn MetadataWriter>) -> Self {
        self.schema_name = Some(schema_name);
        self.writer = Some(writer);
        self
    }

    /// Build an execution plan for a single file with delete filtering
    ///
    /// Creates a Parquet scan wrapped with a delete filter to exclude deleted rows.
    async fn build_exec_for_file_with_deletes(
        &self,
        state: &dyn Session,
        table_file: &DuckLakeTableFile,
        projection: Option<&Vec<usize>>,
        limit: Option<usize>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let file_cfg = self.build_file_read_config(state, &table_file.file).await?;

        // Deletes filter by physical row position, so this is a positional path:
        // it must read the file in row-group-aligned, non-repartitionable,
        // non-pruning partitions and synthesize positions before filtering.
        let deleted_positions = if let Some(ref delete_file) = table_file.delete_file {
            let p = self.read_delete_file_positions(state, delete_file).await?;
            (!p.is_empty()).then_some(p)
        } else {
            None
        };

        let output_schema = match projection {
            Some(indices) => Arc::new(self.schema.project(indices)?),
            None => self.schema.clone(),
        };

        // Explicit parquet projection over `read_schema`. rowid is never
        // projected on this path, so always read only the physical columns —
        // for an embedded-rowid file, `read_schema` has a trailing embedded
        // column we must NOT read here. With `projection = None` that means the
        // physical columns `0..physical_len` (not "all of read_schema").
        let proj_indices: Vec<usize> = match projection {
            Some(indices) => indices.clone(),
            None => (0..self.physical_schema.fields().len()).collect(),
        };

        let exec_after_delete: Arc<dyn ExecutionPlan> = if let Some(positions) = deleted_positions {
            // Positional path: no scan-level limit (would drop rows before the
            // delete filter); DataFusion enforces LIMIT above the table plan.
            let target_partitions = state.config().target_partitions();
            let (file_groups, partition_starts) =
                self.build_row_group_partitions(&table_file.file, &file_cfg, target_partitions)?;

            let source = PositionalFileSource::wrap(Arc::new(
                self.create_parquet_source(file_cfg.read_schema.clone()),
            ));
            let mut builder =
                FileScanConfigBuilder::new(self.object_store_url.as_ref().clone(), source)
                    .with_file_groups(file_groups);
            builder = builder.with_projection_indices(Some(proj_indices.clone()))?;
            let scan = DataSourceExec::from_data_source(builder.build());

            let with_pos: Arc<dyn ExecutionPlan> =
                Arc::new(FileRowNumberExec::new(scan, partition_starts));
            Arc::new(DeleteFilterExec::try_new(
                with_pos,
                table_file.file.path.clone(),
                Arc::new(positions),
            )?)
        } else {
            // No actual deletes for this file: plain scan, scan-level limit OK.
            let resolved_path = self.resolve_file_path(&table_file.file)?;
            let mut pf = PartitionedFile::new(
                &resolved_path,
                validated_file_size(table_file.file.file_size_bytes, &resolved_path)?,
            );
            if let Some(footer_size) = table_file.file.footer_size
                && footer_size > 0
                && let Ok(hint) = usize::try_from(footer_size)
            {
                pf = pf.with_metadata_size_hint(hint);
            }
            let mut builder = FileScanConfigBuilder::new(
                self.object_store_url.as_ref().clone(),
                Arc::new(self.create_parquet_source(file_cfg.read_schema.clone())),
            )
            .with_limit(limit)
            .with_file_group(FileGroup::new(vec![pf]));
            builder = builder.with_projection_indices(Some(proj_indices.clone()))?;
            DataSourceExec::from_data_source(builder.build())
        };

        // ColumnRenameExec presents the catalog schema and, on the positional
        // path, drops the internal `__ducklake_row_pos` column (by name).
        if !file_cfg.name_mapping.is_empty() || exec_after_delete.schema() != output_schema {
            Ok(Arc::new(ColumnRenameExec::new(
                exec_after_delete,
                output_schema,
                file_cfg.name_mapping.clone(),
            )))
        } else {
            Ok(exec_after_delete)
        }
    }

    /// Inspect a single file's parquet metadata for the row-lineage scan
    /// path. Mirrors the per-file logic in `DuckLakeMultiFileReader::
    /// GetVirtualColumnExpression` (ducklake C++): if the file embeds a
    /// column tagged with [`ROW_ID_PARQUET_FIELD_ID`], project that column;
    /// otherwise synthesize rowid from `row_id_start + position`.
    async fn build_file_read_config(
        &self,
        state: &dyn Session,
        file: &DuckLakeFileData,
    ) -> DataFusionResult<Arc<FileReadConfig>> {
        let resolved_path = self.resolve_file_path(file)?;

        {
            let cache = self.file_read_config_cache.lock().unwrap();
            if let Some(cfg) = cache.get(&resolved_path) {
                return Ok(cfg.clone());
            }
        }

        let object_store = state
            .runtime_env()
            .object_store(self.object_store_url.as_ref())?;
        let object_path = ObjectPath::from(resolved_path.as_str());
        let reader = ParquetObjectReader::new(object_store, object_path);

        #[cfg(feature = "encryption")]
        let builder = {
            use parquet::arrow::arrow_reader::ArrowReaderOptions;
            let options = if let Some(ref key) = file.encryption_key {
                if !key.is_empty() {
                    let key_bytes = crate::encryption::DuckLakeEncryptionFactory::decode_key(key)?;
                    let decryption_props =
                        parquet::encryption::decrypt::FileDecryptionProperties::builder(key_bytes)
                            .build()
                            .map_err(|e| {
                                DataFusionError::Execution(format!(
                                    "Failed to create decryption properties: {}",
                                    e
                                ))
                            })?;
                    ArrowReaderOptions::new().with_file_decryption_properties(decryption_props)
                } else {
                    ArrowReaderOptions::new()
                }
            } else {
                ArrowReaderOptions::new()
            };
            ParquetRecordBatchStreamBuilder::new_with_options(reader, options)
                .await
                .map_err(|e| DataFusionError::External(Box::new(e)))?
        };

        #[cfg(not(feature = "encryption"))]
        let builder = ParquetRecordBatchStreamBuilder::new(reader)
            .await
            .map_err(|e| DataFusionError::External(Box::new(e)))?;

        let field_id_map = extract_parquet_field_ids(builder.metadata());

        // Per-row-group starting positions (prefix sums of num_rows), read from
        // the footer we already have open. Drives row-group-aligned scan
        // partitioning on positional paths.
        let row_groups = builder.metadata().row_groups();
        let row_group_count = row_groups.len();
        let mut row_group_starts = Vec::with_capacity(row_group_count);
        let mut row_acc: i64 = 0;
        for rg in row_groups {
            row_group_starts.push(row_acc);
            row_acc = row_acc.saturating_add(rg.num_rows());
        }

        // Standard read_schema + name_mapping for physical columns.
        let (physical_read_schema, mut name_mapping) = if field_id_map.is_empty() {
            (self.physical_schema.as_ref().clone(), HashMap::new())
        } else {
            let (s, m) = build_read_schema_with_field_id_mapping(
                &self.columns,
                &field_id_map,
                Some(builder.schema().as_ref()),
            )
            .map_err(|e| DataFusionError::External(Box::new(e)))?;
            (s, m)
        };

        // Detect the embedded rowid column by reserved field-id.
        let embedded_rowid_parquet_name = field_id_map.get(&ROW_ID_PARQUET_FIELD_ID).cloned();

        let read_schema = if let Some(ref parquet_name) = embedded_rowid_parquet_name {
            // Append the embedded rowid column to read_schema under its
            // parquet name; ParquetExec will project it by name from the
            // file. We add a `parquet_name → "rowid"` rename so the user
            // sees the column as `rowid` (only needed if the names differ).
            let mut fields: Vec<Arc<Field>> =
                physical_read_schema.fields().iter().cloned().collect();
            fields.push(Arc::new(Field::new(
                parquet_name.clone(),
                DataType::Int64,
                true,
            )));
            if parquet_name != ROWID_COLUMN_NAME {
                name_mapping.insert(parquet_name.clone(), ROWID_COLUMN_NAME.to_string());
            }
            Arc::new(Schema::new(fields))
        } else {
            Arc::new(physical_read_schema)
        };

        let cfg = Arc::new(FileReadConfig {
            read_schema,
            name_mapping,
            embedded_rowid_parquet_name,
            row_group_starts,
            row_group_count,
        });

        {
            let mut cache = self.file_read_config_cache.lock().unwrap();
            cache.entry(resolved_path).or_insert_with(|| cfg.clone());
        }

        Ok(cfg)
    }

    /// Build row-group-aligned scan partitions for a single file on a
    /// *positional* path (rowid synthesis and/or delete filtering).
    ///
    /// Returns one [`FileGroup`] per contiguous run of row groups (so each is a
    /// distinct DataFusion partition) together with a `partition_starts` vector
    /// whose `i`-th entry is the **true physical row position of the first row**
    /// of `file_groups[i]`. The two vectors are 1:1; `FileRowNumberExec` uses
    /// `partition_starts[partition]` to seed positions.
    ///
    /// Each chunk carries a whole-row-group `Scan`/`Skip` [`ParquetAccessPlan`]
    /// (never a `RowSelection`), so within a partition the reader emits a
    /// complete, contiguous, in-order run of physical rows. A single chunk
    /// (`target_partitions == 1`, or a file with ≤1 row group) carries no access
    /// plan and reads the whole file in order — identical to the legacy path.
    fn build_row_group_partitions(
        &self,
        file: &DuckLakeFileData,
        read_cfg: &FileReadConfig,
        target_partitions: usize,
    ) -> DataFusionResult<(Vec<FileGroup>, Vec<i64>)> {
        let resolved_path = self.resolve_file_path(file)?;
        let file_size = validated_file_size(file.file_size_bytes, &resolved_path)?;
        let footer_hint = file
            .footer_size
            .filter(|&s| s > 0)
            .and_then(|s| usize::try_from(s).ok());

        let make_pf = |access: Option<ParquetAccessPlan>| {
            let mut pf = PartitionedFile::new(&resolved_path, file_size);
            if let Some(hint) = footer_hint {
                pf = pf.with_metadata_size_hint(hint);
            }
            if let Some(plan) = access {
                pf = pf.with_extensions(Arc::new(plan));
            }
            pf
        };

        let n = read_cfg.row_group_count;
        let k = target_partitions.max(1).min(n.max(1));

        // Single partition: whole file, in order, no access plan. Covers
        // target_partitions == 1 and files with 0 or 1 row groups.
        if k <= 1 {
            return Ok((vec![FileGroup::new(vec![make_pf(None)])], vec![0]));
        }

        // Split the n row groups into k contiguous chunks as evenly as possible
        // (row groups are written near-uniform, so group-count balancing closely
        // tracks row-count balancing). The first `rem` chunks get one extra group.
        let base = n / k;
        let rem = n % k;
        let mut file_groups = Vec::with_capacity(k);
        let mut partition_starts = Vec::with_capacity(k);
        let mut a = 0usize;
        for chunk in 0..k {
            let len = base + usize::from(chunk < rem);
            let b = a + len;
            debug_assert!(b <= n && len > 0);

            let row_groups: Vec<RowGroupAccess> = (0..n)
                .map(|rg| {
                    if rg >= a && rg < b {
                        RowGroupAccess::Scan
                    } else {
                        RowGroupAccess::Skip
                    }
                })
                .collect();

            file_groups.push(FileGroup::new(vec![make_pf(Some(ParquetAccessPlan::new(
                row_groups,
            )))]));
            partition_starts.push(read_cfg.row_group_starts[a]);
            a = b;
        }
        debug_assert_eq!(a, n);

        Ok((file_groups, partition_starts))
    }

    /// Build a plan for a single file when the synthetic `rowid` column is in
    /// the projection. Always uses per-file scans because each file may have a
    /// different layout (embedded rowid vs. synthesized) and a distinct
    /// `row_id_start`.
    ///
    /// Order on the positional path (non-embedded, or any file with deletes):
    ///   DataSourceExec → FileRowNumberExec → DeleteFilterExec(?) → RowIdExec(?)
    ///   → ColumnRenameExec. Embedded-rowid files with no deletes keep a plain
    ///   DataSourceExec → ColumnRenameExec (rowid read from the file).
    async fn build_exec_for_file_with_rowid(
        &self,
        state: &dyn Session,
        table_file: &DuckLakeTableFile,
        user_proj: &[usize],
        rowid_idx: usize,
        limit: Option<usize>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let file_cfg = self.build_file_read_config(state, &table_file.file).await?;
        let has_embedded = file_cfg.embedded_rowid_parquet_name.is_some();

        // Physical columns to read (everything the user asked for except rowid).
        let physical_proj: Vec<usize> = user_proj
            .iter()
            .filter(|&&i| i != rowid_idx)
            .copied()
            .collect();

        // Match the C++ extension: if the file embeds no rowid column AND the
        // catalog didn't record a `row_id_start`, lineage cannot be
        // reconstructed. Hard-error rather than silently emit NULL/garbage.
        if !has_embedded && table_file.row_id_start.is_none() {
            return Err(DataFusionError::Execution(format!(
                "File \"{}\" has no embedded `_ducklake_internal_row_id` column and no \
                 `row_id_start` set in the catalog — row lineage cannot be reconstructed",
                table_file.file.path
            )));
        }

        // Resolve deletes once.
        let deleted_positions = if let Some(ref delete_file) = table_file.delete_file {
            let p = self.read_delete_file_positions(state, delete_file).await?;
            (!p.is_empty()).then_some(p)
        } else {
            None
        };
        let has_deletes = deleted_positions.is_some();

        // We need synthesized physical positions when rowid must be synthesized
        // (non-embedded) or when positional deletes must be applied. Embedded-
        // rowid files with no deletes keep the legacy plain scan (rowid read from
        // the file; reader-side pruning and scan-level limit are safe there).
        let needs_position = !has_embedded || has_deletes;

        // Parquet read projection. For embedded files, also read the embedded
        // rowid column; `ColumnRenameExec` later maps it to `rowid` by name, so
        // its position in the read projection is irrelevant.
        let parquet_projection: Vec<usize> = if has_embedded {
            let rowid_col_in_read_schema = file_cfg.read_schema.fields().len() - 1;
            let mut p = physical_proj.clone();
            p.push(rowid_col_in_read_schema);
            p
        } else {
            physical_proj.clone()
        };

        let after_deletes: Arc<dyn ExecutionPlan> = if needs_position {
            // Positional path: row-group-aligned partitions + a non-repartition,
            // non-pruning source, so each partition emits a complete, contiguous,
            // in-order run of physical rows. No scan-level limit (it would drop
            // rows before delete filtering); DataFusion enforces LIMIT above.
            let target_partitions = state.config().target_partitions();
            let (file_groups, partition_starts) =
                self.build_row_group_partitions(&table_file.file, &file_cfg, target_partitions)?;

            let source = PositionalFileSource::wrap(Arc::new(
                self.create_parquet_source(file_cfg.read_schema.clone()),
            ));
            let mut builder =
                FileScanConfigBuilder::new(self.object_store_url.as_ref().clone(), source)
                    .with_file_groups(file_groups);
            builder = builder.with_projection_indices(Some(parquet_projection))?;
            let scan = DataSourceExec::from_data_source(builder.build());

            // Materialize the physical position, then (optionally) filter deletes
            // by it, then (for non-embedded files) synthesize rowid from it.
            let mut plan: Arc<dyn ExecutionPlan> =
                Arc::new(FileRowNumberExec::new(scan, partition_starts));
            if let Some(p) = deleted_positions {
                plan = Arc::new(DeleteFilterExec::try_new(
                    plan,
                    table_file.file.path.clone(),
                    Arc::new(p),
                )?);
            }
            if !has_embedded {
                plan = Arc::new(RowIdExec::try_new(plan, table_file.row_id_start)?);
            }
            plan
        } else {
            // Embedded rowid, no deletes: legacy plain scan (cardinality-
            // preserving). Keep scan-level limit and reader pruning.
            let resolved_path = self.resolve_file_path(&table_file.file)?;
            let mut pf = PartitionedFile::new(
                &resolved_path,
                validated_file_size(table_file.file.file_size_bytes, &resolved_path)?,
            );
            if let Some(footer_size) = table_file.file.footer_size
                && footer_size > 0
                && let Ok(hint) = usize::try_from(footer_size)
            {
                pf = pf.with_metadata_size_hint(hint);
            }
            let mut builder = FileScanConfigBuilder::new(
                self.object_store_url.as_ref().clone(),
                Arc::new(self.create_parquet_source(file_cfg.read_schema.clone())),
            )
            .with_limit(limit)
            .with_file_group(FileGroup::new(vec![pf]));
            builder = builder.with_projection_indices(Some(parquet_projection))?;
            DataSourceExec::from_data_source(builder.build())
        };

        // Wrap with ColumnRenameExec to present the catalog schema. Required when
        // a physical column was renamed in the catalog, when the embedded rowid
        // column's parquet name differs from `"rowid"` (the common case — it's
        // `_ducklake_internal_row_id`), or when the file's physical Arrow type
        // differs from the catalog type (e.g. a DuckDB ARRAY read as
        // FixedSizeList vs the catalog's List). Coerces each column to
        // `output_schema`.
        let output_schema = self.output_schema_for_projection(user_proj, rowid_idx);
        if !file_cfg.name_mapping.is_empty() || after_deletes.schema() != output_schema {
            Ok(Arc::new(ColumnRenameExec::new(
                after_deletes,
                output_schema,
                file_cfg.name_mapping.clone(),
            )))
        } else {
            Ok(after_deletes)
        }
    }

    /// Output schema for the rowid-projected per-file plan: physical fields
    /// (using their user-facing renamed names from `self.schema`) interleaved
    /// with the synthetic `rowid` field at `rowid_idx`.
    fn output_schema_for_projection(&self, user_proj: &[usize], rowid_idx: usize) -> SchemaRef {
        let mut fields: Vec<Arc<Field>> = Vec::with_capacity(user_proj.len());
        for &i in user_proj {
            if i == rowid_idx {
                fields.push(Arc::new(rowid_field()));
            } else {
                fields.push(self.schema.fields()[i].clone());
            }
        }
        Arc::new(Schema::new(fields))
    }
}

#[async_trait]
impl TableProvider for DuckLakeTable {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn schema(&self) -> SchemaRef {
        Arc::clone(&self.schema)
    }

    fn table_type(&self) -> TableType {
        TableType::Base
    }

    fn statistics(&self) -> Option<Statistics> {
        // Aggregate per-file byte sizes from the cached `table_files`. Mirrors
        // DuckLake's own `ducklake_table_info` aggregate exactly:
        //
        //     total_byte_size == SUM(data_file.file_size_bytes)
        //                       - SUM(delete_file.file_size_bytes)
        //
        // The values come from the ducklake catalog, so this is the same
        // source of truth `ducklake_table_info` uses — no extra round trips
        // and the numbers will match byte-for-byte.
        //
        // Marked `Precision::Inexact` because DataFusion documents
        // `total_byte_size` as the *uncompressed Arrow output* size, while
        // the catalog tracks *compressed parquet* bytes. For wide
        // column types (List(Float64) embeddings) the two are nearly
        // identical; for narrow scalar schemas the on-disk number is 3-5x
        // smaller than Arrow output. Reporting compressed bytes Inexact
        // gives consumers a useful lower-bound estimate without misleading
        // the optimiser into thinking it's exact Arrow size. When
        // `record_count` is plumbed through `DuckLakeFileData`, a follow-up
        // can populate `num_rows` and use `calculate_total_byte_size` for a
        // closer Arrow-side estimate.
        let data_bytes: i64 = self
            .table_files
            .iter()
            .map(|f| f.file.file_size_bytes)
            .sum();
        let delete_bytes: i64 = self
            .table_files
            .iter()
            .filter_map(|f| f.delete_file.as_ref())
            .map(|df| df.file_size_bytes)
            .sum();
        let net_bytes = (data_bytes - delete_bytes).max(0) as usize;

        let mut stats = Statistics::new_unknown(&self.schema);
        stats.total_byte_size = Precision::Inexact(net_bytes);
        Some(stats)
    }

    fn supports_filters_pushdown(
        &self,
        filters: &[&Expr],
    ) -> DataFusionResult<Vec<TableProviderFilterPushDown>> {
        // Mark all filters as Inexact because we apply delete filters after the scan.
        // DataFusion will reapply these filters after DeleteFilterExec to ensure
        // correctness, but Parquet can still use them for:
        // - Row group pruning via statistics
        // - Page-level filtering with late materialization
        // - Bloom filter lookups (if available)
        Ok(filters
            .iter()
            .map(|_| TableProviderFilterPushDown::Inexact)
            .collect())
    }

    async fn scan(
        &self,
        state: &dyn Session,
        projection: Option<&Vec<usize>>,
        // Filters are received here for informational purposes. DataFusion's optimizer
        // automatically pushes them down to the Parquet scanner for row group pruning and
        // page-level filtering since we declared support via supports_filters_pushdown().
        // We mark them as Inexact, so DataFusion will reapply them after our scan.
        _filters: &[Expr],
        limit: Option<usize>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        // Row-lineage detour: when the synthetic `rowid` column is projected,
        // every file needs its own scan because each has a distinct
        // `row_id_start`. `projection == None` with row lineage on means "all
        // columns including rowid", which also routes through this path.
        let rowid_idx = self.rowid_index();
        let rowid_in_proj = match (rowid_idx, projection) {
            (Some(r), Some(p)) => p.contains(&r),
            (Some(_), None) => true,
            (None, _) => false,
        };

        if rowid_in_proj {
            let rowid_idx = rowid_idx.unwrap();
            let user_proj: Vec<usize> = projection
                .cloned()
                .unwrap_or_else(|| (0..self.schema.fields().len()).collect());

            let mut execs: Vec<Arc<dyn ExecutionPlan>> = Vec::new();
            for tf in &self.table_files {
                let exec = self
                    .build_exec_for_file_with_rowid(state, tf, &user_proj, rowid_idx, limit)
                    .await?;
                execs.push(exec);
            }

            if execs.is_empty() {
                use datafusion::physical_plan::empty::EmptyExec;
                let projected_schema = self.output_schema_for_projection(&user_proj, rowid_idx);
                return Ok(Arc::new(EmptyExec::new(projected_schema)));
            }

            return combine_execution_plans(execs);
        }

        // Fast path: rowid not projected. All projection indices refer to
        // physical columns, so the existing logic works untouched.
        let (files_with_deletes, files_without_deletes): (Vec<_>, Vec<_>) = self
            .table_files
            .iter()
            .partition(|tf| tf.delete_file.is_some());

        let mut execs: Vec<Arc<dyn ExecutionPlan>> = Vec::new();

        // Create single exec for all files without deletes (more efficient)
        if !files_without_deletes.is_empty() {
            let exec = self
                .build_exec_for_files_without_deletes(
                    state,
                    &files_without_deletes,
                    projection,
                    limit,
                )
                .await?;
            execs.push(exec);
        }

        // Only create separate execs for files with deletes
        for table_file in files_with_deletes {
            let exec = self
                .build_exec_for_file_with_deletes(state, table_file, projection, limit)
                .await?;
            execs.push(exec);
        }

        // Handle empty tables (no data files)
        if execs.is_empty() {
            use datafusion::physical_plan::empty::EmptyExec;
            let projected_schema = match projection {
                Some(indices) => Arc::new(self.schema.project(indices)?),
                None => self.schema.clone(),
            };
            return Ok(Arc::new(EmptyExec::new(projected_schema)));
        }

        // Combine execution plans
        combine_execution_plans(execs)
    }

    #[cfg(feature = "write")]
    async fn insert_into(
        &self,
        _state: &dyn Session,
        input: Arc<dyn ExecutionPlan>,
        insert_op: InsertOp,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let writer = self.writer.as_ref().ok_or_else(|| {
            DataFusionError::Plan(
                "Table is read-only. Use DuckLakeCatalog::with_writer() to enable writes."
                    .to_string(),
            )
        })?;

        let schema_name = self.schema_name.as_ref().ok_or_else(|| {
            DataFusionError::Internal("Schema name not set for writable table".to_string())
        })?;

        let write_mode = match insert_op {
            InsertOp::Append => WriteMode::Append,
            InsertOp::Overwrite | InsertOp::Replace => WriteMode::Replace,
        };

        Ok(Arc::new(DuckLakeInsertExec::new(
            input,
            Arc::clone(writer),
            schema_name.clone(),
            self.table_name.clone(),
            self.schema(),
            write_mode,
            self.object_store_url.clone(),
        )))
    }
}

/// Combines multiple execution plans into a single plan
fn combine_execution_plans(
    execs: Vec<Arc<dyn ExecutionPlan>>,
) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
    if execs.len() == 1 {
        Ok(execs.into_iter().next().unwrap())
    } else {
        use datafusion::physical_plan::union::UnionExec;
        UnionExec::try_new(execs)
    }
}

/// Extract deleted row positions from a delete file RecordBatch
///
/// Delete files have schema: (file_path: VARCHAR, pos: INT64)
/// We only extract the "pos" column - the "file_path" column is metadata/documentation
/// only (for Iceberg compatibility). The metadata catalog already tells us which delete
/// file is associated with which data file.
fn extract_deleted_positions_from_batch(
    batch: &RecordBatch,
    positions: &mut HashSet<i64>,
) -> DataFusionResult<()> {
    // Get the pos column index by name (not magic number)
    let schema = batch.schema();
    let pos_idx = schema.index_of(DELETE_POS_COL)?;

    // Get the pos column
    let pos_array = batch
        .column(pos_idx)
        .as_any()
        .downcast_ref::<Int64Array>()
        .ok_or_else(|| {
            DataFusionError::Internal(format!("{} column not found or wrong type", DELETE_POS_COL))
        })?;

    // Extract all non-null positions
    for i in 0..batch.num_rows() {
        if !pos_array.is_null(i) {
            positions.insert(pos_array.value(i));
        }
    }

    Ok(())
}

/// Check if a DataFusion error is caused by an object store NotFound error.
fn is_object_store_not_found(err: &DataFusionError) -> bool {
    if let DataFusionError::ObjectStore(os_err) = err {
        return matches!(&**os_err, object_store::Error::NotFound { .. });
    }
    let mut source = std::error::Error::source(err);
    while let Some(e) = source {
        if let Some(os_err) = e.downcast_ref::<object_store::Error>() {
            return matches!(os_err, object_store::Error::NotFound { .. });
        }
        source = e.source();
    }
    false
}

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

    #[test]
    fn test_validated_file_size_positive() {
        assert_eq!(validated_file_size(0, "test.parquet").unwrap(), 0);
        assert_eq!(validated_file_size(1024, "test.parquet").unwrap(), 1024);
        assert_eq!(
            validated_file_size(i64::MAX, "test.parquet").unwrap(),
            i64::MAX as u64
        );
    }

    #[test]
    fn test_validated_file_size_negative() {
        let err = validated_file_size(-1, "data/test.parquet").unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("-1"),
            "Error should contain the negative value: {}",
            msg
        );
        assert!(
            msg.contains("data/test.parquet"),
            "Error should contain the file path: {}",
            msg
        );
    }

    #[test]
    fn test_validated_file_size_large_negative() {
        let err = validated_file_size(i64::MIN, "bad.parquet").unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("bad.parquet"));
        assert!(msg.contains(&i64::MIN.to_string()));
    }

    #[test]
    fn test_validated_record_count_positive() {
        assert_eq!(validated_record_count(0, "test.parquet").unwrap(), 0);
        assert_eq!(validated_record_count(100, "test.parquet").unwrap(), 100);
        assert_eq!(
            validated_record_count(i64::MAX, "test.parquet").unwrap(),
            i64::MAX as u64
        );
    }

    #[test]
    fn test_validated_record_count_negative() {
        let err = validated_record_count(-1, "data/test.parquet").unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("-1"),
            "Error should contain the negative value: {}",
            msg
        );
        assert!(
            msg.contains("data/test.parquet"),
            "Error should contain the file path: {}",
            msg
        );
        assert!(
            msg.contains("record_count"),
            "Error should mention record_count: {}",
            msg
        );
    }

    #[test]
    fn test_validated_record_count_large_negative() {
        let err = validated_record_count(i64::MIN, "bad.parquet").unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("bad.parquet"));
        assert!(msg.contains(&i64::MIN.to_string()));
    }
}