iceberg/scan/

mod.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Table scan api.

mod cache;
use cache::*;
mod context;
use context::*;
mod task;

use std::collections::HashMap;
use std::future::Future;
use std::sync::Arc;

use arrow_array::RecordBatch;
use futures::channel::mpsc::{Sender, channel};
use futures::stream::BoxStream;
use futures::{SinkExt, StreamExt, TryStreamExt};
pub use task::*;

use crate::arrow::ArrowReaderBuilder;
pub use crate::arrow::{ScanMetrics, ScanResult};
use crate::delete_file_index::DeleteFileIndex;
use crate::error::invalid_data;
use crate::expr::visitors::inclusive_metrics_evaluator::InclusiveMetricsEvaluator;
use crate::expr::{Bind, BoundPredicate, Predicate};
use crate::io::FileIO;
use crate::metadata_columns::{
    RESERVED_FIELD_ID_PARTITION, get_metadata_field_id, is_metadata_column_name,
};
use crate::runtime::Runtime;
use crate::spec::{DataContentType, Schema, SchemaRef, SnapshotRef, SortOrderRef, StructType};
use crate::table::Table;
use crate::util::available_parallelism;
use crate::util::snapshot::snapshot_id_as_of_time;
use crate::{Error, ErrorKind, Result};

/// A stream of arrow [`RecordBatch`]es.
pub type ArrowRecordBatchStream = BoxStream<'static, Result<RecordBatch>>;

/// Resolves a column name to its field ID, honouring the scan's case sensitivity.
fn resolve_field_id(
    schema: &Schema,
    column_name: &str,
    case_sensitive: bool,
) -> Result<Option<i32>> {
    if case_sensitive {
        Ok(schema.field_id_by_name(column_name))
    } else {
        let field = schema.field_by_name_case_insensitive_checked(column_name)?;
        Ok(field.map(|field| field.id))
    }
}

fn collect_scan_field_ids(
    schema: &Schema,
    column_names: Option<&[String]>,
    case_sensitive: bool,
) -> Result<Vec<i32>> {
    let Some(column_names) = column_names else {
        return Ok(schema.as_struct().fields().iter().map(|f| f.id).collect());
    };

    column_names
        .iter()
        .map(|column_name| {
            if is_metadata_column_name(column_name) {
                return get_metadata_field_id(column_name);
            }

            let field_id = resolve_field_id(schema, column_name, case_sensitive)?.ok_or_else(|| {
                invalid_data!("Column {column_name} not found in table. Schema: {schema}")
            })?;

            schema
                .as_struct()
                .field_by_id(field_id)
                .ok_or_else(|| {
                    Error::new(
                        ErrorKind::FeatureUnsupported,
                        format!(
                            "Column {column_name} is not a direct child of schema but a nested field, which is not supported now. Schema: {schema}"
                        ),
                    )
                })?;

            Ok(field_id)
        })
        .collect()
}

fn bind_scan_predicate(
    schema: &SchemaRef,
    predicate: Option<&Predicate>,
    case_sensitive: bool,
) -> Result<Option<Arc<BoundPredicate>>> {
    predicate
        .map(|predicate| predicate.bind(schema.clone(), case_sensitive))
        .transpose()
        .map(|predicate| predicate.map(Arc::new))
}

fn projected_partition_type(
    table: &Table,
    schema: &Schema,
    field_ids: &[i32],
) -> Result<Option<Arc<StructType>>> {
    if !field_ids.contains(&RESERVED_FIELD_ID_PARTITION) {
        return Ok(None);
    }

    table
        .metadata()
        .unified_partition_type(schema)
        .map(Arc::new)
        .map(Some)
}

/// Selects a snapshot by ID or timestamp.
enum SnapshotSelection {
    SnapshotId(i64),
    AsOfTime(i64),
}

/// Builder to create table scan.
pub struct TableScanBuilder<'a> {
    table: &'a Table,
    // Defaults to none which means select all columns
    column_names: Option<Vec<String>>,
    snapshot_selection: Option<SnapshotSelection>,
    batch_size: Option<usize>,
    case_sensitive: bool,
    filter: Option<Predicate>,
    concurrency_limit_data_files: usize,
    concurrency_limit_manifest_entries: usize,
    concurrency_limit_manifest_files: usize,
    row_group_filtering_enabled: bool,
    row_selection_enabled: bool,
    bloom_filter_enabled: bool,
}

impl<'a> TableScanBuilder<'a> {
    pub(crate) fn new(table: &'a Table) -> Self {
        let num_cpus = available_parallelism().get();

        Self {
            table,
            column_names: None,
            snapshot_selection: None,
            batch_size: None,
            case_sensitive: true,
            filter: None,
            concurrency_limit_data_files: num_cpus,
            concurrency_limit_manifest_entries: num_cpus,
            concurrency_limit_manifest_files: num_cpus,
            row_group_filtering_enabled: true,
            row_selection_enabled: false,
            bloom_filter_enabled: false,
        }
    }

    /// Sets the desired size of batches in the response
    /// to something other than the default
    pub fn with_batch_size(mut self, batch_size: Option<usize>) -> Self {
        self.batch_size = batch_size;
        self
    }

    /// Sets the scan's case sensitivity
    pub fn with_case_sensitive(mut self, case_sensitive: bool) -> Self {
        self.case_sensitive = case_sensitive;
        self
    }

    /// Specifies a predicate to use as a filter
    pub fn with_filter(mut self, predicate: Predicate) -> Self {
        // calls rewrite_not to remove Not nodes, which must be absent
        // when applying the manifest evaluator
        self.filter = Some(predicate.rewrite_not());
        self
    }

    /// Select all columns.
    pub fn select_all(mut self) -> Self {
        self.column_names = None;
        self
    }

    /// Select empty columns.
    pub fn select_empty(mut self) -> Self {
        self.column_names = Some(vec![]);
        self
    }

    /// Select some columns of the table.
    pub fn select(mut self, column_names: impl IntoIterator<Item = impl ToString>) -> Self {
        self.column_names = Some(
            column_names
                .into_iter()
                .map(|item| item.to_string())
                .collect(),
        );
        self
    }

    /// Set the snapshot to scan. The scan projects the schema that snapshot was
    /// written with.
    ///
    /// The last call to this or [`Self::as_of_time`] takes precedence.
    /// If neither is set, the scan reads the current snapshot with the table's
    /// current schema.
    pub fn snapshot_id(mut self, snapshot_id: i64) -> Self {
        self.snapshot_selection = Some(SnapshotSelection::SnapshotId(snapshot_id));
        self
    }

    /// Select a snapshot at or before a timestamp in milliseconds since the Unix epoch.
    ///
    /// Uses the table's main history, and projects the schema the selected snapshot
    /// was written with. The last call to this or [`Self::snapshot_id`] takes precedence.
    ///
    /// ```
    /// # use iceberg::TableIdent;
    /// # use iceberg::io::FileIO;
    /// # use iceberg::table::StaticTable;
    /// # #[tokio::main]
    /// # async fn main() -> iceberg::Result<()> {
    /// # let location = concat!(env!("CARGO_MANIFEST_DIR"), "/testdata/example_table_metadata_v2.json");
    /// # let ident = TableIdent::from_strs(["ns", "t"])?;
    /// # let table = StaticTable::from_metadata_file(location, ident, FileIO::new_with_fs()).await?;
    /// // The snapshot log has entries at 1515100955770 and 1555100955770.
    /// let scan = table.scan().as_of_time(1_555_100_955_769).build()?;
    /// assert_eq!(scan.snapshot().unwrap().snapshot_id(), 3051729675574597004);
    /// # Ok(())
    /// # }
    /// ```
    pub fn as_of_time(mut self, timestamp_ms: i64) -> Self {
        self.snapshot_selection = Some(SnapshotSelection::AsOfTime(timestamp_ms));
        self
    }

    /// Sets the concurrency limit for both manifest files and manifest
    /// entries for this scan
    pub fn with_concurrency_limit(mut self, limit: usize) -> Self {
        self.concurrency_limit_manifest_files = limit;
        self.concurrency_limit_manifest_entries = limit;
        self.concurrency_limit_data_files = limit;
        self
    }

    /// Sets the data file concurrency limit for this scan
    pub fn with_data_file_concurrency_limit(mut self, limit: usize) -> Self {
        self.concurrency_limit_data_files = limit;
        self
    }

    /// Sets the manifest entry concurrency limit for this scan
    pub fn with_manifest_entry_concurrency_limit(mut self, limit: usize) -> Self {
        self.concurrency_limit_manifest_entries = limit;
        self
    }

    /// Determines whether to enable row group filtering.
    /// When enabled, if a read is performed with a filter predicate,
    /// then the metadata for each row group in the parquet file is
    /// evaluated against the filter predicate and row groups
    /// that cant contain matching rows will be skipped entirely.
    ///
    /// Defaults to enabled, as it generally improves performance or
    /// keeps it the same, with performance degradation unlikely.
    pub fn with_row_group_filtering_enabled(mut self, row_group_filtering_enabled: bool) -> Self {
        self.row_group_filtering_enabled = row_group_filtering_enabled;
        self
    }

    /// Determines whether to enable row selection.
    /// When enabled, if a read is performed with a filter predicate,
    /// then (for row groups that have not been skipped) the page index
    /// for each row group in a parquet file is parsed and evaluated
    /// against the filter predicate to determine if ranges of rows
    /// within a row group can be skipped, based upon the page-level
    /// statistics for each column.
    ///
    /// Defaults to being disabled. Enabling requires parsing the parquet page
    /// index, which can be slow enough that parsing the page index outweighs any
    /// gains from the reduced number of rows that need scanning.
    /// It is recommended to experiment with partitioning, sorting, row group size,
    /// page size, and page row limit Iceberg settings on the table being scanned in
    /// order to get the best performance from using row selection.
    pub fn with_row_selection_enabled(mut self, row_selection_enabled: bool) -> Self {
        self.row_selection_enabled = row_selection_enabled;
        self
    }

    /// Determines whether to enable bloom filter-based row group filtering.
    ///
    /// When enabled, if a read is performed with an equality or IN predicate,
    /// the bloom filter for relevant columns in each row group is read and
    /// checked. Row groups where the bloom filter proves the value is absent
    /// are skipped entirely.
    ///
    /// Defaults to disabled, as reading bloom filters requires additional I/O
    /// per column per row group.
    pub fn with_bloom_filter_enabled(mut self, bloom_filter_enabled: bool) -> Self {
        self.bloom_filter_enabled = bloom_filter_enabled;
        self
    }

    /// Build the table scan.
    pub fn build(self) -> Result<TableScan> {
        let selected_snapshot = match self.snapshot_selection {
            Some(SnapshotSelection::SnapshotId(id)) => {
                Some((id, format!("Snapshot with id {id} not found")))
            }
            Some(SnapshotSelection::AsOfTime(timestamp_ms)) => {
                let id = snapshot_id_as_of_time(&self.table.metadata_ref(), timestamp_ms)?;
                // History outlives expired snapshots; name the timestamp that resolved to `id`.
                Some((
                    id,
                    format!(
                        "Snapshot with id {id} resolved from timestamp {timestamp_ms} ms not found; it may have expired"
                    ),
                ))
            }
            None => None,
        };
        let is_time_travel = selected_snapshot.is_some();
        let snapshot = match selected_snapshot {
            Some((snapshot_id, not_found_message)) => self
                .table
                .metadata()
                .snapshot_by_id(snapshot_id)
                .ok_or_else(|| invalid_data!(not_found_message))?
                .clone(),
            None => {
                let Some(current_snapshot_id) = self.table.metadata().current_snapshot() else {
                    return Ok(TableScan {
                        batch_size: self.batch_size,
                        column_names: self.column_names,
                        file_io: self.table.file_io().clone(),
                        plan_context: None,
                        concurrency_limit_data_files: self.concurrency_limit_data_files,
                        concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries,
                        concurrency_limit_manifest_files: self.concurrency_limit_manifest_files,
                        row_group_filtering_enabled: self.row_group_filtering_enabled,
                        row_selection_enabled: self.row_selection_enabled,
                        bloom_filter_enabled: self.bloom_filter_enabled,
                        runtime: self.table.runtime().clone(),
                    });
                };
                current_snapshot_id.clone()
            }
        };

        // A scan of the current state uses the table's current schema; time travel uses the
        // schema its snapshot was written with.
        let schema = if is_time_travel {
            snapshot.schema(self.table.metadata())?
        } else {
            self.table.metadata().current_schema().clone()
        };
        let field_ids =
            collect_scan_field_ids(&schema, self.column_names.as_deref(), self.case_sensitive)?;
        let snapshot_bound_predicate =
            bind_scan_predicate(&schema, self.filter.as_ref(), self.case_sensitive)?;
        let name_mapping = self
            .table
            .metadata()
            .table_properties()
            .default_name_mapping()?
            .map(Arc::new);
        let unified_partition_type = projected_partition_type(self.table, &schema, &field_ids)?;

        // Precompute the table's sort orders once, keyed by id, so each manifest-file
        // context carries only this narrow map instead of the full table metadata.
        let sort_orders = Arc::new(
            self.table
                .metadata()
                .sort_orders_iter()
                .map(|order| (order.order_id, order.clone()))
                .collect::<HashMap<i64, SortOrderRef>>(),
        );

        let plan_context = PlanContext {
            snapshot,
            table_metadata: self.table.metadata_ref(),
            snapshot_schema: schema,
            case_sensitive: self.case_sensitive,
            predicate: self.filter.map(Arc::new),
            snapshot_bound_predicate,
            object_cache: self.table.object_cache(),
            field_ids: Arc::new(field_ids),
            name_mapping,
            partition_filter_cache: Arc::new(PartitionFilterCache::new()),
            manifest_evaluator_cache: Arc::new(ManifestEvaluatorCache::new()),
            expression_evaluator_cache: Arc::new(ExpressionEvaluatorCache::new()),
            unified_partition_type,
            sort_orders,
        };

        Ok(TableScan {
            batch_size: self.batch_size,
            column_names: self.column_names,
            file_io: self.table.file_io().clone(),
            plan_context: Some(plan_context),
            concurrency_limit_data_files: self.concurrency_limit_data_files,
            concurrency_limit_manifest_entries: self.concurrency_limit_manifest_entries,
            concurrency_limit_manifest_files: self.concurrency_limit_manifest_files,
            row_group_filtering_enabled: self.row_group_filtering_enabled,
            row_selection_enabled: self.row_selection_enabled,
            bloom_filter_enabled: self.bloom_filter_enabled,
            runtime: self.table.runtime().clone(),
        })
    }
}

/// Table scan.
#[derive(Debug)]
pub struct TableScan {
    /// A [PlanContext], if this table has at least one snapshot, otherwise None.
    ///
    /// If this is None, then the scan contains no rows.
    plan_context: Option<PlanContext>,
    batch_size: Option<usize>,
    file_io: FileIO,
    column_names: Option<Vec<String>>,
    /// The maximum number of manifest files that will be
    /// retrieved from [`FileIO`] concurrently
    concurrency_limit_manifest_files: usize,

    /// The maximum number of [`ManifestEntry`]s that will
    /// be processed in parallel
    concurrency_limit_manifest_entries: usize,

    /// The maximum number of [`ManifestEntry`]s that will
    /// be processed in parallel
    concurrency_limit_data_files: usize,

    row_group_filtering_enabled: bool,
    row_selection_enabled: bool,
    bloom_filter_enabled: bool,

    runtime: Runtime,
}

impl TableScan {
    /// Returns a stream of [`FileScanTask`]s.
    pub async fn plan_files(&self) -> Result<FileScanTaskStream> {
        self.plan_data_files(|ctx| ctx.into_file_scan_task()).await
    }

    pub(crate) async fn plan_cow_rewrite_files(
        &self,
    ) -> Result<BoxStream<'static, Result<crate::cow_rewrite::CowRewriteFile>>> {
        self.plan_data_files(|ctx| ctx.into_cow_rewrite_file())
            .await
    }

    async fn plan_data_files<T, F, Fut>(
        &self,
        build_result: F,
    ) -> Result<BoxStream<'static, Result<T>>>
    where
        T: Send + 'static,
        F: Fn(ManifestEntryContext) -> Fut + Copy + Send + Sync + 'static,
        Fut: Future<Output = Result<T>> + Send + 'static,
    {
        let Some(plan_context) = self.plan_context.as_ref() else {
            return Ok(Box::pin(futures::stream::empty()));
        };

        let concurrency_limit_manifest_files = self.concurrency_limit_manifest_files;
        let concurrency_limit_manifest_entries = self.concurrency_limit_manifest_entries;

        // used to stream ManifestEntryContexts between stages of the file plan operation
        let (manifest_entry_data_ctx_tx, manifest_entry_data_ctx_rx) =
            channel(concurrency_limit_manifest_files);
        let (manifest_entry_delete_ctx_tx, manifest_entry_delete_ctx_rx) =
            channel(concurrency_limit_manifest_files);

        // used to stream the planned data file results back to the caller
        let (planned_file_tx, planned_file_rx) = channel(concurrency_limit_manifest_entries);

        let (delete_file_idx, delete_file_tx) = DeleteFileIndex::new(self.runtime.clone());

        let manifest_list = plan_context.get_manifest_list().await?;

        // get the [`ManifestFile`]s from the [`ManifestList`], filtering out any
        // whose partitions cannot match this
        // scan's filter
        let manifest_file_contexts = plan_context.build_manifest_file_contexts(
            manifest_list,
            manifest_entry_data_ctx_tx,
            delete_file_idx.clone(),
            manifest_entry_delete_ctx_tx,
        )?;

        let mut channel_for_manifest_error = planned_file_tx.clone();
        let mut channel_for_data_manifest_entry_error = planned_file_tx.clone();
        let mut channel_for_delete_manifest_entry_error = planned_file_tx.clone();

        let rt = self.runtime.clone();

        // Concurrently load all [`Manifest`]s and stream their [`ManifestEntry`]s
        rt.io().spawn(async move {
            let result = futures::stream::iter(manifest_file_contexts)
                .try_for_each_concurrent(concurrency_limit_manifest_files, |ctx| async move {
                    ctx.fetch_manifest_and_stream_manifest_entries().await
                })
                .await;

            if let Err(error) = result {
                let _ = channel_for_manifest_error.send(Err(error)).await;
            }
        });

        // Process the delete file [`ManifestEntry`] stream in parallel
        {
            let rt = rt.clone();
            let rt_inner = rt.clone();
            rt.cpu().spawn(async move {
                let result = manifest_entry_delete_ctx_rx
                    .map(|me_ctx| Ok((me_ctx, delete_file_tx.clone())))
                    .try_for_each_concurrent(
                        concurrency_limit_manifest_entries,
                        |(manifest_entry_context, tx)| {
                            let rt_inner = rt_inner.clone();
                            async move {
                                rt_inner
                                    .cpu()
                                    .spawn(async move {
                                        Self::process_delete_manifest_entry(
                                            manifest_entry_context,
                                            tx,
                                        )
                                        .await
                                    })
                                    .await?
                            }
                        },
                    )
                    .await;

                if let Err(error) = result {
                    let _ = channel_for_delete_manifest_entry_error
                        .send(Err(error))
                        .await;
                }
            });
        }

        // Process the data file [`ManifestEntry`] stream in parallel
        {
            let rt_inner = rt.clone();
            rt.cpu().spawn(async move {
                let result = manifest_entry_data_ctx_rx
                    .map(|me_ctx| Ok((me_ctx, planned_file_tx.clone())))
                    .try_for_each_concurrent(
                        concurrency_limit_manifest_entries,
                        |(manifest_entry_context, tx)| {
                            let rt_inner = rt_inner.clone();
                            async move {
                                rt_inner
                                    .cpu()
                                    .spawn(async move {
                                        Self::process_data_manifest_entry(
                                            manifest_entry_context,
                                            tx,
                                            build_result,
                                        )
                                        .await
                                    })
                                    .await?
                            }
                        },
                    )
                    .await;

                if let Err(error) = result {
                    let _ = channel_for_data_manifest_entry_error.send(Err(error)).await;
                }
            });
        }

        Ok(planned_file_rx.boxed())
    }

    /// Returns an [`ArrowRecordBatchStream`].
    pub async fn to_arrow(&self) -> Result<ArrowRecordBatchStream> {
        let mut arrow_reader_builder =
            ArrowReaderBuilder::new(self.file_io.clone(), self.runtime.clone())
                .with_data_file_concurrency_limit(self.concurrency_limit_data_files)
                .with_row_group_filtering_enabled(self.row_group_filtering_enabled)
                .with_row_selection_enabled(self.row_selection_enabled)
                .with_bloom_filter_enabled(self.bloom_filter_enabled);

        if let Some(batch_size) = self.batch_size {
            arrow_reader_builder = arrow_reader_builder.with_batch_size(batch_size);
        }

        arrow_reader_builder
            .build()
            .read(self.plan_files().await?)
            .map(|result| result.stream())
    }

    /// Returns a reference to the column names of the table scan.
    pub fn column_names(&self) -> Option<&[String]> {
        self.column_names.as_deref()
    }

    /// Returns a reference to the snapshot of the table scan.
    pub fn snapshot(&self) -> Option<&SnapshotRef> {
        self.plan_context.as_ref().map(|x| &x.snapshot)
    }

    async fn process_data_manifest_entry<T, F, Fut>(
        manifest_entry_context: ManifestEntryContext,
        mut planned_file_tx: Sender<Result<T>>,
        build_result: F,
    ) -> Result<()>
    where
        F: Fn(ManifestEntryContext) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<T>> + Send + 'static,
    {
        // abort the plan if we encounter a manifest entry for a delete file
        if manifest_entry_context.manifest_entry.content_type() != DataContentType::Data {
            return Err(Error::new(
                ErrorKind::FeatureUnsupported,
                "Encountered an entry for a delete file in a data file manifest",
            ));
        }

        if let Some(ref bound_predicates) = manifest_entry_context.bound_predicates {
            let BoundPredicates {
                snapshot_bound_predicate,
                partition_bound_predicate,
            } = bound_predicates.as_ref();

            let expression_evaluator_cache =
                manifest_entry_context.expression_evaluator_cache.as_ref();

            let expression_evaluator = expression_evaluator_cache.get(
                manifest_entry_context.partition_spec_id,
                partition_bound_predicate,
            )?;

            // skip any data file whose partition data indicates that it can't contain
            // any data that matches this scan's filter
            if !expression_evaluator.eval(manifest_entry_context.manifest_entry.data_file())? {
                return Ok(());
            }

            // skip any data file whose metrics don't match this scan's filter
            if !InclusiveMetricsEvaluator::eval(
                snapshot_bound_predicate,
                manifest_entry_context.manifest_entry.data_file(),
                false,
            )? {
                return Ok(());
            }
        }

        // congratulations! the manifest entry has made its way through the
        // entire plan without getting filtered out. Build the planned file before
        // sending so delete-file lookup preserves the original scan timing.
        planned_file_tx
            .send(Ok(build_result(manifest_entry_context).await?))
            .await?;

        Ok(())
    }

    async fn process_delete_manifest_entry(
        manifest_entry_context: ManifestEntryContext,
        mut delete_file_ctx_tx: Sender<DeleteFileContext>,
    ) -> Result<()> {
        // abort the plan if we encounter a manifest entry that is not for a delete file
        if manifest_entry_context.manifest_entry.content_type() == DataContentType::Data {
            return Err(Error::new(
                ErrorKind::FeatureUnsupported,
                "Encountered an entry for a data file in a delete manifest",
            ));
        }

        if let Some(ref bound_predicates) = manifest_entry_context.bound_predicates {
            let expression_evaluator_cache =
                manifest_entry_context.expression_evaluator_cache.as_ref();

            let expression_evaluator = expression_evaluator_cache.get(
                manifest_entry_context.partition_spec_id,
                &bound_predicates.partition_bound_predicate,
            )?;

            // skip any data file whose partition data indicates that it can't contain
            // any data that matches this scan's filter
            if !expression_evaluator.eval(manifest_entry_context.manifest_entry.data_file())? {
                return Ok(());
            }
        }

        delete_file_ctx_tx
            .send(DeleteFileContext {
                manifest_entry: manifest_entry_context.manifest_entry.clone(),
                partition_spec_id: manifest_entry_context.partition_spec_id,
            })
            .await?;

        Ok(())
    }
}

pub(crate) struct BoundPredicates {
    partition_bound_predicate: BoundPredicate,
    snapshot_bound_predicate: BoundPredicate,
}

#[cfg(test)]
mod tests {
    //! shared tests for the table scan API

    use std::collections::HashMap;
    use std::fs;
    use std::fs::File;
    use std::sync::Arc;

    use arrow_array::cast::AsArray;
    use arrow_array::types::Int32Type;
    use arrow_array::{
        Array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, RecordBatch, RunArray,
        StringArray,
    };
    use futures::{TryStreamExt, stream};
    use parquet::arrow::ArrowWriter;
    use uuid::Uuid;

    use super::resolve_field_id;
    use crate::arrow::ArrowReaderBuilder;
    use crate::expr::{BoundPredicate, Reference};
    use crate::io::FileIO;
    use crate::metadata_columns::{
        RESERVED_COL_NAME_FILE, RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER,
        RESERVED_COL_NAME_POS, RESERVED_COL_NAME_SPEC_ID, RESERVED_FIELD_ID_POS,
    };
    use crate::scan::{FileScanTask, FileScanTaskDeleteFile};
    use crate::spec::{
        DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, MAIN_BRANCH,
        ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, MappedField,
        NameMapping, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType, Schema,
        Snapshot, SortDirection, SortField, SortOrder, Struct, Summary, TableMetadataBuilder,
        TableProperties, Transform, Type, UnboundPartitionSpec,
    };
    use crate::table::Table;
    use crate::test_utils::scan::{
        ROWS_PER_FIXTURE_FILE, TableTestFixture, assert_last_updated_seq_all,
    };
    use crate::test_utils::test_runtime;
    use crate::util::snapshot::snapshot_id_as_of_time;
    use crate::{ErrorKind, TableIdent};

    #[tokio::test]
    async fn test_table_scan_columns() {
        let table = TableTestFixture::new().table;

        let table_scan = table.scan().select(["x", "y"]).build().unwrap();
        assert_eq!(
            Some(vec!["x".to_string(), "y".to_string()]),
            table_scan.column_names
        );

        let table_scan = table
            .scan()
            .select(["x", "y"])
            .select(["z"])
            .build()
            .unwrap();
        assert_eq!(Some(vec!["z".to_string()]), table_scan.column_names);
    }

    #[tokio::test]
    async fn test_select_all() {
        let table = TableTestFixture::new().table;

        let table_scan = table.scan().select_all().build().unwrap();
        assert!(table_scan.column_names.is_none());
    }

    #[test]
    fn test_select_no_exist_column() {
        let table = TableTestFixture::new().table;

        let table_scan = table.scan().select(["x", "y", "z", "a", "b"]).build();
        assert!(table_scan.is_err());
    }

    #[test]
    fn test_case_sensitive_scan_rejects_mismatched_column_case() {
        let table = TableTestFixture::new().table;

        // Case sensitivity defaults to true, so "X" must not resolve to "x".
        assert!(table.scan().select(["X"]).build().is_err());
        assert!(
            table
                .scan()
                .with_filter(Reference::new("X").greater_than(Datum::long(1)))
                .build()
                .is_err()
        );
    }

    #[test]
    fn test_resolve_field_id_case_insensitive_rejects_ambiguous_name() {
        let schema = Schema::builder()
            .with_fields(vec![
                NestedField::optional(1, "id", PrimitiveType::Int).into(),
                NestedField::optional(2, "ID", PrimitiveType::Int).into(),
                NestedField::optional(3, "data", PrimitiveType::Int).into(),
            ])
            .build()
            .unwrap();

        assert_eq!(resolve_field_id(&schema, "id", true).unwrap(), Some(1));
        assert_eq!(resolve_field_id(&schema, "DATA", false).unwrap(), Some(3));
        let error = resolve_field_id(&schema, "Id", false).unwrap_err();
        assert_eq!(error.kind(), ErrorKind::DataInvalid);
        assert!(error.message().contains("case-insensitively"), "{error}");
    }

    #[test]
    fn test_case_insensitive_scan_with_ambiguous_schema() {
        let fixture = TableTestFixture::new();
        let mut metadata = fixture.table.metadata().clone();
        let schema = Schema::builder()
            .with_schema_id(metadata.current_schema_id())
            .with_fields(vec![
                NestedField::optional(1, "id", PrimitiveType::Int).into(),
                NestedField::optional(2, "ID", PrimitiveType::Int).into(),
                NestedField::optional(3, "data", PrimitiveType::Int).into(),
            ])
            .build()
            .unwrap();
        metadata
            .schemas
            .insert(metadata.current_schema_id(), Arc::new(schema));
        let table = fixture.table.with_metadata(Arc::new(metadata));

        table
            .scan()
            .with_case_sensitive(false)
            .select(["DATA"])
            .build()
            .unwrap();
        let error = table
            .scan()
            .with_case_sensitive(false)
            .select(["Id"])
            .build()
            .unwrap_err();
        assert_eq!(error.kind(), ErrorKind::DataInvalid);
        assert!(error.message().contains("case-insensitively"), "{error}");
    }

    #[tokio::test]
    async fn test_case_insensitive_scan_resolves_mismatched_column_case() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // The schema declares lowercase "x" and "z"; a case-insensitive scan must
        // resolve the upper-cased names in both the projection and the filter.
        let table_scan = fixture
            .table
            .scan()
            .with_case_sensitive(false)
            .select(["X", "Z"])
            .with_filter(Reference::new("Y").greater_than(Datum::long(1)))
            .build()
            .unwrap();

        let batches: Vec<_> = table_scan
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        assert_eq!(batches[0].num_columns(), 2);
        assert_eq!(
            batches[0]
                .column_by_name("x")
                .unwrap()
                .as_any()
                .downcast_ref::<Int64Array>()
                .unwrap()
                .value(0),
            1
        );
        assert_eq!(
            batches[0]
                .column_by_name("z")
                .unwrap()
                .as_any()
                .downcast_ref::<Int64Array>()
                .unwrap()
                .value(0),
            3
        );
    }

    #[tokio::test]
    async fn test_table_scan_default_snapshot_id() {
        let table = TableTestFixture::new().table;

        let table_scan = table.scan().build().unwrap();
        assert_eq!(
            table.metadata().current_snapshot().unwrap().snapshot_id(),
            table_scan.snapshot().unwrap().snapshot_id()
        );
    }

    #[test]
    fn test_table_scan_non_exist_snapshot_id() {
        let table = TableTestFixture::new().table;

        let table_scan = table.scan().snapshot_id(1024).build();
        assert!(table_scan.is_err());
    }

    #[tokio::test]
    async fn test_table_scan_with_snapshot_id() {
        let table = TableTestFixture::new().table;

        let table_scan = table
            .scan()
            .snapshot_id(3051729675574597004)
            .with_row_selection_enabled(true)
            .build()
            .unwrap();
        assert_eq!(
            table_scan.snapshot().unwrap().snapshot_id(),
            3051729675574597004
        );
    }

    #[test]
    fn test_table_scan_as_of_time_after_rollback() {
        for elapsed_ms in [0, 1000] {
            let table = TableTestFixture::new().table;
            let mut metadata = table.metadata().clone();
            let original = metadata.history()[0].snapshot_id;
            let previous = metadata.history()[1].clone();
            let rollback_time = previous.timestamp_ms + elapsed_ms;
            metadata.snapshot_log.push(crate::spec::SnapshotLog {
                timestamp_ms: rollback_time,
                snapshot_id: original,
            });
            metadata.current_snapshot_id = Some(original);
            metadata.refs.get_mut(MAIN_BRANCH).unwrap().snapshot_id = original;
            let table = table.with_metadata(Arc::new(metadata));
            let scan = table.scan().as_of_time(rollback_time).build().unwrap();
            // A rollback at the same timestamp keeps the earlier history entry.
            let expected = if elapsed_ms == 0 {
                previous.snapshot_id
            } else {
                original
            };
            assert_eq!(scan.snapshot().unwrap().snapshot_id(), expected);
        }
    }

    #[test]
    fn test_table_scan_as_of_time_rejects_empty_history() {
        let table = TableTestFixture::new_empty().table;
        assert!(table.scan().build().unwrap().snapshot().is_none());
        let err = table.scan().as_of_time(0).build().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
        assert!(err.message().contains("No snapshot history"));
    }

    #[test]
    fn test_table_scan_as_of_time_rejects_before_history() {
        let table = TableTestFixture::new().table;
        let before = table.metadata().history()[0].timestamp_ms - 1;
        let err = table.scan().as_of_time(before).build().unwrap_err();
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
        assert!(err.message().contains(&before.to_string()));
    }

    #[test]
    fn test_table_scan_as_of_time_rejects_all_qualifying_snapshots_expired() {
        let table = TableTestFixture::new_with_deep_history().table;
        let mut metadata = table.metadata().clone();
        let entry = metadata.history()[3].clone();
        let expired_ids: Vec<_> = metadata
            .history()
            .iter()
            .filter(|log| log.timestamp_ms <= entry.timestamp_ms)
            .map(|log| log.snapshot_id)
            .collect();
        assert_eq!(expired_ids.len(), 4);
        for snapshot_id in expired_ids {
            metadata.snapshots.remove(&snapshot_id);
        }
        let table = table.with_metadata(Arc::new(metadata));
        // History resolution succeeds even though every qualifying snapshot expired.
        assert_eq!(
            snapshot_id_as_of_time(&table.metadata_ref(), entry.timestamp_ms).unwrap(),
            entry.snapshot_id
        );
        assert!(table.scan().build().unwrap().snapshot().is_some());
        let err = table
            .scan()
            .as_of_time(entry.timestamp_ms)
            .build()
            .unwrap_err();
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
        assert_eq!(
            err.message(),
            format!(
                "Snapshot with id {} resolved from timestamp {} ms not found; it may have expired",
                entry.snapshot_id, entry.timestamp_ms
            )
        );
    }

    #[test]
    fn test_table_scan_last_snapshot_selection_wins() {
        let table = TableTestFixture::new().table;
        let first = &table.metadata().history()[0];
        let last = &table.metadata().history()[1];
        let cases = [
            (
                table
                    .scan()
                    .snapshot_id(last.snapshot_id)
                    .as_of_time(first.timestamp_ms),
                first.snapshot_id,
            ),
            (
                table
                    .scan()
                    .as_of_time(first.timestamp_ms)
                    .snapshot_id(last.snapshot_id),
                last.snapshot_id,
            ),
            // An overwritten selector is not resolved or validated.
            (
                table.scan().snapshot_id(-1).as_of_time(first.timestamp_ms),
                first.snapshot_id,
            ),
            (
                table
                    .scan()
                    .as_of_time(i64::MIN)
                    .snapshot_id(last.snapshot_id),
                last.snapshot_id,
            ),
            (
                table
                    .scan()
                    .snapshot_id(first.snapshot_id)
                    .snapshot_id(last.snapshot_id),
                last.snapshot_id,
            ),
            (
                table
                    .scan()
                    .as_of_time(i64::MAX)
                    .as_of_time(first.timestamp_ms),
                first.snapshot_id,
            ),
        ];
        for (builder, expected) in cases {
            let scan = builder.build().unwrap();
            assert_eq!(scan.snapshot().unwrap().snapshot_id(), expected);
        }
    }

    #[test]
    fn test_table_scan_invalid_last_snapshot_selection() {
        let table = TableTestFixture::new().table;
        let first = &table.metadata().history()[0];
        let cases = [
            (
                table
                    .scan()
                    .snapshot_id(first.snapshot_id)
                    .as_of_time(first.timestamp_ms - 1),
                "No snapshot history",
            ),
            (
                table.scan().as_of_time(first.timestamp_ms).snapshot_id(-1),
                "Snapshot with id -1 not found",
            ),
        ];
        for (builder, expected) in cases {
            let err = builder.build().unwrap_err();
            assert_eq!(err.kind(), ErrorKind::DataInvalid);
            assert!(err.message().contains(expected));
        }
    }

    #[test]
    fn test_table_scan_as_of_time_uses_snapshot_schema() {
        for select_current in [false, true] {
            let table = TableTestFixture::new().table;
            let mut metadata = table.metadata().clone();
            let entry = metadata.history()[usize::from(select_current)].clone();
            // Simulate a schema-only update after this snapshot: schema 0 has x only,
            // while the current table schema also contains y and other columns.
            Arc::make_mut(metadata.snapshots.get_mut(&entry.snapshot_id).unwrap()).schema_id =
                Some(0);
            let table = table.with_metadata(Arc::new(metadata));
            let scan = table
                .scan()
                .as_of_time(entry.timestamp_ms)
                .select(["x"])
                .with_filter(Reference::new("x").greater_than(Datum::long(0)))
                .build()
                .unwrap();
            let context = scan.plan_context.as_ref().unwrap();
            assert_eq!(context.snapshot_schema.schema_id(), 0);
            assert!(context.snapshot_bound_predicate.is_some());
            assert!(
                table
                    .scan()
                    .as_of_time(entry.timestamp_ms)
                    .select(["y"])
                    .build()
                    .is_err()
            );
            assert!(
                table
                    .scan()
                    .as_of_time(entry.timestamp_ms)
                    .with_filter(Reference::new("y").greater_than(Datum::long(0)))
                    .build()
                    .is_err()
            );
        }
    }

    #[test]
    fn test_table_scan_as_of_time_schema_compatibility() {
        let table = TableTestFixture::new().table;
        let entry = table.metadata().history()[0].clone();
        // Older snapshots may omit schema-id and use the current-schema fallback.
        assert!(
            table
                .metadata()
                .snapshot_by_id(entry.snapshot_id)
                .unwrap()
                .schema_id()
                .is_none()
        );
        let scan = table.scan().as_of_time(entry.timestamp_ms).build().unwrap();
        assert_eq!(
            scan.plan_context.as_ref().unwrap().snapshot_schema,
            *table.metadata().current_schema()
        );

        let mut metadata = table.metadata().clone();
        Arc::make_mut(metadata.snapshots.get_mut(&entry.snapshot_id).unwrap()).schema_id =
            Some(1234);
        let table = table.with_metadata(Arc::new(metadata));
        let err = table
            .scan()
            .as_of_time(entry.timestamp_ms)
            .build()
            .unwrap_err();
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
        assert!(err.message().contains("Schema with id 1234 not found"));
    }

    #[tokio::test]
    async fn test_table_scan_as_of_time_reads_historical_rows() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;
        let entry = fixture.table.metadata().history()[0].clone();

        // The older snapshot has only x, while the current snapshot has eight columns.
        let mut metadata = fixture.table.metadata().clone();
        Arc::make_mut(metadata.snapshots.get_mut(&entry.snapshot_id).unwrap()).schema_id = Some(0);
        fixture.table = fixture.table.with_metadata(Arc::new(metadata));
        let snapshot = fixture
            .table
            .metadata()
            .snapshot_by_id(entry.snapshot_id)
            .unwrap();
        let schema = snapshot.schema(fixture.table.metadata()).unwrap();
        let arrow_schema = Arc::new(crate::arrow::schema_to_arrow_schema(&schema).unwrap());

        // Write distinct historical data: x = 100, versus x = 1 in the current files.
        let path = format!("{}/historical.parquet", fixture.table_location);
        let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(Int64Array::from(
            vec![100],
        ))])
        .unwrap();
        let mut writer =
            ArrowWriter::try_new(File::create(&path).unwrap(), arrow_schema, None).unwrap();
        writer.write(&batch).unwrap();
        writer.close().unwrap();

        let mut manifest_writer = ManifestWriterBuilder::new(
            fixture.next_manifest_file(),
            Some(snapshot.snapshot_id()),
            schema,
            fixture
                .table
                .metadata()
                .default_partition_spec()
                .as_ref()
                .clone(),
        )
        .build_v2_data();
        manifest_writer
            .add_entry(
                ManifestEntry::builder()
                    .status(ManifestStatus::Added)
                    .data_file(
                        DataFileBuilder::default()
                            .partition_spec_id(0)
                            .content(DataContentType::Data)
                            .file_format(DataFileFormat::Parquet)
                            .file_size_in_bytes(fs::metadata(&path).unwrap().len())
                            .file_path(path)
                            .record_count(1)
                            .partition(Struct::from_iter([Some(Literal::long(100))]))
                            .build()
                            .unwrap(),
                    )
                    .build(),
            )
            .unwrap();
        let manifest = manifest_writer.write_manifest_file().await.unwrap();
        let output = fixture
            .table
            .file_io()
            .new_output(snapshot.manifest_list())
            .unwrap()
            .writer()
            .await
            .unwrap();
        let mut list_writer = ManifestListWriter::v2(
            output,
            snapshot.snapshot_id(),
            snapshot.parent_snapshot_id(),
            snapshot.sequence_number(),
        );
        list_writer.add_manifests([manifest].into_iter()).unwrap();
        list_writer.close().await.unwrap();

        let table = fixture.table;
        let [by_time, by_id] = [
            table.scan().as_of_time(entry.timestamp_ms),
            table.scan().snapshot_id(entry.snapshot_id),
        ]
        .map(|scan| {
            scan.with_filter(Reference::new("x").greater_than_or_equal_to(Datum::long(100)))
                .build()
                .unwrap()
        });
        let current = table.scan().build().unwrap();
        let mut planned_files = Vec::new();
        let mut rows = Vec::new();
        let mut schemas = Vec::new();
        for scan in [by_time, by_id, current] {
            let mut tasks: Vec<_> = scan
                .plan_files()
                .await
                .unwrap()
                .try_collect()
                .await
                .unwrap();
            tasks.sort_by_key(|task| task.data_file_path().to_string());
            planned_files.push(tasks);
            let batches: Vec<_> = scan.to_arrow().await.unwrap().try_collect().await.unwrap();
            assert!(!batches.is_empty());
            schemas.push(batches[0].schema());
            let mut values: Vec<i64> = batches
                .iter()
                .flat_map(|batch| {
                    batch
                        .column_by_name("x")
                        .unwrap()
                        .as_any()
                        .downcast_ref::<Int64Array>()
                        .unwrap()
                        .values()
                        .iter()
                        .copied()
                })
                .collect();
            values.sort_unstable();
            rows.push(values);
        }
        assert!(!planned_files[0].is_empty());
        assert_eq!(planned_files[0], planned_files[1]);
        assert_eq!(rows[0], vec![100]);
        assert_eq!(rows[0], rows[1]);
        // The current snapshot contains two live fixture files.
        assert_eq!(rows[2], vec![1; 2 * ROWS_PER_FIXTURE_FILE]);
        assert_eq!(schemas[0], schemas[1]);
        assert_eq!(schemas[0].fields().len(), 1);
        assert_eq!(schemas[2].fields().len(), 8);
    }

    /// A scan of the current state reads the current snapshot with the table's
    /// current schema, not the older one that snapshot was written with.
    #[test]
    fn test_scan_of_current_state_uses_current_schema() {
        let table = TableTestFixture::new_with_column_added_after_last_write().table;

        let table_scan = table.scan().build().unwrap();
        assert_eq!(table_scan.snapshot(), table.metadata().current_snapshot());
        assert_eq!(
            table_scan.plan_context.as_ref().unwrap().snapshot_schema,
            *table.metadata().current_schema()
        );
    }

    /// A scan pinned to a snapshot uses the schema that snapshot was written
    /// with, even when it is the current snapshot.
    #[test]
    fn test_time_travel_scan_uses_snapshot_schema() {
        let table = TableTestFixture::new_with_column_added_after_last_write().table;
        let snapshot = table.metadata().current_snapshot().unwrap();

        let table_scan = table
            .scan()
            .snapshot_id(snapshot.snapshot_id())
            .build()
            .unwrap();
        assert_eq!(table_scan.snapshot(), Some(snapshot));
        assert_eq!(
            table_scan.plan_context.as_ref().unwrap().snapshot_schema,
            snapshot.schema(table.metadata()).unwrap()
        );
    }

    fn table_with_property(key: &str, value: &str) -> Table {
        let fixture = TableTestFixture::new();
        let mut metadata = fixture.table.metadata().clone();
        metadata
            .properties
            .insert(key.to_string(), value.to_string());
        Table::builder()
            .metadata(metadata)
            .identifier(fixture.table.identifier().clone())
            .file_io(fixture.table.file_io().clone())
            .metadata_location(fixture.table.metadata_location().unwrap().to_string())
            .runtime(test_runtime())
            .build()
            .unwrap()
    }

    #[test]
    fn test_table_scan_without_name_mapping_property() {
        let table = TableTestFixture::new().table;

        let table_scan = table.scan().build().unwrap();
        assert!(
            table_scan
                .plan_context
                .as_ref()
                .unwrap()
                .name_mapping
                .is_none()
        );
    }

    #[test]
    fn test_table_scan_with_name_mapping_property() {
        let mapping_json = r#"[{"field-id":1,"names":["id","record_id"]}]"#;
        let table =
            table_with_property(TableProperties::PROPERTY_DEFAULT_NAME_MAPPING, mapping_json);

        let table_scan = table.scan().build().unwrap();
        let mapping = table_scan
            .plan_context
            .as_ref()
            .unwrap()
            .name_mapping
            .as_ref()
            .expect("name_mapping should be parsed from the table property");
        let fields = mapping.fields();
        assert_eq!(fields.len(), 1);
        assert_eq!(fields[0].field_id(), Some(1));
        assert_eq!(fields[0].names(), &[
            "id".to_string(),
            "record_id".to_string()
        ]);
    }

    #[test]
    fn test_table_scan_with_malformed_name_mapping_property() {
        let table = table_with_property(
            TableProperties::PROPERTY_DEFAULT_NAME_MAPPING,
            "{ not valid json",
        );

        let err = table
            .scan()
            .build()
            .expect_err("malformed name mapping should fail to parse");
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
    }

    #[tokio::test]
    async fn test_plan_files_carries_name_mapping_into_file_scan_task() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        let mapping_json = r#"[{"field-id":1,"names":["id","record_id"]}]"#;
        let mut metadata = fixture.table.metadata().clone();
        metadata.properties.insert(
            TableProperties::PROPERTY_DEFAULT_NAME_MAPPING.to_string(),
            mapping_json.to_string(),
        );
        let table = Table::builder()
            .metadata(metadata)
            .identifier(fixture.table.identifier().clone())
            .file_io(fixture.table.file_io().clone())
            .metadata_location(fixture.table.metadata_location().unwrap().to_string())
            .runtime(test_runtime())
            .build()
            .unwrap();

        let tasks: Vec<_> = table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        assert!(!tasks.is_empty(), "expected at least one FileScanTask");
        for task in &tasks {
            let mapping = task
                .name_mapping()
                .expect("name_mapping should reach the FileScanTask");
            assert_eq!(mapping.fields().len(), 1);
            assert_eq!(mapping.fields()[0].field_id(), Some(1));
        }
    }

    #[tokio::test]
    async fn test_plan_files_carries_sort_order_into_file_scan_task() {
        let mut fixture = TableTestFixture::new();

        // Inject the reserved unsorted order (id 0) inline rather than editing the shared
        // testdata fixture, so the id-0 file below exercises the `!is_unsorted()` filter
        // branch instead of the `and_then` short-circuit an absent entry would take.
        let mut metadata = fixture.table.metadata().clone();
        metadata
            .sort_orders
            .insert(0, Arc::new(SortOrder::unsorted_order()));
        fixture.table = fixture.table.with_metadata(Arc::new(metadata));

        let expected_sort_order = fixture
            .table
            .metadata()
            .sort_order_by_id(3)
            .unwrap()
            .clone();

        // sort_order_ids: resolvable (3), absent, unresolvable (99), reserved unsorted (0).
        fixture
            .setup_manifest_files_with_sort_order_ids([Some(3), None, Some(99), Some(0)])
            .await;

        let tasks: Vec<_> = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        assert_eq!(tasks.len(), 4, "expected all four FileScanTasks");

        // Aggregates catch a systemic regression (every entry resolving to id 3, or
        // resolution dropping entirely) that the per-file checks below would each still pass.
        assert_eq!(
            tasks.iter().filter(|t| t.sort_order().is_some()).count(),
            1,
            "exactly one file resolves to a sort order"
        );
        assert_eq!(
            tasks.iter().filter(|t| t.sort_order_id().is_some()).count(),
            3,
            "three files carry a raw sort_order_id"
        );

        let resolved = tasks
            .iter()
            .find(|t| t.data_file_path().ends_with("1.parquet"))
            .unwrap();
        assert_eq!(resolved.sort_order_id(), Some(3));
        assert_eq!(
            resolved.sort_order(),
            Some(&expected_sort_order),
            "sort_order_id 3 should resolve to the table's sort order at id 3"
        );

        let missing = tasks
            .iter()
            .find(|t| t.data_file_path().ends_with("2.parquet"))
            .unwrap();
        assert_eq!(missing.sort_order_id(), None);
        assert!(
            missing.sort_order().is_none(),
            "a file with no sort_order_id carries no sort_order"
        );

        let unresolvable = tasks
            .iter()
            .find(|t| t.data_file_path().ends_with("3.parquet"))
            .unwrap();
        assert_eq!(
            unresolvable.sort_order_id(),
            Some(99),
            "the raw id is preserved even when it does not resolve"
        );
        assert!(
            unresolvable.sort_order().is_none(),
            "an unresolvable sort_order_id resolves to no sort_order"
        );

        let unsorted = tasks
            .iter()
            .find(|t| t.data_file_path().ends_with("4.parquet"))
            .unwrap();
        assert_eq!(unsorted.sort_order_id(), Some(0));
        assert!(
            unsorted.sort_order().is_none(),
            "the reserved unsorted order (id 0) resolves to no sort_order"
        );
    }

    #[tokio::test]
    async fn test_plan_files_on_table_without_any_snapshots() {
        let table = TableTestFixture::new_empty().table;
        let batch_stream = table.scan().build().unwrap().to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert!(batches.is_empty());
    }

    #[tokio::test]
    async fn test_plan_files_no_deletions() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Create table scan for current snapshot and plan files
        let table_scan = fixture
            .table
            .scan()
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let mut tasks = table_scan
            .plan_files()
            .await
            .unwrap()
            .try_fold(vec![], |mut acc, task| async move {
                acc.push(task);
                Ok(acc)
            })
            .await
            .unwrap();

        assert_eq!(tasks.len(), 2);

        tasks.sort_by_key(|t| t.data_file_path().to_string());

        // Check first task is added data file
        assert_eq!(
            tasks[0].data_file_path(),
            format!("{}/1.parquet", &fixture.table_location)
        );

        // Check second task is existing data file
        assert_eq!(
            tasks[1].data_file_path(),
            format!("{}/3.parquet", &fixture.table_location)
        );
    }

    #[tokio::test]
    async fn test_scan_widens_old_partition_so_pruning_and_deletes_match() {
        // Guards the reader-side normalization through a real scan: a data manifest
        // written when the partition source `x` was `int` must have its `int(7)`
        // partition widened to `long(7)` during planning, so (a) a `x = 7` (long)
        // filter keeps the file and (b) the equality delete keyed by `long(7)`
        // attaches. Without the normalization the int/long mismatch drops the delete
        // (and prunes the file under the filter).
        let mut fixture = TableTestFixture::new();
        fixture.setup_promoted_partition_with_eq_delete(false).await;

        let tasks: Vec<FileScanTask> = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(
            tasks[0].deletes().len(),
            1,
            "equality delete must attach after widening the old int partition to long"
        );
        assert_eq!(
            tasks[0].partition(),
            Some(&Struct::from_iter([Some(Literal::long(7))])),
            "the widened long(7) tuple must reach FileScanTask::partition()"
        );

        let tasks: Vec<FileScanTask> = fixture
            .table
            .scan()
            .with_filter(Reference::new("x").equal_to(Datum::long(7)))
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(
            tasks.len(),
            1,
            "filter on the promoted type must not prune the file"
        );
        assert_eq!(tasks[0].deletes().len(), 1);

        let tasks: Vec<FileScanTask> = fixture
            .table
            .scan()
            .with_filter(Reference::new("x").equal_to(Datum::long(8)))
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert!(
            tasks.is_empty(),
            "a file in partition x=7 must be pruned by an x=8 filter"
        );
    }

    #[tokio::test]
    async fn test_scan_widens_old_partition_with_pre_promotion_delete() {
        // Data file and equality delete both written before the promotion (int
        // partition). They matched on main; after both sides widen to long they must
        // still match, so the delete stays attached.
        let mut fixture = TableTestFixture::new();
        fixture.setup_promoted_partition_with_eq_delete(true).await;

        let tasks: Vec<FileScanTask> = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].deletes().len(), 1);
    }

    #[tokio::test]
    async fn test_scan_widens_old_partition_so_deletion_vector_matches() {
        // A v3 deletion vector written after the promotion (long partition) references
        // an older data file (int partition). The widening must make the partitions
        // agree so the DV attaches, instead of failing the DV partition check.
        let mut fixture = TableTestFixture::new();
        fixture.setup_promoted_partition_with_dv().await;

        let tasks: Vec<FileScanTask> = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].deletes().len(), 1);
    }

    #[tokio::test]
    async fn test_scan_keeps_eq_delete_after_dropping_promoted_source() {
        // Partition source `x` promoted int -> long, then dropped. The data file (int)
        // and its equality delete (long) must still normalize to the same long partition
        // so the delete stays attached; otherwise it silently detaches.
        let mut fixture = TableTestFixture::new();
        fixture.setup_promoted_partition_with_eq_delete(false).await;
        fixture.drop_partition_source(1);
        let table = fixture.table;

        let tasks: Vec<FileScanTask> = table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].deletes().len(), 1);
    }

    #[tokio::test]
    async fn test_scan_keeps_deletion_vector_after_dropping_promoted_source() {
        // Same as above for a deletion vector: without widening the dropped numeric
        // source the DV partition check errors (long(7) vs int(7)).
        let mut fixture = TableTestFixture::new();
        fixture.setup_promoted_partition_with_dv().await;
        fixture.drop_partition_source(1);
        let table = fixture.table;

        let tasks: Vec<FileScanTask> = table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].deletes().len(), 1);
    }

    #[tokio::test]
    async fn test_scan_keeps_eq_delete_when_dropped_source_is_date() {
        // A promoted numeric source `x` sits beside a dropped `date` source `d`. The gate
        // fires on `x`'s int -> long widening, but `d` (stored as an int) must be copied
        // through, not widened to long, or the data file's partition diverges from the
        // equality delete's and detaches it.
        let mut fixture = TableTestFixture::new();
        fixture
            .setup_promoted_partition_with_eq_delete_and_dropped_date()
            .await;

        let tasks: Vec<FileScanTask> = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(tasks[0].deletes().len(), 1);
    }

    #[tokio::test]
    async fn test_plan_files_carries_row_lineage_into_file_scan_task() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        let mut tasks: Vec<_> = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        assert_eq!(tasks.len(), 2);
        tasks.sort_by_key(|task| task.data_file_path().to_string());

        // The added file inherits the current snapshot's data sequence number,
        // the existing file keeps the one it was written with.
        assert_eq!(
            tasks[0].data_file_path(),
            format!("{}/1.parquet", &fixture.table_location)
        );
        assert_eq!(tasks[0].data_sequence_number(), Some(1));
        assert_eq!(
            tasks[1].data_file_path(),
            format!("{}/3.parquet", &fixture.table_location)
        );
        assert_eq!(tasks[1].data_sequence_number(), Some(0));

        // first_row_id is a v3 concept; a v2 manifest carries none.
        assert!(tasks.iter().all(|task| task.first_row_id().is_none()));
    }

    #[tokio::test]
    async fn test_plan_files_carries_row_lineage_from_v3_manifest() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_v3_manifest_files().await;

        let task = fixture
            .table
            .scan()
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect::<Vec<_>>()
            .await
            .unwrap()
            .into_iter()
            .next()
            .expect("expected one FileScanTask");

        // The manifest-level first_row_id (42) is inherited onto the entry on
        // read, then carried onto the task.
        assert_eq!(task.first_row_id(), Some(42));
        // The data sequence number is threaded through the same v3 read path.
        assert_eq!(task.data_sequence_number(), Some(1));
    }

    #[tokio::test]
    async fn test_filtered_scan_with_dropped_partition_source_column() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Baseline: the same filtered scan against the table before evolution.
        let baseline = scan_y_gte_5(&fixture.table).await;
        assert!(!baseline.is_empty());
        assert!(baseline.iter().all(|y| *y >= 5));

        // Evolve the table so that the manifests reference a historical spec whose source
        // column is no longer in the current schema: make an unpartitioned spec the
        // default, then drop the original spec's source column from the schema.
        let current_schema = fixture.table.metadata().current_schema();
        let evolved_schema = Schema::builder()
            .with_fields(
                current_schema
                    .as_struct()
                    .fields()
                    .iter()
                    .filter(|field| field.id != 1)
                    .cloned(),
            )
            .with_identifier_field_ids(vec![2])
            .build()
            .unwrap();
        let evolved =
            TableMetadataBuilder::new_from_metadata(fixture.table.metadata().clone(), None)
                .add_default_partition_spec(UnboundPartitionSpec::builder().build())
                .unwrap()
                .add_current_schema(evolved_schema)
                .unwrap()
                .build()
                .unwrap()
                .metadata;

        // a commit after the evolution carries the previous manifests forward: the new
        // snapshot uses the evolved schema while its manifests still use historical spec 0
        let parent = evolved.current_snapshot().unwrap().clone();
        let snapshot = Snapshot::builder()
            .with_snapshot_id(parent.snapshot_id() + 1)
            .with_parent_snapshot_id(Some(parent.snapshot_id()))
            .with_sequence_number(evolved.last_sequence_number() + 1)
            .with_timestamp_ms(evolved.last_updated_ms + 1)
            .with_schema_id(evolved.current_schema_id())
            .with_manifest_list(parent.manifest_list())
            .with_summary(Summary {
                operation: Operation::Append,
                additional_properties: HashMap::new(),
            })
            .build();
        let metadata = TableMetadataBuilder::new_from_metadata(evolved, None)
            .set_branch_snapshot(snapshot, MAIN_BRANCH)
            .unwrap()
            .build()
            .unwrap()
            .metadata;
        let table = fixture.table.clone().with_metadata(Arc::new(metadata));

        // Planning and reading must succeed, and the results must match the table before
        // evolution: no rows wrongly pruned and none returned unfiltered.
        let evolved = scan_y_gte_5(&table).await;
        assert_eq!(evolved, baseline);
    }

    async fn scan_y_gte_5(table: &Table) -> Vec<i64> {
        let table_scan = table
            .scan()
            .select(["y"])
            .with_filter(Reference::new("y").greater_than_or_equal_to(Datum::long(5)))
            .build()
            .unwrap();
        let batches: Vec<_> = table_scan
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        let mut values: Vec<i64> = batches
            .iter()
            .flat_map(|batch| {
                let col = batch.column_by_name("y").unwrap();
                let arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
                (0..arr.len()).map(|i| arr.value(i)).collect::<Vec<_>>()
            })
            .collect();
        values.sort_unstable();
        values
    }

    #[tokio::test]
    async fn test_open_parquet_no_deletions() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Create table scan for current snapshot and plan files
        let table_scan = fixture
            .table
            .scan()
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        let col = batches[0].column_by_name("x").unwrap();

        let int64_arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 1);
    }

    #[tokio::test]
    async fn test_open_parquet_no_deletions_by_separate_reader() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Create table scan for current snapshot and plan files
        let table_scan = fixture
            .table
            .scan()
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let mut plan_task: Vec<_> = table_scan
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(plan_task.len(), 2);

        let reader = ArrowReaderBuilder::new(
            fixture.table.file_io().clone(),
            fixture.table.runtime().clone(),
        )
        .build();
        let batch_stream = reader
            .clone()
            .read(Box::pin(stream::iter(vec![Ok(plan_task.remove(0))])))
            .unwrap()
            .stream();
        let batch_1: Vec<_> = batch_stream.try_collect().await.unwrap();

        let reader = ArrowReaderBuilder::new(
            fixture.table.file_io().clone(),
            fixture.table.runtime().clone(),
        )
        .build();
        let batch_stream = reader
            .read(Box::pin(stream::iter(vec![Ok(plan_task.remove(0))])))
            .unwrap()
            .stream();
        let batch_2: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batch_1, batch_2);
    }

    #[tokio::test]
    async fn test_open_parquet_with_projection() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Create table scan for current snapshot and plan files
        let table_scan = fixture
            .table
            .scan()
            .select(["x", "z"])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_columns(), 2);

        let col1 = batches[0].column_by_name("x").unwrap();
        let int64_arr = col1.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 1);

        let col2 = batches[0].column_by_name("z").unwrap();
        let int64_arr = col2.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 3);

        // test empty scan
        let table_scan = fixture.table.scan().select_empty().build().unwrap();
        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_columns(), 0);
        assert_eq!(batches[0].num_rows(), 1024);
    }

    #[tokio::test]
    async fn test_filter_on_arrow_lt() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: y < 3
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("y").less_than(Datum::long(3));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_rows(), 512);

        let col = batches[0].column_by_name("x").unwrap();
        let int64_arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 1);

        let col = batches[0].column_by_name("y").unwrap();
        let int64_arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 2);
    }

    #[tokio::test]
    async fn test_filter_on_arrow_gt_eq() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: y >= 5
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("y").greater_than_or_equal_to(Datum::long(5));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_rows(), 12);

        let col = batches[0].column_by_name("x").unwrap();
        let int64_arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 1);

        let col = batches[0].column_by_name("y").unwrap();
        let int64_arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 5);
    }

    #[tokio::test]
    async fn test_filter_double_eq() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: dbl == 150.0
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("dbl").equal_to(Datum::double(150.0f64));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches.len(), 2);
        assert_eq!(batches[0].num_rows(), 12);

        let col = batches[0].column_by_name("dbl").unwrap();
        let f64_arr = col.as_any().downcast_ref::<Float64Array>().unwrap();
        assert_eq!(f64_arr.value(1), 150.0f64);
    }

    #[tokio::test]
    async fn test_filter_int_eq() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: i32 == 150
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("i32").equal_to(Datum::int(150i32));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches.len(), 2);
        assert_eq!(batches[0].num_rows(), 12);

        let col = batches[0].column_by_name("i32").unwrap();
        let i32_arr = col.as_any().downcast_ref::<Int32Array>().unwrap();
        assert_eq!(i32_arr.value(1), 150i32);
    }

    #[tokio::test]
    async fn test_filter_long_eq() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: i64 == 150
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("i64").equal_to(Datum::long(150i64));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches.len(), 2);
        assert_eq!(batches[0].num_rows(), 12);

        let col = batches[0].column_by_name("i64").unwrap();
        let i64_arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(i64_arr.value(1), 150i64);
    }

    #[tokio::test]
    async fn test_filter_bool_eq() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: bool == true
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("bool").equal_to(Datum::bool(true));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches.len(), 2);
        assert_eq!(batches[0].num_rows(), 512);

        let col = batches[0].column_by_name("bool").unwrap();
        let bool_arr = col.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(bool_arr.value(1));
    }

    #[tokio::test]
    async fn test_filter_on_arrow_is_null() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: y is null
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("y").is_null();
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches.len(), 0);
    }

    #[tokio::test]
    async fn test_filter_on_arrow_is_not_null() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: y is not null
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("y").is_not_null();
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches[0].num_rows(), 1024);
    }

    #[tokio::test]
    async fn test_filter_on_arrow_lt_and_gt() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: y < 5 AND z >= 4
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("y")
            .less_than(Datum::long(5))
            .and(Reference::new("z").greater_than_or_equal_to(Datum::long(4)));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches[0].num_rows(), 500);

        let col = batches[0].column_by_name("x").unwrap();
        let expected_x = Arc::new(Int64Array::from_iter_values(vec![1; 500])) as ArrayRef;
        assert_eq!(col, &expected_x);

        let col = batches[0].column_by_name("y").unwrap();
        let mut values = vec![];
        values.append(vec![3; 200].as_mut());
        values.append(vec![4; 300].as_mut());
        let expected_y = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
        assert_eq!(col, &expected_y);

        let col = batches[0].column_by_name("z").unwrap();
        let expected_z = Arc::new(Int64Array::from_iter_values(vec![4; 500])) as ArrayRef;
        assert_eq!(col, &expected_z);
    }

    #[tokio::test]
    async fn test_filter_on_arrow_lt_or_gt() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: y < 5 AND z >= 4
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("y")
            .less_than(Datum::long(5))
            .or(Reference::new("z").greater_than_or_equal_to(Datum::long(4)));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches[0].num_rows(), 1024);

        let col = batches[0].column_by_name("x").unwrap();
        let expected_x = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
        assert_eq!(col, &expected_x);

        let col = batches[0].column_by_name("y").unwrap();
        let mut values = vec![2; 512];
        values.append(vec![3; 200].as_mut());
        values.append(vec![4; 300].as_mut());
        values.append(vec![5; 12].as_mut());
        let expected_y = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
        assert_eq!(col, &expected_y);

        let col = batches[0].column_by_name("z").unwrap();
        let mut values = vec![3; 512];
        values.append(vec![4; 512].as_mut());
        let expected_z = Arc::new(Int64Array::from_iter_values(values)) as ArrayRef;
        assert_eq!(col, &expected_z);
    }

    #[tokio::test]
    async fn test_filter_on_arrow_startswith() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: a STARTSWITH "Ice"
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("a").starts_with(Datum::string("Ice"));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_rows(), 512);

        let col = batches[0].column_by_name("a").unwrap();
        let string_arr = col.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(string_arr.value(0), "Iceberg");
    }

    #[tokio::test]
    async fn test_filter_on_arrow_not_startswith() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: a NOT STARTSWITH "Ice"
        let mut builder = fixture.table.scan();
        let predicate = Reference::new("a").not_starts_with(Datum::string("Ice"));
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_rows(), 512);

        let col = batches[0].column_by_name("a").unwrap();
        let string_arr = col.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(string_arr.value(0), "Apache");
    }

    #[tokio::test]
    async fn test_filter_on_arrow_in() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: a IN ("Sioux", "Iceberg")
        let mut builder = fixture.table.scan();
        let predicate =
            Reference::new("a").is_in([Datum::string("Sioux"), Datum::string("Iceberg")]);
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_rows(), 512);

        let col = batches[0].column_by_name("a").unwrap();
        let string_arr = col.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(string_arr.value(0), "Iceberg");
    }

    #[tokio::test]
    async fn test_filter_on_arrow_not_in() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Filter: a NOT IN ("Sioux", "Iceberg")
        let mut builder = fixture.table.scan();
        let predicate =
            Reference::new("a").is_not_in([Datum::string("Sioux"), Datum::string("Iceberg")]);
        builder = builder
            .with_filter(predicate)
            .with_row_selection_enabled(true);
        let table_scan = builder.build().unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_rows(), 512);

        let col = batches[0].column_by_name("a").unwrap();
        let string_arr = col.as_any().downcast_ref::<StringArray>().unwrap();
        assert_eq!(string_arr.value(0), "Apache");
    }

    fn file_scan_task_test_schema(primitive_type: PrimitiveType) -> Arc<Schema> {
        Arc::new(
            Schema::builder()
                .with_fields(vec![Arc::new(NestedField::required(
                    1,
                    "x",
                    Type::Primitive(primitive_type),
                ))])
                .build()
                .unwrap(),
        )
    }

    fn assert_file_scan_task_serde_round_trip(task: FileScanTask) {
        // Regression test for https://github.com/apache/iceberg-rust/issues/3089.
        let serialized = serde_json::to_string(&task).unwrap();
        let deserialized: FileScanTask = serde_json::from_str(&serialized).unwrap();

        assert_eq!(task, deserialized);
    }

    fn file_scan_task_with_partition(
        primitive_type: PrimitiveType,
        transform: Transform,
        partition_value: Literal,
    ) -> FileScanTask {
        let schema = file_scan_task_test_schema(primitive_type);
        let partition_spec = Arc::new(
            PartitionSpec::builder(schema.clone())
                .add_partition_field("x", "x_partition", transform)
                .unwrap()
                .build()
                .unwrap(),
        );
        FileScanTask::builder()
            .with_data_file_path("data_file_path".to_string())
            .with_file_size_in_bytes(123)
            .with_start(10)
            .with_length(100)
            .with_project_field_ids(vec![1])
            .with_schema(schema)
            .with_data_file_format(DataFileFormat::Parquet)
            .with_partition(Some(Struct::from_iter([Some(partition_value)])))
            .with_partition_spec(Some(partition_spec))
            .with_case_sensitive(true)
            .build()
            .unwrap()
    }

    #[test]
    fn test_file_scan_task_serde_without_predicate() {
        let task = FileScanTask::builder()
            .with_data_file_path("data_file_path".to_string())
            .with_file_size_in_bytes(0)
            .with_start(0)
            .with_length(100)
            .with_project_field_ids(vec![1, 2, 3])
            .with_schema(file_scan_task_test_schema(PrimitiveType::Binary))
            .with_record_count(Some(100))
            .with_first_row_id(Some(1000))
            .with_data_sequence_number(Some(5))
            .with_data_file_format(DataFileFormat::Parquet)
            .with_case_sensitive(false)
            .build()
            .unwrap();
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_predicate() {
        let task = FileScanTask::builder()
            .with_data_file_path("data_file_path".to_string())
            .with_file_size_in_bytes(0)
            .with_start(0)
            .with_length(100)
            .with_project_field_ids(vec![1, 2, 3])
            .with_predicate(Some(BoundPredicate::AlwaysTrue))
            .with_schema(file_scan_task_test_schema(PrimitiveType::Binary))
            .with_data_file_format(DataFileFormat::Avro)
            .with_case_sensitive(false)
            .build()
            .unwrap();

        let serialized = serde_json::to_value(&task).unwrap();
        assert!(serialized.get("record_count").is_none());
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_unpartitioned_file_scan_task_serde() {
        let task = FileScanTask::builder()
            .with_data_file_path("data_file_path".to_string())
            .with_file_size_in_bytes(0)
            .with_start(0)
            .with_length(100)
            .with_project_field_ids(vec![1, 2, 3])
            .with_schema(file_scan_task_test_schema(PrimitiveType::Binary))
            .with_data_file_format(DataFileFormat::Parquet)
            .with_partition(Some(Struct::empty()))
            .with_case_sensitive(false)
            .build()
            .unwrap();
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_all_optional_fields() {
        let schema = file_scan_task_test_schema(PrimitiveType::Long);
        let partition_spec = Arc::new(
            PartitionSpec::builder(schema.clone())
                .add_partition_field("x", "x", Transform::Identity)
                .unwrap()
                .build()
                .unwrap(),
        );
        let unified_partition_type = Arc::new(partition_spec.partition_type(&schema).unwrap());
        let sort_order = Arc::new(
            SortOrder::builder()
                .with_order_id(1)
                .with_sort_field(
                    SortField::builder()
                        .source_id(1)
                        .transform(Transform::Identity)
                        .direction(SortDirection::Ascending)
                        .null_order(NullOrder::First)
                        .build(),
                )
                .build(&schema)
                .unwrap(),
        );
        let task = FileScanTask::builder()
            .with_data_file_path("data_file_path".to_string())
            .with_file_size_in_bytes(123)
            .with_start(10)
            .with_length(100)
            .with_project_field_ids(vec![1])
            .with_schema(schema)
            .with_data_file_format(DataFileFormat::Parquet)
            .with_deletes(vec![
                FileScanTaskDeleteFile::builder()
                    .with_file_path("delete_file_path".to_string())
                    .with_file_size_in_bytes(23)
                    .with_file_type(DataContentType::EqualityDeletes)
                    .with_file_format(DataFileFormat::Parquet)
                    .with_partition_spec_id(0)
                    .with_equality_ids(Some(vec![1]))
                    .with_referenced_data_file(Some("data_file_path".to_string()))
                    .with_content_offset(Some(12))
                    .with_content_size_in_bytes(Some(34))
                    .with_record_count(Some(5))
                    .with_key_metadata(Some(vec![4, 5, 6].into_boxed_slice()))
                    .build()
                    .unwrap(),
            ])
            .with_partition(Some(Struct::from_iter([Some(Literal::long(42))])))
            .with_partition_spec(Some(partition_spec))
            .with_name_mapping(Some(Arc::new(NameMapping::new(vec![MappedField::new(
                Some(1),
                vec!["x".to_string()],
                vec![],
            )]))))
            .with_unified_partition_type(Some(unified_partition_type))
            .with_sort_order_id(Some(1))
            .with_sort_order(Some(sort_order))
            .with_case_sensitive(true)
            .with_key_metadata(Some(vec![1, 2, 3].into_boxed_slice()))
            .build()
            .unwrap();
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_date_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::Date,
            Transform::Identity,
            Literal::date(19_000),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_timestamp_ns_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::TimestampNs,
            Transform::Identity,
            Literal::timestamp_nano(1_510_871_468_123_456_789),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_timestamptz_ns_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::TimestamptzNs,
            Transform::Identity,
            Literal::timestamptz_nano(1_510_871_468_123_456_789),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_decimal_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::Decimal {
                precision: 9,
                scale: 2,
            },
            Transform::Identity,
            Literal::decimal(12_345),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_uuid_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::Uuid,
            Transform::Identity,
            Literal::uuid(Uuid::from_u128(0x12345678_90ab_cdef_1234_567890abcdef)),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_fixed_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::Fixed(4),
            Transform::Identity,
            Literal::fixed([1, 2, 3, 4]),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[test]
    fn test_file_scan_task_serde_with_bucket_partition() {
        let task = file_scan_task_with_partition(
            PrimitiveType::String,
            Transform::Bucket(4),
            Literal::int(2),
        );
        assert_file_scan_task_serde_round_trip(task);
    }

    #[tokio::test]
    async fn test_select_with_file_column() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select regular columns plus the _file column
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_FILE])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Verify we have 2 columns: x and _file
        assert_eq!(batches[0].num_columns(), 2);

        // Verify the x column exists and has correct data
        let x_col = batches[0].column_by_name("x").unwrap();
        let x_arr = x_col.as_primitive::<arrow_array::types::Int64Type>();
        assert_eq!(x_arr.value(0), 1);

        // Verify the _file column exists
        let file_col = batches[0].column_by_name(RESERVED_COL_NAME_FILE);
        assert!(
            file_col.is_some(),
            "_file column should be present in the batch"
        );

        // Verify the _file column contains a file path
        let file_col = file_col.unwrap();
        assert!(
            matches!(
                file_col.data_type(),
                arrow_schema::DataType::RunEndEncoded(_, _)
            ),
            "_file column should use RunEndEncoded type"
        );

        // Decode the RunArray to verify it contains the file path
        let run_array = file_col
            .as_any()
            .downcast_ref::<RunArray<Int32Type>>()
            .expect("_file column should be a RunArray");

        let values = run_array.values();
        let string_values = values.as_string::<i32>();
        assert_eq!(string_values.len(), 1, "Should have a single file path");

        let file_path = string_values.value(0);
        assert!(
            file_path.ends_with(".parquet"),
            "File path should end with .parquet, got: {file_path}"
        );
    }

    #[tokio::test]
    async fn test_select_file_column_position() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select columns in specific order: x, _file, z
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_FILE, "z"])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_columns(), 3);

        // Verify column order: x at position 0, _file at position 1, z at position 2
        let schema = batches[0].schema();
        assert_eq!(schema.field(0).name(), "x");
        assert_eq!(schema.field(1).name(), RESERVED_COL_NAME_FILE);
        assert_eq!(schema.field(2).name(), "z");

        // Verify columns by name also works
        assert!(batches[0].column_by_name("x").is_some());
        assert!(batches[0].column_by_name(RESERVED_COL_NAME_FILE).is_some());
        assert!(batches[0].column_by_name("z").is_some());
    }

    #[tokio::test]
    async fn test_select_file_column_only() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select only the _file column
        let table_scan = fixture
            .table
            .scan()
            .select([RESERVED_COL_NAME_FILE])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Should have exactly 1 column
        assert_eq!(batches[0].num_columns(), 1);

        // Verify it's the _file column
        let schema = batches[0].schema();
        assert_eq!(schema.field(0).name(), RESERVED_COL_NAME_FILE);

        // Verify the batch has the correct number of rows
        // The scan reads files 1.parquet and 3.parquet (2.parquet is deleted)
        // Each file has 1024 rows, so total is 2048 rows
        let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
        assert_eq!(total_rows, 2048);
    }

    #[tokio::test]
    async fn test_file_column_with_multiple_files() {
        use std::collections::HashSet;

        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select x and _file columns
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_FILE])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Collect all unique file paths from the batches
        let mut file_paths = HashSet::new();
        for batch in &batches {
            let file_col = batch.column_by_name(RESERVED_COL_NAME_FILE).unwrap();
            let run_array = file_col
                .as_any()
                .downcast_ref::<RunArray<Int32Type>>()
                .expect("_file column should be a RunArray");

            let values = run_array.values();
            let string_values = values.as_string::<i32>();
            for i in 0..string_values.len() {
                file_paths.insert(string_values.value(i).to_string());
            }
        }

        // We should have multiple files (the test creates 1.parquet and 3.parquet)
        assert!(!file_paths.is_empty(), "Should have at least one file path");

        // All paths should end with .parquet
        for path in &file_paths {
            assert!(
                path.ends_with(".parquet"),
                "All file paths should end with .parquet, got: {path}"
            );
        }
    }

    #[tokio::test]
    async fn test_file_column_at_start() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select _file at the start
        let table_scan = fixture
            .table
            .scan()
            .select([RESERVED_COL_NAME_FILE, "x", "y"])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_columns(), 3);

        // Verify _file is at position 0
        let schema = batches[0].schema();
        assert_eq!(schema.field(0).name(), RESERVED_COL_NAME_FILE);
        assert_eq!(schema.field(1).name(), "x");
        assert_eq!(schema.field(2).name(), "y");
    }

    #[tokio::test]
    async fn test_file_column_at_end() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select _file at the end
        let table_scan = fixture
            .table
            .scan()
            .select(["x", "y", RESERVED_COL_NAME_FILE])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        assert_eq!(batches[0].num_columns(), 3);

        // Verify _file is at position 2 (the end)
        let schema = batches[0].schema();
        assert_eq!(schema.field(0).name(), "x");
        assert_eq!(schema.field(1).name(), "y");
        assert_eq!(schema.field(2).name(), RESERVED_COL_NAME_FILE);
    }

    #[tokio::test]
    async fn test_select_with_repeated_column_names() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select with repeated column names - both regular columns and virtual columns
        // Repeated columns should appear multiple times in the result (duplicates are allowed)
        let table_scan = fixture
            .table
            .scan()
            .select([
                "x",
                RESERVED_COL_NAME_FILE,
                "x", // x repeated
                "y",
                RESERVED_COL_NAME_FILE, // _file repeated
                "y",                    // y repeated
            ])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Verify we have exactly 6 columns (duplicates are allowed and preserved)
        assert_eq!(
            batches[0].num_columns(),
            6,
            "Should have exactly 6 columns with duplicates"
        );

        let schema = batches[0].schema();

        // Verify columns appear in the exact order requested: x, _file, x, y, _file, y
        assert_eq!(schema.field(0).name(), "x", "Column 0 should be x");
        assert_eq!(
            schema.field(1).name(),
            RESERVED_COL_NAME_FILE,
            "Column 1 should be _file"
        );
        assert_eq!(
            schema.field(2).name(),
            "x",
            "Column 2 should be x (duplicate)"
        );
        assert_eq!(schema.field(3).name(), "y", "Column 3 should be y");
        assert_eq!(
            schema.field(4).name(),
            RESERVED_COL_NAME_FILE,
            "Column 4 should be _file (duplicate)"
        );
        assert_eq!(
            schema.field(5).name(),
            "y",
            "Column 5 should be y (duplicate)"
        );

        // Verify all columns have correct data types
        assert!(
            matches!(schema.field(0).data_type(), arrow_schema::DataType::Int64),
            "Column x should be Int64"
        );
        assert!(
            matches!(schema.field(2).data_type(), arrow_schema::DataType::Int64),
            "Column x (duplicate) should be Int64"
        );
        assert!(
            matches!(schema.field(3).data_type(), arrow_schema::DataType::Int64),
            "Column y should be Int64"
        );
        assert!(
            matches!(schema.field(5).data_type(), arrow_schema::DataType::Int64),
            "Column y (duplicate) should be Int64"
        );
        assert!(
            matches!(
                schema.field(1).data_type(),
                arrow_schema::DataType::RunEndEncoded(_, _)
            ),
            "_file column should use RunEndEncoded type"
        );
        assert!(
            matches!(
                schema.field(4).data_type(),
                arrow_schema::DataType::RunEndEncoded(_, _)
            ),
            "_file column (duplicate) should use RunEndEncoded type"
        );
    }

    /// Builds a minimal single-snapshot table (no manifests on disk) whose schema
    /// contains a column with the given name, so scan planning resolves column names
    /// against a real schema. `TableScan::build()` only reads metadata, not manifest
    /// files, so a snapshot pointing at a dummy manifest list is enough. Used to
    /// reproduce issue #2837.
    fn table_with_data_column(column_name: &str) -> Table {
        use crate::spec::{
            FormatVersion, MAIN_BRANCH, Operation, Snapshot, SnapshotReference, SnapshotRetention,
            SortOrder, Summary, TableMetadataBuilder, UnboundPartitionSpec,
        };

        let schema = Schema::builder()
            .with_schema_id(0)
            .with_fields(vec![
                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
                NestedField::required(2, column_name, Type::Primitive(PrimitiveType::Int)).into(),
            ])
            .build()
            .unwrap();

        let snapshot = Snapshot::builder()
            .with_snapshot_id(1)
            .with_timestamp_ms(1)
            .with_sequence_number(0)
            .with_schema_id(0)
            .with_manifest_list("/snap-1.avro")
            .with_summary(Summary {
                operation: Operation::Append,
                additional_properties: HashMap::new(),
            })
            .build();

        let metadata = TableMetadataBuilder::new(
            schema,
            UnboundPartitionSpec::builder().with_spec_id(0).build(),
            SortOrder::unsorted_order(),
            "s3://bucket/table".to_string(),
            FormatVersion::V2,
            HashMap::new(),
        )
        .unwrap()
        .add_snapshot(snapshot)
        .unwrap()
        .set_ref(MAIN_BRANCH, SnapshotReference {
            snapshot_id: 1,
            retention: SnapshotRetention::Branch {
                min_snapshots_to_keep: None,
                max_snapshot_age_ms: None,
                max_ref_age_ms: None,
            },
        })
        .unwrap()
        .build()
        .unwrap()
        .metadata;

        Table::builder()
            .metadata(metadata)
            .identifier(TableIdent::from_strs(["db", "table1"]).unwrap())
            .file_io(FileIO::new_with_fs())
            .runtime(test_runtime())
            .build()
            .unwrap()
    }

    /// A user data column named `pos` (a delete-file internal column name that is not a
    /// data-table metadata column) must be projectable rather than shadowed. Regression
    /// test for issue #2837.
    #[test]
    fn test_scan_projects_data_column_named_like_delete_file_column() {
        for column_name in ["pos", "file_path"] {
            let table = table_with_data_column(column_name);

            // Projecting the data column must succeed and resolve to its real field id (2),
            // not the reserved delete-file field id.
            let table_scan = table
                .scan()
                .select([column_name])
                .build()
                .unwrap_or_else(|e| panic!("scan of data column `{column_name}` failed: {e}"));

            assert_eq!(
                table_scan.plan_context.as_ref().unwrap().field_ids.as_ref(),
                &[2]
            );

            // The default projection (all columns) must resolve to the real field ids
            // too, not shadow the data column with a reserved delete-file id.
            let default_scan = table.scan().build().unwrap();
            assert_eq!(
                default_scan
                    .plan_context
                    .as_ref()
                    .unwrap()
                    .field_ids
                    .as_ref(),
                &[1, 2]
            );
        }
    }

    /// Projecting a genuinely absent column still fails with a clear "not found" error
    /// rather than being silently accepted as a metadata column.
    #[test]
    fn test_scan_rejects_unknown_column_named_like_delete_file_column() {
        // This table has no `pos` column (only `id` and `file_path`).
        let table = table_with_data_column("file_path");

        let err = table
            .scan()
            .select(["pos"])
            .build()
            .expect_err("projecting an absent column should fail");
        assert_eq!(err.kind(), ErrorKind::DataInvalid);
        assert!(err.to_string().contains("not found"));
    }

    #[tokio::test]
    async fn test_scan_deadlock() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_deadlock_manifests().await;

        // Create table scan with concurrency limit 1
        // This sets channel size to 1.
        // Data manifest has 10 entries -> will block producer.
        // Delete manifest is 2nd in list -> won't be processed.
        // Consumer 2 (Data) not started -> blocked.
        // Consumer 1 (Delete) waiting -> blocked.
        let table_scan = fixture
            .table
            .scan()
            .with_concurrency_limit(1)
            .build()
            .unwrap();

        // This should timeout/hang if deadlock exists
        // We can use tokio::time::timeout
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), async {
            table_scan
                .plan_files()
                .await
                .unwrap()
                .try_collect::<Vec<_>>()
                .await
        })
        .await;

        // Assert it finished (didn't timeout)
        assert!(result.is_ok(), "Scan timed out - deadlock detected");
    }

    #[tokio::test]
    async fn test_select_with_spec_id_column() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select regular columns plus the _spec_id column
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_SPEC_ID, "z"])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Verify we have 3 columns: x, _spec_id, and z
        assert_eq!(batches[0].num_columns(), 3);

        // Verify the x column exists and has correct data
        let col1 = batches[0].column_by_name("x").unwrap();
        let int64_arr = col1.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 1);

        // Verify the _spec_id column exists
        let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID);
        assert!(
            spec_id_col.is_some(),
            "_spec_id column should be present in the batch"
        );

        // Verify the _spec_id data type
        let spec_id_col = spec_id_col.unwrap();
        assert!(
            matches!(
                spec_id_col.data_type(),
                arrow_schema::DataType::RunEndEncoded(_, _)
            ),
            "_spec_id column should use RunEndEncoded type"
        );

        // Decode the RunArray to verify it contains the spec id
        let run_array = spec_id_col
            .as_any()
            .downcast_ref::<RunArray<Int32Type>>()
            .expect("_spec_id column should be a RunArray");

        let values = run_array.values();
        let int_values = values.as_primitive::<Int32Type>();
        assert_eq!(int_values.len(), 1, "Should have a single _spec_id");

        let spec_id = int_values.value(0);
        assert_eq!(spec_id, 0, "_spec_id should be 0, got: {spec_id}");

        // Verify 'z' column exists
        assert!(batches[0].column_by_name("z").is_some());
    }

    #[tokio::test]
    async fn test_select_with_last_updated_sequence_number_column() {
        // A v2 fixture: data files have a null first_row_id. Per the spec's Row
        // Lineage read rules, a file with a null first_row_id produces a null
        // _last_updated_sequence_number for all rows; both lineage columns are
        // gated on first_row_id.
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batches: Vec<_> = table_scan
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        // Every row's value is null (v2 files have no first_row_id).
        assert_last_updated_seq_all(&batches, None);
    }

    #[tokio::test]
    async fn test_select_with_last_updated_sequence_number_column_v3() {
        // A v3 fixture: the data file inherits a first_row_id and a data sequence
        // number through manifest read. End to end, the projected
        // _last_updated_sequence_number materializes to the file's data sequence
        // number for every row, exercising the full inherit, populate and
        // materialize wiring, not just a hand-built task.
        let mut fixture = TableTestFixture::new();
        fixture.setup_v3_manifest_files().await;

        // The added file inherits the current snapshot's sequence number; derive it
        // from the fixture rather than hardcoding so the assertion tracks the fixture.
        let expected_seq = fixture
            .table
            .metadata()
            .current_snapshot()
            .unwrap()
            .sequence_number();

        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_LAST_UPDATED_SEQUENCE_NUMBER])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batches: Vec<_> = table_scan
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        assert_last_updated_seq_all(&batches, Some(expected_seq));
    }

    #[tokio::test]
    async fn test_select_with_spec_id_column_from_unpartitioned_table() {
        let mut fixture = TableTestFixture::new_unpartitioned();
        fixture.setup_unpartitioned_manifest_files().await;

        // Select regular columns plus the _spec_id column
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_SPEC_ID])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Verify we have 2 columns: x and _spec_id
        assert_eq!(batches[0].num_columns(), 2);

        // Verify the _spec_id column exists
        let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID);
        assert!(
            spec_id_col.is_some(),
            "_spec_id column should be present in the batch"
        );

        // Verify the _spec_id data type
        let spec_id_col = spec_id_col.unwrap();
        assert!(
            matches!(
                spec_id_col.data_type(),
                arrow_schema::DataType::RunEndEncoded(_, _)
            ),
            "_spec_id column should use RunEndEncoded type"
        );

        // Decode the RunArray to verify it contains the spec id
        let run_array = spec_id_col
            .as_any()
            .downcast_ref::<RunArray<Int32Type>>()
            .expect("_spec_id column should be a RunArray");

        let values = run_array.values();
        let int_values = values.as_primitive::<Int32Type>();
        assert_eq!(int_values.len(), 1, "Should have a single _spec_id");

        let spec_id = int_values.value(0);
        assert_eq!(spec_id, 0, "_spec_id should be 0, got: {spec_id}");
    }

    #[tokio::test]
    async fn test_select_with_spec_id_column_with_partition_evolution() {
        let mut fixture = TableTestFixture::new_with_partition_evolution();
        fixture
            .setup_manifest_files_with_partition_evolution()
            .await;

        // Select regular columns plus the _spec_id column
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_SPEC_ID, "z"])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();

        // Verify the x column exists and has correct data
        let col1 = batches[0].column_by_name("x").unwrap();
        let int64_arr = col1.as_any().downcast_ref::<Int64Array>().unwrap();
        assert_eq!(int64_arr.value(0), 1);

        // Verify the _spec_id column exists
        let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID);
        assert!(
            spec_id_col.is_some(),
            "_spec_id column should be present in the batch"
        );

        // Verify the _spec_id data type
        let spec_id_col = spec_id_col.unwrap();
        assert!(
            matches!(
                spec_id_col.data_type(),
                arrow_schema::DataType::RunEndEncoded(_, _)
            ),
            "_spec_id column should use RunEndEncoded type"
        );

        // Decode the RunArray to verify it contains the spec id
        let run_array = spec_id_col
            .as_any()
            .downcast_ref::<RunArray<Int32Type>>()
            .expect("_spec_id column should be a RunArray");

        let values = run_array.values();
        let int_values = values.as_primitive::<Int32Type>();
        assert_eq!(int_values.len(), 1, "Should have a single _spec_id");

        let spec_id = int_values.value(0);
        assert_eq!(spec_id, 2, "_spec_id should be 2, got: {spec_id}");
    }

    #[tokio::test]
    async fn test_select_with_pos_and_file_columns() {
        use arrow_array::cast::AsArray;

        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // Select regular columns plus the _pos column
        let table_scan = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_POS, RESERVED_COL_NAME_FILE])
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches.len(), 2);

        // Examine batches are 1.paruqet and 3.parquet, 2.parquet is deleted.
        for batch in batches.iter() {
            // Verify we have 3 columns: x, _pos and _file
            assert_eq!(batch.num_columns(), 3);

            // Verify the x column exists and has correct data
            let x_col = batch.column_by_name("x").unwrap();
            let x_arr = x_col.as_primitive::<arrow_array::types::Int64Type>();
            assert_eq!(x_arr.value(0), 1);

            // The _pos column exists and verify it is Int64Array with the expected values
            let pos_col = batch.column(1);
            let pos_array: &Int64Array = pos_col
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("_pos column should be a Int64Array");
            assert_eq!(*pos_array, Int64Array::from_iter_values(0i64..1024));

            // Verify the _file column exists
            let file_col = batch.column_by_name(RESERVED_COL_NAME_FILE);
            assert!(
                file_col.is_some(),
                "_file column should be present in the batch"
            );
        }
    }

    #[tokio::test]
    async fn test_pos_column_at_start_with_filters() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // y is in [4, 5)
        let predicate = Reference::new("y")
            .greater_than(Datum::long(4i64))
            .and(Reference::new("y").less_than_or_equal_to(Datum::long(5i64)));
        // Select _pos at the start
        let table_scan = fixture
            .table
            .scan()
            .select([RESERVED_COL_NAME_POS, "x", "y"])
            .with_filter(predicate)
            .with_row_selection_enabled(true)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();
        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches.len(), 2);

        // Examine batches are 1.paruqet and 3.parquet, 2.parquet is deleted.
        for batch in batches.iter() {
            assert_eq!(batch.num_columns(), 3);
            assert_eq!(batch.num_rows(), 12);

            // Verify _pos is at position 0
            let schema = batch.schema();
            assert_eq!(schema.field(0).name(), RESERVED_COL_NAME_POS);
            assert_eq!(schema.field(1).name(), "x");
            assert_eq!(schema.field(2).name(), "y");

            let pos_col = batch.column(0);
            let pos_array: &Int64Array = pos_col
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("_pos column should be a Int64Array");
            assert_eq!(*pos_array, Int64Array::from_iter_values(1012i64..1024));
        }
    }

    #[tokio::test]
    async fn test_repeated_pos_column_with_filter() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_manifest_files().await;

        // a NOT STARTSWITH "Apa"
        let predicate = Reference::new("a").not_starts_with(Datum::string("Apa"));
        // Select '_pos' columns twice
        let table_scan = fixture
            .table
            .scan()
            .select([RESERVED_COL_NAME_POS, "a", RESERVED_COL_NAME_POS, "x"])
            .with_row_selection_enabled(true)
            .with_filter(predicate)
            .build()
            .unwrap();

        let batch_stream = table_scan.to_arrow().await.unwrap();

        let batches: Vec<_> = batch_stream.try_collect().await.unwrap();
        assert_eq!(batches.len(), 2);

        // Examine batches are 1.paruqet and 3.parquet, 2.parquet is deleted.
        for batch in batches.iter() {
            assert_eq!(batch.num_rows(), 512);

            // fetch the 1st _pos column by name and verify it is Int64Array with the expected values
            let pos_col = batch
                .column_by_name("_pos")
                .expect("_pos column should be present in the batch");
            let pos_array: &Int64Array = pos_col
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("_pos column should be a Int64Array");
            assert_eq!(*pos_array, Int64Array::from_iter_values(512i64..1024));

            // fetch the 2nd _pos column by index and verify it is Int64Array with the expected values
            let pos_col = batch.column(2);
            let pos_array: &Int64Array = pos_col
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("_pos column should be a Int64Array");
            assert_eq!(*pos_array, Int64Array::from_iter_values(512i64..1024));
        }
    }

    /// End-to-end through `TableScan`: a data file with three row groups planned
    /// as a single whole-file `FileScanTask` must yield contiguous, file-absolute
    /// `_pos` values (0..300) across the row-group boundaries.
    #[tokio::test]
    async fn test_pos_across_row_groups_via_table_scan() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_multi_row_group_manifest(&[]).await;

        // Planning must produce exactly one whole-file task with _pos projected and
        // no delete files, confirming TableScan does not sub-split the file.
        let tasks: Vec<_> = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_POS])
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1, "expected a single FileScanTask");
        let task = &tasks[0];
        assert!(
            task.project_field_ids().contains(&RESERVED_FIELD_ID_POS),
            "_pos field id must be projected into the FileScanTask"
        );
        assert_eq!(task.start(), 0, "TableScan should plan whole-file tasks");
        assert_eq!(task.length(), task.file_size_in_bytes());
        assert!(task.deletes().is_empty());

        // Reading that task yields absolute _pos 0..300 in order.
        let batches: Vec<_> = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_POS])
            .build()
            .unwrap()
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        let pos: Vec<i64> = batches
            .iter()
            .flat_map(|b| {
                b.column_by_name(RESERVED_COL_NAME_POS)
                    .expect("_pos column should be present")
                    .as_any()
                    .downcast_ref::<Int64Array>()
                    .expect("_pos column should be a Int64Array")
                    .values()
                    .to_vec()
            })
            .collect();
        assert_eq!(pos, (0..300).collect::<Vec<i64>>());

        // Sanity: x == 1000 + _pos, proving _pos aligns with the actual rows read.
        let x: Vec<i64> = batches
            .iter()
            .flat_map(|b| {
                b.column_by_name("x")
                    .unwrap()
                    .as_primitive::<arrow_array::types::Int64Type>()
                    .values()
                    .to_vec()
            })
            .collect();
        assert_eq!(x, (1000..1300).collect::<Vec<i64>>());
    }

    /// A positional delete file registered in the manifest must be attached to the planned
    /// `FileScanTask` and applied on read, while surviving `_pos` values stay file-absolute.
    #[tokio::test]
    async fn test_pos_with_positional_deletes_via_table_scan() {
        let mut fixture = TableTestFixture::new();
        // Delete file-absolute positions 150 (middle row group) and 299 (last row).
        fixture.setup_multi_row_group_manifest(&[150, 299]).await;

        // Planning must attach the positional delete file to the task.
        let tasks: Vec<_> = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_POS])
            .build()
            .unwrap()
            .plan_files()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();
        assert_eq!(tasks.len(), 1);
        assert_eq!(
            tasks[0].deletes().len(),
            1,
            "positional delete file should be planned into the task"
        );
        assert_eq!(
            tasks[0].deletes()[0].file_type(),
            DataContentType::PositionDeletes
        );

        // Reading applies the deletes; _pos must skip 150 and 299 and stay absolute.
        let batches: Vec<_> = fixture
            .table
            .scan()
            .select(["x", RESERVED_COL_NAME_POS])
            .build()
            .unwrap()
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        let pos: Vec<i64> = batches
            .iter()
            .flat_map(|b| {
                b.column_by_name(RESERVED_COL_NAME_POS)
                    .expect("_pos column should be present")
                    .as_any()
                    .downcast_ref::<Int64Array>()
                    .expect("_pos column should be a Int64Array")
                    .values()
                    .to_vec()
            })
            .collect();

        let total: usize = batches.iter().map(|b| b.num_rows()).sum();
        assert_eq!(
            total, 298,
            "two rows should be removed by positional deletes"
        );
        assert!(!pos.contains(&150) && !pos.contains(&299), "got {pos:?}");
        let expected: Vec<i64> = (0..150).chain(151..299).collect();
        assert_eq!(pos, expected);
    }

    /// A filter that only matches the middle row group (`y` in [1100, 1200)) and
    /// prunes the other two row groups by statistics, so only the middle row group
    /// is read. `_pos` must report the file-absolute positions 100..200 for those
    /// rows, not values reset to 0..100.
    ///
    /// `y` is a non-partition column, so pruning here is driven purely by Parquet
    /// row-group statistics (the TableTestFixture's partition column is `x`).
    #[tokio::test]
    async fn test_pos_reads_only_middle_row_group_via_filter() {
        let mut fixture = TableTestFixture::new();
        fixture.setup_multi_row_group_manifest(&[]).await;

        // Middle row group holds y = 1100..1200 at file positions 100..200.
        let predicate = Reference::new("y")
            .greater_than_or_equal_to(Datum::long(1100))
            .and(Reference::new("y").less_than(Datum::long(1200)));

        let batches: Vec<_> = fixture
            .table
            .scan()
            .select(["y", RESERVED_COL_NAME_POS])
            .with_filter(predicate)
            .with_row_group_filtering_enabled(true)
            .build()
            .unwrap()
            .to_arrow()
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        let total: usize = batches.iter().map(|b| b.num_rows()).sum();
        assert_eq!(total, 100, "only the middle row group should be read");

        let pos: Vec<i64> = batches
            .iter()
            .flat_map(|b| {
                b.column_by_name(RESERVED_COL_NAME_POS)
                    .expect("_pos column should be present")
                    .as_any()
                    .downcast_ref::<Int64Array>()
                    .expect("_pos column should be a Int64Array")
                    .values()
                    .to_vec()
            })
            .collect();
        assert_eq!(
            pos,
            (100..200).collect::<Vec<i64>>(),
            "_pos must be file-absolute for the middle row group"
        );

        // Cross-check: y == 1000 + _pos for every surviving row.
        let y: Vec<i64> = batches
            .iter()
            .flat_map(|b| {
                b.column_by_name("y")
                    .unwrap()
                    .as_primitive::<arrow_array::types::Int64Type>()
                    .values()
                    .to_vec()
            })
            .collect();
        assert_eq!(y, (1100..1200).collect::<Vec<i64>>());
    }
}