datafusion-ducklake 0.7.0

//! Table changes (CDC) functionality for DuckLake
//!
//! This module provides the `ducklake_table_changes()` table function that returns
//! actual row data from Parquet files with additional CDC metadata columns —
//! inserts, deletes (with the deleted rows' old values), and UPDATEs correlated
//! into `update_preimage`/`update_postimage` pairs, matching official DuckLake.
//!
//! Note: Ordering across files is undefined unless explicitly requested via ORDER BY.

use std::collections::{HashMap, HashSet};
use std::fmt;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};

use arrow::array::{Array, ArrayRef, BooleanArray, Int64Array, StringArray, UInt32Array};
use arrow::compute::take;
use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use async_trait::async_trait;
use datafusion::catalog::Session;
use datafusion::common::Result as DataFusionResult;
use datafusion::datasource::listing::PartitionedFile;
use datafusion::datasource::physical_plan::{FileGroup, FileScanConfigBuilder, ParquetSource};
use datafusion::datasource::source::DataSourceExec;
use datafusion::datasource::{TableProvider, TableType};
use datafusion::error::DataFusionError;
use datafusion::execution::object_store::ObjectStoreUrl;
use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext};
use datafusion::physical_expr::expressions::Column;
use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
use datafusion::physical_plan::projection::ProjectionExec;
use datafusion::physical_plan::union::UnionExec;
use datafusion::physical_plan::{
    DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties,
};
use futures::{Stream, StreamExt};

use crate::column_rename::ColumnRenameExec;
use crate::metadata_provider::{DataFileChange, DuckLakeTableColumn, MetadataProvider};
use crate::path_resolver::resolve_path;
use crate::positional_source::PositionalFileSource;
use crate::row_id::{FileRowNumberExec, ROW_POS_COLUMN_NAME, SNAPSHOT_ID_PARQUET_FIELD_ID};
use crate::table::{
    ParquetFileLayout, delete_file_schema, read_parquet_file_layout, read_parquet_footer_facts,
    validated_file_size, validated_record_count,
};
use crate::types::ABSENT_FIELD_PREFIX;

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

/// Type of change captured in CDC output.
///
/// [`UpdatePreimage`](ChangeType::UpdatePreimage) /
/// [`UpdatePostimage`](ChangeType::UpdatePostimage) are the paired old/new row
/// versions of an `UPDATE`: `ducklake_table_changes` correlates a same-snapshot
/// delete + insert that share a rowid into this pair (the DuckLake spirit of an
/// update in a change feed) instead of surfacing them as an unrelated delete and
/// insert. The `as_str` values match the DuckLake change-feed spec strings.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChangeType {
    Insert,
    Delete,
    /// The old version of a row that an `UPDATE` rewrote in this snapshot.
    UpdatePreimage,
    /// The new version of a row that an `UPDATE` rewrote in this snapshot.
    UpdatePostimage,
}

impl ChangeType {
    /// Returns the string representation for Arrow output
    fn as_str(&self) -> &'static str {
        match self {
            ChangeType::Insert => "insert",
            ChangeType::Delete => "delete",
            ChangeType::UpdatePreimage => "update_preimage",
            ChangeType::UpdatePostimage => "update_postimage",
        }
    }
}

impl fmt::Display for ChangeType {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

/// Present a per-file CDC scan under the catalog schema: `table_fields` resolved
/// by field id and coerced to their catalog types, then whatever internal columns
/// the scan appended (embedded rowid, embedded per-row snapshot, physical
/// position) passed through unchanged, by name.
///
/// The wrap is decided against a target that does NOT depend on the file, because
/// a union's branches must all end up with the SAME output schema: `UnionExec`
/// validates the field count only, adopts the first branch's names, and panics on
/// a type mismatch. Wrapping "only the files that need renaming" would emit the
/// pre-rename column name for the others.
///
/// Two latent hazards, both shared with the ordinary table scan, which presents
/// its positional reads the same way:
///
/// - A NOT NULL top-level column that an older file does not carry is read as an
///   all-NULL array and then presented under the non-nullable catalog field, which
///   `record_batch_with_schema` rejects. DuckDB's own extension cannot create that
///   state (`ADD COLUMN … NOT NULL` is refused as "Adding columns with constraints
///   not yet supported"), but this crate's writer and third-party implementations
///   can.
/// - The internal column names are not reserved: nothing stops a table column from
///   being called [`ROW_POS_COLUMN_NAME`], and nothing stops an older file from
///   physically carrying a column called `__ducklake_absent_field_<id>`. Either
///   makes two fields of one schema share a name, and the by-name lookup then
///   binds the first — reading the table column where the position was meant, or
///   real data where a null-fill was meant. Reserving the names belongs with the
///   writer's name validation, not here.
pub(crate) fn present_catalog_schema(
    scan: Arc<dyn ExecutionPlan>,
    table_fields: &[FieldRef],
    name_mapping: &HashMap<String, String>,
) -> Arc<dyn ExecutionPlan> {
    let scan_schema = scan.schema();
    debug_assert!(scan_schema.fields().len() >= table_fields.len());
    let mut fields: Vec<FieldRef> = table_fields.to_vec();
    fields.extend(
        scan_schema
            .fields()
            .iter()
            .skip(table_fields.len())
            .cloned(),
    );
    let output_schema: SchemaRef = Arc::new(Schema::new(fields));
    if !name_mapping.is_empty() || scan_schema != output_schema {
        Arc::new(ColumnRenameExec::new(
            scan,
            output_schema,
            name_mapping.clone(),
        ))
    } else {
        scan
    }
}

/// Refuse a column list that disagrees with the table schema built from it.
///
/// Every CDC read path locates a scan batch's internal columns (embedded rowid,
/// embedded per-row snapshot, physical position) by arithmetic on `table_len`,
/// while the per-file read schema is built from the column list. If the list is
/// LONGER, those indices address data columns and the feed reports a data value
/// as a rowid — wrong answers, no error. Both come from one
/// `get_table_structure` call inside the table functions, so this is
/// unreachable there; the providers' constructors and `with_columns` are
/// public, and this is the boundary check for a caller that pairs a schema with
/// the wrong columns.
pub(crate) fn check_column_count(table_len: usize, columns_len: usize) -> DataFusionResult<()> {
    if table_len != columns_len {
        return Err(DataFusionError::External(
            format!(
                "change feed built with {columns_len} column(s) but a {table_len}-field table \
                 schema; the schema and the column list must describe the same table"
            )
            .into(),
        ));
    }
    Ok(())
}

/// Index of a read-schema column the file actually carries. A CDC-columns-only
/// projection (`SELECT snapshot_id FROM …`, `COUNT(*)`) reads one data column
/// purely for its row count, and a column the file predates is materialised as a
/// null array rather than read from it.
fn row_count_probe_index(read_schema: &Schema) -> usize {
    read_schema
        .fields()
        .iter()
        .position(|f| !f.name().starts_with(ABSENT_FIELD_PREFIX))
        .unwrap_or(0)
}

/// Positions of the CDC metadata columns in the feed's output schema. They
/// LEAD the table columns, matching official DuckLake's `ducklake_table_changes`
/// projection (`SELECT snapshot_id, rowid, change_type, ...`).
const SNAPSHOT_ID_IDX: usize = 0;
const ROWID_IDX: usize = 1;
const CHANGE_TYPE_IDX: usize = 2;
/// Number of CDC metadata columns preceding the table columns.
const CDC_COLS: usize = 3;

/// Custom execution plan that prepends CDC columns (snapshot_id, rowid, change_type) to each batch
///
/// This plan wraps a ParquetExec and appends CDC metadata columns to each output batch.
/// It supports projection pushdown by:
/// - Reading only requested table columns from Parquet
/// - Including only requested CDC columns in output
/// - Optionally skipping input columns entirely when only CDC columns are needed
#[derive(Debug)]
pub struct PrependCDCColumnsExec {
    /// The input execution plan (typically ParquetExec)
    input: Arc<dyn ExecutionPlan>,
    /// Snapshot ID for this file
    snapshot_id: i64,
    /// Change type for this file
    change_type: ChangeType,
    /// Whether to include a rowid column in output. On this insert-only path
    /// rowid cannot be synthesized, so it is emitted as an all-NULL column
    /// (used only for encrypted tables, where the correlated path can't run).
    include_rowid: bool,
    /// Whether to include snapshot_id in output
    include_snapshot_id: bool,
    /// Whether to include change_type in output
    include_change_type: bool,
    /// If true, input columns are dummy (for row count only) and should not be included
    skip_input_columns: bool,
    /// Output schema (projected input schema + requested CDC columns)
    output_schema: SchemaRef,
    /// Cached plan properties with updated schema
    properties: Arc<PlanProperties>,
}

impl PrependCDCColumnsExec {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        input: Arc<dyn ExecutionPlan>,
        snapshot_id: i64,
        change_type: ChangeType,
        include_rowid: bool,
        include_snapshot_id: bool,
        include_change_type: bool,
        skip_input_columns: bool,
        output_schema: SchemaRef,
    ) -> Self {
        // Create new equivalence properties with the output schema.
        // We preserve partitioning and execution semantics from input.
        // Note: This resets equivalences which is pessimistic but correct.
        // Future optimization: carry forward equivalences for projected table columns.
        let eq_properties = EquivalenceProperties::new(output_schema.clone());

        let properties = Arc::new(PlanProperties::new(
            eq_properties,
            input.output_partitioning().clone(),
            input.pipeline_behavior(),
            input.boundedness(),
        ));

        Self {
            input,
            snapshot_id,
            change_type,
            include_rowid,
            include_snapshot_id,
            include_change_type,
            skip_input_columns,
            output_schema,
            properties,
        }
    }
}

impl DisplayAs for PrependCDCColumnsExec {
    fn fmt_as(&self, t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match t {
            DisplayFormatType::Default
            | DisplayFormatType::Verbose
            | DisplayFormatType::TreeRender => {
                write!(
                    f,
                    "PrependCDCColumnsExec: snapshot_id={}, change_type={}, \
                     include_snapshot={}, include_change={}, skip_input={}",
                    self.snapshot_id,
                    self.change_type,
                    self.include_snapshot_id,
                    self.include_change_type,
                    self.skip_input_columns
                )
            },
        }
    }
}

impl ExecutionPlan for PrependCDCColumnsExec {
    fn name(&self) -> &str {
        "PrependCDCColumnsExec"
    }

    fn properties(&self) -> &Arc<PlanProperties> {
        &self.properties
    }

    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
        vec![&self.input]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        if children.len() != 1 {
            return Err(DataFusionError::Internal(
                "PrependCDCColumnsExec expects exactly one child".into(),
            ));
        }

        Ok(Arc::new(PrependCDCColumnsExec::new(
            children[0].clone(),
            self.snapshot_id,
            self.change_type,
            self.include_rowid,
            self.include_snapshot_id,
            self.include_change_type,
            self.skip_input_columns,
            self.output_schema.clone(),
        )))
    }

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

    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> DataFusionResult<SendableRecordBatchStream> {
        let input_stream = self.input.execute(partition, context)?;

        Ok(Box::pin(PrependCDCColumnsStream {
            input: input_stream,
            snapshot_id: self.snapshot_id,
            change_type: self.change_type,
            include_rowid: self.include_rowid,
            include_snapshot_id: self.include_snapshot_id,
            include_change_type: self.include_change_type,
            skip_input_columns: self.skip_input_columns,
            output_schema: self.output_schema.clone(),
        }))
    }
}

/// Stream that appends CDC columns to input batches
struct PrependCDCColumnsStream {
    input: SendableRecordBatchStream,
    snapshot_id: i64,
    change_type: ChangeType,
    include_rowid: bool,
    include_snapshot_id: bool,
    include_change_type: bool,
    skip_input_columns: bool,
    output_schema: SchemaRef,
}

impl Stream for PrependCDCColumnsStream {
    type Item = DataFusionResult<RecordBatch>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match Pin::new(&mut self.input).poll_next(cx) {
            Poll::Ready(Some(Ok(batch))) => {
                let result = self.transform_batch(&batch);
                Poll::Ready(Some(result))
            },
            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

impl PrependCDCColumnsStream {
    fn transform_batch(&self, batch: &RecordBatch) -> DataFusionResult<RecordBatch> {
        let num_rows = batch.num_rows();
        let mut columns: Vec<ArrayRef> = Vec::new();

        // Prepend requested CDC columns, in the order snapshot_id, rowid,
        // change_type (official DuckLake order). rowid is all-NULL here: this
        // insert-only path can't synthesize it (used for encrypted tables).
        if self.include_snapshot_id {
            columns.push(Arc::new(Int64Array::from(vec![self.snapshot_id; num_rows])));
        }
        if self.include_rowid {
            columns.push(Arc::new(Int64Array::from(vec![None::<i64>; num_rows])));
        }
        if self.include_change_type {
            columns.push(Arc::new(StringArray::from(vec![
                self.change_type.as_str();
                num_rows
            ])));
        }

        // Then the input columns, unless we're skipping them
        if !self.skip_input_columns {
            columns.extend(batch.columns().iter().cloned());
        }

        // A zero-column projection (e.g. `COUNT(*)`) still needs the row count.
        RecordBatch::try_new_with_options(
            self.output_schema.clone(),
            columns,
            &arrow::record_batch::RecordBatchOptions::new().with_row_count(Some(num_rows)),
        )
        .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
    }
}

impl RecordBatchStream for PrependCDCColumnsStream {
    fn schema(&self) -> SchemaRef {
        self.output_schema.clone()
    }
}

/// Projection analysis result: maps logical projection to physical components
struct ProjectionInfo {
    /// Table column indices to read from Parquet (in original order)
    table_indices: Vec<usize>,
    /// Whether rowid is requested
    need_rowid: bool,
    /// Whether snapshot_id is requested
    need_snapshot_id: bool,
    /// Whether change_type is requested
    need_change_type: bool,
    /// The projected output schema
    output_schema: SchemaRef,
}

#[derive(Debug)]
pub struct TableChangesTable {
    provider: Arc<dyn MetadataProvider>,
    table_id: i64,
    start_snapshot: i64,
    end_snapshot: i64,
    /// Object store URL for resolving file paths
    object_store_url: Arc<ObjectStoreUrl>,
    /// Table path for resolving relative file paths
    table_path: String,
    /// Original table schema (without CDC columns)
    table_schema: SchemaRef,
    /// Combined schema: table columns + snapshot_id + change_type
    output_schema: SchemaRef,
    /// The table's columns as of `end_snapshot`, carrying the field ids each data
    /// file's columns are resolved by. Set via [`Self::with_columns`]; fetched
    /// from the metadata provider on demand when it is not.
    columns: Option<Arc<Vec<DuckLakeTableColumn>>>,
    /// Per-file read layout, memoized by resolved path. A file inserted and then
    /// deleted inside the window is reached from both sides of the feed, and a
    /// delete's source data file can be reached by several delete records.
    layout_cache: Mutex<HashMap<String, Arc<ParquetFileLayout>>>,
    /// When set, the delete side is never read: every row added in the window
    /// surfaces as `insert` (the `ducklake_table_insertions` feed).
    insertions_only: bool,
}

impl TableChangesTable {
    pub fn new(
        provider: Arc<dyn MetadataProvider>,
        table_id: i64,
        start_snapshot: i64,
        end_snapshot: i64,
        object_store_url: Arc<ObjectStoreUrl>,
        table_path: String,
        table_schema: SchemaRef,
    ) -> Self {
        // Build output schema: CDC metadata columns leading — (snapshot_id,
        // rowid, change_type), official DuckLake's ducklake_table_changes
        // column order — then the table columns.
        let mut fields: Vec<Field> = Vec::with_capacity(table_schema.fields().len() + CDC_COLS);
        fields.push(Field::new("snapshot_id", DataType::Int64, false));
        // rowid is nullable: it is NULL on encrypted (PME) tables, where the
        // correlated feed cannot decrypt footers to resolve rowids.
        fields.push(Field::new("rowid", DataType::Int64, true));
        fields.push(Field::new("change_type", DataType::Utf8, false));
        fields.extend(table_schema.fields().iter().map(|f| f.as_ref().clone()));
        let output_schema = Arc::new(Schema::new(fields));

        Self {
            provider,
            table_id,
            start_snapshot,
            end_snapshot,
            object_store_url,
            table_path,
            table_schema,
            output_schema,
            columns: None,
            layout_cache: Mutex::new(HashMap::new()),
            insertions_only: false,
        }
    }

    /// Supply the table's columns as of the window's end snapshot.
    ///
    /// A data file records each column under the name it had when the file was
    /// written, tagged with the column's field id, so the feed resolves columns by
    /// field id rather than by name. Without this the columns are fetched from the
    /// metadata provider on each scan; passing them in avoids that query when the
    /// caller already holds them.
    pub fn with_columns(mut self, columns: Vec<DuckLakeTableColumn>) -> Self {
        self.columns = Some(Arc::new(columns));
        self
    }

    /// Turn this feed into `ducklake_table_insertions`: the delete side is
    /// never read, so every row added in the window — plain inserts, UPDATE
    /// postimages, in-window rows of merged partial files — surfaces as
    /// `insert`, matching official DuckLake's insertions feed.
    pub fn insertions_only(mut self) -> Self {
        self.insertions_only = true;
        self
    }

    /// The table's columns as of the window's end snapshot: whatever
    /// [`Self::with_columns`] was given, else a fresh metadata read.
    fn resolve_columns(&self) -> DataFusionResult<Arc<Vec<DuckLakeTableColumn>>> {
        match &self.columns {
            Some(columns) => Ok(Arc::clone(columns)),
            None => {
                let columns = self
                    .provider
                    .get_table_structure(self.table_id, self.end_snapshot)
                    .map_err(|e| DataFusionError::External(Box::new(e)))?;
                Ok(Arc::new(columns))
            },
        }
    }

    /// The read layout of one data file, memoized by resolved path.
    async fn file_layout(
        &self,
        state: &dyn Session,
        columns: &[DuckLakeTableColumn],
        path: &str,
        is_relative: bool,
    ) -> DataFusionResult<Arc<ParquetFileLayout>> {
        let resolved = resolve_path(&self.table_path, path, is_relative)
            .map_err(|e| DataFusionError::External(Box::new(e)))?;
        {
            let cache = self.layout_cache.lock().unwrap();
            if let Some(layout) = cache.get(&resolved) {
                return Ok(Arc::clone(layout));
            }
        }
        // No encryption key: this path never reaches an encrypted file (see the
        // guard in `scan`), and a footer it cannot decrypt is not readable here.
        let layout = read_parquet_file_layout(
            state,
            self.object_store_url.as_ref(),
            &resolved,
            None,
            columns,
            &self.table_schema,
        )
        .await?;
        self.layout_cache
            .lock()
            .unwrap()
            .entry(resolved)
            .or_insert_with(|| Arc::clone(&layout));
        Ok(layout)
    }

    /// Analyze projection and split into table columns and CDC columns.
    /// CDC columns lead the table columns in the order `snapshot_id`, `rowid`,
    /// `change_type` (official DuckLake order).
    fn analyze_projection(&self, projection: Option<&Vec<usize>>) -> ProjectionInfo {
        let num_table_cols = self.table_schema.fields().len();

        match projection {
            None => {
                // No projection - read all columns
                ProjectionInfo {
                    table_indices: (0..num_table_cols).collect(),
                    need_rowid: true,
                    need_snapshot_id: true,
                    need_change_type: true,
                    output_schema: self.output_schema.clone(),
                }
            },
            Some(indices) => {
                // Split indices into table columns and CDC columns
                let mut table_indices: Vec<usize> = Vec::new();
                let mut need_rowid = false;
                let mut need_snapshot_id = false;
                let mut need_change_type = false;

                for &idx in indices {
                    match idx {
                        SNAPSHOT_ID_IDX => need_snapshot_id = true,
                        ROWID_IDX => need_rowid = true,
                        CHANGE_TYPE_IDX => need_change_type = true,
                        _ if idx < num_table_cols + CDC_COLS => {
                            table_indices.push(idx - CDC_COLS);
                        },
                        _ => {},
                    }
                }

                // Build projected output schema in the order requested
                let mut fields: Vec<Field> = Vec::with_capacity(indices.len());
                for &idx in indices {
                    fields.push(self.output_schema.field(idx).clone());
                }
                let output_schema = Arc::new(Schema::new(fields));

                ProjectionInfo {
                    table_indices,
                    need_rowid,
                    need_snapshot_id,
                    need_change_type,
                    output_schema,
                }
            },
        }
    }

    /// Build the schema that PrependCDCColumnsExec will output. On this
    /// (encryption-aware, insert-only) path rowid cannot be synthesized, so when
    /// requested it is emitted as a nullable, all-NULL column.
    fn build_cdc_exec_schema(
        &self,
        table_indices: &[usize],
        need_rowid: bool,
        need_snapshot_id: bool,
        need_change_type: bool,
    ) -> SchemaRef {
        let mut fields: Vec<Field> = Vec::with_capacity(table_indices.len() + CDC_COLS);

        if need_snapshot_id {
            fields.push(Field::new("snapshot_id", DataType::Int64, false));
        }
        if need_rowid {
            fields.push(Field::new("rowid", DataType::Int64, true));
        }
        if need_change_type {
            fields.push(Field::new("change_type", DataType::Utf8, false));
        }
        fields.extend(
            table_indices
                .iter()
                .map(|&i| self.table_schema.field(i).clone()),
        );

        Arc::new(Schema::new(fields))
    }

    /// Build a ParquetExec wrapped with PrependCDCColumnsExec for a single file.
    /// `layout` is `None` only for an encrypted file, whose footer this path
    /// cannot decrypt: its columns are then matched by name against the catalog
    /// schema (see the guard in [`TableProvider::scan`]).
    #[cfg(feature = "encryption")]
    async fn build_exec_for_file(
        &self,
        state: &dyn Session,
        data_file: &DataFileChange,
        layout: Option<&ParquetFileLayout>,
        proj_info: &ProjectionInfo,
        encryption_factory: &Option<Arc<dyn EncryptionFactory>>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let read_schema = self.file_read_schema(layout);
        let parquet_source = if let Some(factory) = encryption_factory {
            ParquetSource::new(read_schema).with_encryption_factory(Arc::clone(factory))
        } else {
            ParquetSource::new(read_schema)
        };
        self.build_exec_for_file_impl(state, data_file, layout, proj_info, parquet_source)
            .await
    }

    /// Build a ParquetExec wrapped with PrependCDCColumnsExec for a single file
    #[cfg(not(feature = "encryption"))]
    async fn build_exec_for_file(
        &self,
        state: &dyn Session,
        data_file: &DataFileChange,
        layout: Option<&ParquetFileLayout>,
        proj_info: &ProjectionInfo,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let parquet_source = ParquetSource::new(self.file_read_schema(layout));
        self.build_exec_for_file_impl(state, data_file, layout, proj_info, parquet_source)
            .await
    }

    /// The schema to scan one data file's table columns with: the file's own
    /// field-id-resolved read schema, or the catalog schema when there is no
    /// layout (an encrypted file).
    fn file_read_schema(&self, layout: Option<&ParquetFileLayout>) -> SchemaRef {
        match layout {
            Some(layout) => Arc::clone(&layout.read_schema),
            None => Arc::clone(&self.table_schema),
        }
    }

    /// Internal implementation for building a ParquetExec wrapped with PrependCDCColumnsExec
    async fn build_exec_for_file_impl(
        &self,
        _state: &dyn Session,
        data_file: &DataFileChange,
        layout: Option<&ParquetFileLayout>,
        proj_info: &ProjectionInfo,
        parquet_source: ParquetSource,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        // Resolve file path
        let resolved_path = resolve_path(
            &self.table_path,
            &data_file.path,
            data_file.path_is_relative,
        )
        .map_err(|e| DataFusionError::External(Box::new(e)))?;

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

        // Determine what to read from Parquet
        let parquet_projection = if proj_info.table_indices.is_empty() {
            // Only CDC columns requested - read minimal data for row counts
            Some(vec![row_count_probe_index(&self.file_read_schema(layout))])
        } else {
            Some(proj_info.table_indices.clone())
        };

        // Create file scan config with projection pushdown
        let mut builder = FileScanConfigBuilder::new(
            self.object_store_url.as_ref().clone(),
            Arc::new(parquet_source),
        )
        .with_file_group(FileGroup::new(vec![pf]));

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

        let file_scan_config = builder.build();

        // Use DataSourceExec directly to preserve our ParquetSource with encryption factory
        let mut parquet_exec: Arc<dyn ExecutionPlan> =
            DataSourceExec::from_data_source(file_scan_config);

        // Determine if we should skip input columns (only CDC columns requested)
        let skip_input_columns = proj_info.table_indices.is_empty();

        // Present the projected catalog fields, so every file's branch of the
        // union below carries the same schema and `PrependCDCColumnsExec` can
        // build batches under it. Nothing to present when the table columns are
        // dropped anyway (a CDC-columns-only projection).
        if !skip_input_columns {
            let table_fields: Vec<FieldRef> = proj_info
                .table_indices
                .iter()
                .map(|&i| Arc::clone(&self.table_schema.fields()[i]))
                .collect();
            let name_mapping = layout.map(|l| l.name_mapping.clone()).unwrap_or_default();
            parquet_exec = present_catalog_schema(parquet_exec, &table_fields, &name_mapping);
        }

        // Build output schema for PrependCDCColumnsExec
        let cdc_exec_schema = if skip_input_columns {
            // Only CDC columns - build schema with just those
            let mut fields = Vec::new();
            if proj_info.need_snapshot_id {
                fields.push(Field::new("snapshot_id", DataType::Int64, false));
            }
            if proj_info.need_rowid {
                fields.push(Field::new("rowid", DataType::Int64, true));
            }
            if proj_info.need_change_type {
                fields.push(Field::new("change_type", DataType::Utf8, false));
            }
            Arc::new(Schema::new(fields))
        } else {
            self.build_cdc_exec_schema(
                &proj_info.table_indices,
                proj_info.need_rowid,
                proj_info.need_snapshot_id,
                proj_info.need_change_type,
            )
        };

        Ok(Arc::new(PrependCDCColumnsExec::new(
            parquet_exec,
            data_file.begin_snapshot,
            ChangeType::Insert,
            proj_info.need_rowid,
            proj_info.need_snapshot_id,
            proj_info.need_change_type,
            skip_input_columns,
            cdc_exec_schema,
        )))
    }

    /// Read a DELETE file's footer and return the physical name of its embedded
    /// per-row snapshot column ([`SNAPSHOT_ID_PARQUET_FIELD_ID`]) when present.
    /// Current-spec delete files are cumulative and carry one.
    ///
    /// A delete file holds `(file_path, pos[, snapshot])`, none of it table
    /// columns, so it needs the raw footer rather than a read layout.
    async fn detect_delete_file_snapshot_name(
        &self,
        state: &dyn Session,
        path: &str,
        is_relative: bool,
    ) -> DataFusionResult<Option<String>> {
        let resolved = resolve_path(&self.table_path, path, is_relative)
            .map_err(|e| DataFusionError::External(Box::new(e)))?;
        let facts =
            read_parquet_footer_facts(state, self.object_store_url.as_ref(), &resolved, None)
                .await?;
        Ok(facts.field_ids.get(&SNAPSHOT_ID_PARQUET_FIELD_ID).cloned())
    }

    /// The read schema for a data file: its table columns under the physical names
    /// THIS file gives them, plus the embedded rowid / per-row snapshot columns
    /// when present, in that order.
    fn read_schema_with_embedded(
        &self,
        layout: &ParquetFileLayout,
        rowid_name: &Option<String>,
        snapshot_name: &Option<String>,
    ) -> SchemaRef {
        match (rowid_name, snapshot_name) {
            (None, None) => Arc::clone(&layout.read_schema),
            (rowid, snapshot) => {
                let mut fields: Vec<FieldRef> =
                    layout.read_schema.fields().iter().cloned().collect();
                if let Some(name) = rowid {
                    fields.push(Arc::new(Field::new(name, DataType::Int64, true)));
                }
                if let Some(name) = snapshot {
                    fields.push(Arc::new(Field::new(name, DataType::Int64, true)));
                }
                Arc::new(Schema::new(fields))
            },
        }
    }

    /// The catalog fields the per-file scans of the correlated feed present, so
    /// every one of them hands `correlate_changes` the same table columns.
    fn catalog_table_fields(&self) -> Vec<FieldRef> {
        self.table_schema.fields().iter().cloned().collect()
    }

    /// Scan of an inserted data file for the correlated feed. A file with an
    /// embedded rowid (an UPDATE / compaction postimage) is scanned plainly — its
    /// rowid IS the embedded column. A plain insert is scanned positionally
    /// (`PositionalFileSource` + `FileRowNumberExec`) so its rowid can be
    /// synthesized as `row_id_start + position` — but only when `need_rowid`;
    /// otherwise it is a plain scan with no position column.
    ///
    /// Either way the plan is presented under the catalog schema, ABOVE
    /// `FileRowNumberExec` so the position column it appends passes straight
    /// through.
    fn build_insert_scan(
        &self,
        data_file: &DataFileChange,
        layout: &ParquetFileLayout,
        need_rowid_resolution: bool,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let resolved = resolve_path(
            &self.table_path,
            &data_file.path,
            data_file.path_is_relative,
        )
        .map_err(|e| DataFusionError::External(Box::new(e)))?;
        let mut pf = PartitionedFile::new(
            &resolved,
            validated_file_size(data_file.file_size_bytes, &resolved)?,
        );
        if let Some(footer) = data_file.footer_size
            && footer > 0
            && let Ok(hint) = usize::try_from(footer)
        {
            pf = pf.with_metadata_size_hint(hint);
        }
        // The per-row snapshot column is read only for partial (merged) files.
        let snapshot_name = if data_file.partial_max.is_some() {
            layout.embedded_snapshot_parquet_name.clone()
        } else {
            None
        };
        let embedded_rowid = &layout.embedded_rowid_parquet_name;
        let read_schema = self.read_schema_with_embedded(layout, embedded_rowid, &snapshot_name);
        let plain_scan = |pf: PartitionedFile, schema: SchemaRef| {
            let builder = FileScanConfigBuilder::new(
                self.object_store_url.as_ref().clone(),
                Arc::new(ParquetSource::new(schema)),
            )
            .with_file_group(FileGroup::new(vec![pf]));
            DataSourceExec::from_data_source(builder.build())
        };
        let scan: Arc<dyn ExecutionPlan> = match embedded_rowid {
            // Postimage / rewrite / non-adjacent merge: rowid IS the embedded
            // column — a plain scan.
            Some(_) => plain_scan(pf, read_schema),
            // No embedded rowid but resolution required (rowid projected, or a
            // partial file whose real rowids feed the update correlation): scan
            // positionally to synthesize rowid = row_id_start + position.
            None if need_rowid_resolution => {
                let source = PositionalFileSource::wrap(Arc::new(ParquetSource::new(read_schema)));
                let builder =
                    FileScanConfigBuilder::new(self.object_store_url.as_ref().clone(), source)
                        .with_file_group(FileGroup::new(vec![pf]))
                        .with_partitioned_by_file_group(true);
                let scan = DataSourceExec::from_data_source(builder.build());
                Arc::new(FileRowNumberExec::new(scan, vec![0]))
            },
            // Plain insert, rowid not needed: a plain scan, no positions.
            None => plain_scan(pf, read_schema),
        };
        Ok(present_catalog_schema(
            scan,
            &self.catalog_table_fields(),
            &layout.name_mapping,
        ))
    }

    /// Positional scan of a delete's source data file: table columns, the
    /// embedded rowid column when present, and the internal physical-position
    /// column. `PositionalFileSource` + `FileRowNumberExec` guarantee true
    /// physical positions so deleted rows can be matched to the delete file's
    /// `pos` set regardless of scan partitioning.
    fn build_delete_data_scan(
        &self,
        resolved_path: &str,
        size_bytes: i64,
        footer_size: i64,
        layout: &ParquetFileLayout,
        embedded_name: &Option<String>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let mut pf = PartitionedFile::new(
            resolved_path,
            validated_file_size(size_bytes, resolved_path)?,
        );
        if footer_size > 0
            && let Ok(hint) = usize::try_from(footer_size)
        {
            pf = pf.with_metadata_size_hint(hint);
        }
        let read_schema = self.read_schema_with_embedded(layout, embedded_name, &None);
        let source = PositionalFileSource::wrap(Arc::new(ParquetSource::new(read_schema)));
        let builder = FileScanConfigBuilder::new(self.object_store_url.as_ref().clone(), source)
            .with_file_group(FileGroup::new(vec![pf]))
            .with_partitioned_by_file_group(true);
        let scan = DataSourceExec::from_data_source(builder.build());
        Ok(present_catalog_schema(
            Arc::new(FileRowNumberExec::new(scan, vec![0])),
            &self.catalog_table_fields(),
            &layout.name_mapping,
        ))
    }

    /// Scan of a positional delete file (the standard `(file_path, pos)` schema);
    /// the correlation path reads its `pos` column to find newly-deleted rows.
    fn build_delete_file_scan(
        &self,
        path: &str,
        is_relative: bool,
        size_bytes: i64,
        footer_size: i64,
        snapshot_name: &Option<String>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let resolved = resolve_path(&self.table_path, path, is_relative)
            .map_err(|e| DataFusionError::External(Box::new(e)))?;
        let mut pf = PartitionedFile::new(&resolved, validated_file_size(size_bytes, &resolved)?);
        if footer_size > 0
            && let Ok(hint) = usize::try_from(footer_size)
        {
            pf = pf.with_metadata_size_hint(hint);
        }
        // A cumulative (current-spec) delete file embeds each row's delete
        // snapshot as a third column; read it when present.
        let schema = match snapshot_name {
            Some(name) => {
                let mut fields: Vec<Field> = delete_file_schema()
                    .fields()
                    .iter()
                    .map(|f| f.as_ref().clone())
                    .collect();
                fields.push(Field::new(name, DataType::Int64, true));
                Arc::new(Schema::new(fields))
            },
            None => delete_file_schema(),
        };
        let builder = FileScanConfigBuilder::new(
            self.object_store_url.as_ref().clone(),
            Arc::new(ParquetSource::new(schema)),
        )
        .with_file_group(FileGroup::new(vec![pf]));
        Ok(DataSourceExec::from_data_source(builder.build()))
    }

    /// Refuse the feed when an encrypted table's columns changed identity between
    /// the oldest data file in the window and the window's end.
    ///
    /// Encrypted files' footers cannot be decrypted on this path, so their columns
    /// can only be matched by name — correct only while every in-window file
    /// records the columns under the names the end snapshot uses. A rename or a
    /// drop-and-re-add breaks that, and a by-name read would then return NULL or
    /// another column's values with no indication anything was wrong. Adding a
    /// column, and dropping one outright, stay safe: the new name is in no older
    /// file (so it null-fills) and the dropped one is asked for by nobody.
    ///
    /// The identity below is coarser than "was a column renamed", so this
    /// refuses strictly more than it has to: a type widened by `ALTER … TYPE`
    /// and a struct child added inside an existing column both change it, and
    /// both would in fact read correctly by name. Erring toward refusal is the
    /// point — the alternative on this path is silently wrong values — but a
    /// finer check (compare names per field id, ignore the rest) would let those
    /// two through if they turn out to matter. `COMPATIBILITY.md` says so.
    #[cfg(feature = "encryption")]
    fn reject_evolved_encrypted_table(
        &self,
        columns: &[DuckLakeTableColumn],
        data_files: &[DataFileChange],
    ) -> DataFusionResult<()> {
        let Some(oldest) = data_files.iter().map(|f| f.begin_snapshot).min() else {
            return Ok(());
        };
        if oldest >= self.end_snapshot {
            return Ok(());
        }
        let then = self
            .provider
            .get_table_structure(self.table_id, oldest)
            .map_err(|e| DataFusionError::External(Box::new(e)))?;

        // A column's identity is its name plus its resolved type — the type is
        // what carries a NESTED field's name, so a renamed struct child shows up
        // here too.
        let identity = |column: &DuckLakeTableColumn| -> DataFusionResult<(String, String)> {
            let data_type = column
                .data_type()
                .map_err(|e| DataFusionError::External(Box::new(e)))?;
            Ok((column.column_name.clone(), format!("{data_type:?}")))
        };
        let mut then_by_id: HashMap<i64, (String, String)> = HashMap::new();
        let mut then_names: HashSet<String> = HashSet::new();
        for column in &then {
            then_by_id.insert(column.column_id, identity(column)?);
            then_names.insert(column.column_name.clone());
        }

        let mut evolved = false;
        for column in columns {
            evolved = match then_by_id.get(&column.column_id) {
                Some(previous) => *previous != identity(column)?,
                // A column id the older snapshot did not have: safe unless its
                // name did exist back then under a different id — that is a drop
                // and re-add, and a by-name read would find the dropped column's
                // data in the older files.
                None => then_names.contains(&column.column_name),
            };
            if evolved {
                break;
            }
        }

        if evolved {
            return Err(DataFusionError::External(
                format!(
                    "table {} has encrypted data files and its columns changed between \
                     snapshot {oldest} and snapshot {}: a column was renamed, or dropped and \
                     re-added under the same name. Change feeds resolve columns by field id, \
                     which requires reading each file's parquet footer, and an encrypted \
                     footer cannot be read here — so the feed would return another column's \
                     values. Query a snapshot window whose files all predate the change, or \
                     read the table directly.",
                    self.table_id, self.end_snapshot
                )
                .into(),
            ));
        }
        Ok(())
    }

    /// Build the correlated change feed: pair a same-snapshot delete + insert
    /// that share a rowid into `update_preimage` (old) + `update_postimage`
    /// (new); surface unmatched inserts as `insert` and unmatched deletes as
    /// `delete` (carrying the deleted rows' old values), matching official
    /// DuckLake's `ducklake_table_changes`.
    async fn build_correlated_changes(
        &self,
        state: &dyn Session,
        columns: &[DuckLakeTableColumn],
        data_files: &[DataFileChange],
        delete_files: &[crate::metadata_provider::DeleteFileChange],
        layouts: &[Arc<ParquetFileLayout>],
        projection: Option<&Vec<usize>>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        let table_len = self.table_schema.fields().len();
        // Both come from the same column list, and the index arithmetic below
        // (which column of a scan batch is the embedded rowid, the embedded
        // snapshot, the position) assumes they agree. A disagreement would
        // silently misalign every per-file scan — with more columns than
        // `table_len` the internal-column indices land on DATA columns and the
        // feed emits a data value as a rowid. Unreachable through the table
        // functions, which derive both from one `get_table_structure` call, but
        // the constructor and `with_columns` are public: refuse loudly.
        check_column_count(table_len, columns.len())?;
        // Whether the caller wants the rowid column (at ROWID_IDX among the
        // leading CDC columns). When it does not, plain inserts skip the
        // positional scan and rowid synthesis entirely.
        let need_rowid = projection.is_none_or(|idx| idx.contains(&ROWID_IDX));

        let mut insert_units = Vec::with_capacity(data_files.len());
        for (df, layout) in data_files.iter().zip(layouts.iter()) {
            // Column layout of the scan batch after the `table_len` table
            // columns: [embedded rowid?][embedded snapshot? (partial files
            // only)][position? (appended by FileRowNumberExec)].
            let is_partial = df.partial_max.is_some();
            if is_partial && layout.embedded_snapshot_parquet_name.is_none() {
                return Err(DataFusionError::External(
                    format!(
                        "data file {} is a merged partial file (partial_max set) but carries \
                         no embedded per-row snapshot column; cannot attribute its rows to \
                         snapshots",
                        df.path
                    )
                    .into(),
                ));
            }
            // A partial file's rows carry REAL rowids into the update
            // correlation (a placeholder could false-pair), so rowid is
            // resolved for them even when it is not projected.
            let resolve_rowid = need_rowid || is_partial;
            let has_embedded_rowid = layout.embedded_rowid_parquet_name.is_some();
            let mut next_idx = table_len;
            let embedded_col_idx = has_embedded_rowid.then(|| {
                let i = next_idx;
                next_idx += 1;
                i
            });
            let snapshot_col_idx = (is_partial && layout.embedded_snapshot_parquet_name.is_some())
                .then(|| {
                    let i = next_idx;
                    next_idx += 1;
                    i
                });
            let pos_col_idx = (!has_embedded_rowid && resolve_rowid).then_some(next_idx);
            insert_units.push(InsertUnit {
                snapshot_id: df.begin_snapshot,
                scan: self.build_insert_scan(df, layout, resolve_rowid)?,
                embedded_col_idx,
                snapshot_col_idx,
                pos_col_idx,
                row_id_start: df.row_id_start,
            });
        }

        // Every delete in range is read: unmatched ones surface as `delete`
        // rows, and those sharing a (snapshot_id, rowid) with an embedded-rowid
        // insert pair into update preimages.
        let delete_units = {
            let mut delete_units = Vec::with_capacity(delete_files.len());
            for dfc in delete_files {
                validated_record_count(dfc.data_record_count, &dfc.data_file_path)?;
                let resolved = resolve_path(
                    &self.table_path,
                    &dfc.data_file_path,
                    dfc.data_file_path_is_relative,
                )
                .map_err(|e| DataFusionError::External(Box::new(e)))?;
                // The source data file can predate the window (and any rename in
                // it), so its columns need resolving by field id too.
                let source_layout = self
                    .file_layout(
                        state,
                        columns,
                        &dfc.data_file_path,
                        dfc.data_file_path_is_relative,
                    )
                    .await?;
                let old_embedded = source_layout.embedded_rowid_parquet_name.clone();
                let data_scan = self.build_delete_data_scan(
                    &resolved,
                    dfc.data_file_size_bytes,
                    dfc.data_file_footer_size.unwrap_or(0),
                    &source_layout,
                    &old_embedded,
                )?;
                // A cumulative (current-spec) delete file embeds each row's
                // delete snapshot; its positions are windowed per row and no
                // previous-file subtraction applies. Legacy 2-column files keep
                // the delta-vs-previous model.
                let snapshot_name = match &dfc.current_delete_path {
                    Some(p) => {
                        self.detect_delete_file_snapshot_name(
                            state,
                            p,
                            dfc.current_delete_path_is_relative.unwrap_or(true),
                        )
                        .await?
                    },
                    None => None,
                };
                if snapshot_name.is_none() && dfc.snapshot_id < self.start_snapshot {
                    return Err(DataFusionError::External(
                        format!(
                            "delete file {:?} begins before the query window but carries no \
                             embedded per-row snapshot column; its deletions cannot be attributed",
                            dfc.current_delete_path
                        )
                        .into(),
                    ));
                }
                let cumulative = snapshot_name.is_some();
                let current_delete_scan = match &dfc.current_delete_path {
                    Some(p) => Some(self.build_delete_file_scan(
                        p,
                        dfc.current_delete_path_is_relative.unwrap_or(true),
                        dfc.current_delete_file_size_bytes.unwrap_or(0),
                        dfc.current_delete_footer_size.unwrap_or(0),
                        &snapshot_name,
                    )?),
                    None => None,
                };
                let previous_delete_scan = match &dfc.previous_delete_path {
                    Some(p) if !cumulative => Some(self.build_delete_file_scan(
                        p,
                        dfc.previous_delete_path_is_relative.unwrap_or(true),
                        dfc.previous_delete_file_size_bytes.unwrap_or(0),
                        dfc.previous_delete_footer_size.unwrap_or(0),
                        &None,
                    )?),
                    _ => None,
                };
                delete_units.push(DeleteUnit {
                    snapshot_id: dfc.snapshot_id,
                    data_scan,
                    embedded_col_idx: old_embedded.as_ref().map(|_| table_len),
                    current_delete_scan,
                    previous_delete_scan,
                    cumulative,
                    record_count: dfc.data_record_count,
                    row_id_start: dfc.data_row_id_start,
                });
            }
            delete_units
        };

        let full: Arc<dyn ExecutionPlan> = Arc::new(TableChangesExec::new(
            insert_units,
            delete_units,
            self.table_schema.clone(),
            self.output_schema.clone(),
            table_len,
            need_rowid,
            (self.start_snapshot, self.end_snapshot),
        ));

        // The exec emits the full `[snapshot_id, rowid, change_type, table
        // columns]` schema; honor the requested projection with a
        // ProjectionExec on top.
        match projection {
            None => Ok(full),
            Some(indices) => {
                let exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = indices
                    .iter()
                    .map(|&i| {
                        let f = self.output_schema.field(i);
                        (
                            Arc::new(Column::new(f.name(), i)) as Arc<dyn PhysicalExpr>,
                            f.name().to_string(),
                        )
                    })
                    .collect();
                Ok(Arc::new(ProjectionExec::try_new(exprs, full)?))
            },
        }
    }
}

#[async_trait]
impl TableProvider for TableChangesTable {
    fn schema(&self) -> SchemaRef {
        self.output_schema.clone()
    }

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

    async fn scan(
        &self,
        state: &dyn Session,
        projection: Option<&Vec<usize>>,
        _filters: &[datafusion::prelude::Expr],
        _limit: Option<usize>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        // Analyze projection to determine what to read
        let proj_info = self.analyze_projection(projection);

        // The columns as of the window's END snapshot, carrying the field ids the
        // per-file read schemas are built from — official DuckLake resolves a
        // change feed against exactly that generation of the schema.
        let columns = self.resolve_columns()?;

        // Get data files added between snapshots (INSERT changes)
        let data_files = self
            .provider
            .get_data_files_added_between_snapshots(
                self.table_id,
                self.start_snapshot,
                self.end_snapshot,
            )
            .map_err(|e| DataFusionError::External(Box::new(e)))?;

        // Deletes applied in the window surface as `delete` rows (and pair into
        // update preimages), so they participate in both the empty check and
        // the path decision — a delete-only window is NOT empty. The insertions
        // feed never reads the delete side.
        let delete_files = if self.insertions_only {
            Vec::new()
        } else {
            self.provider
                .get_delete_files_added_between_snapshots(
                    self.table_id,
                    self.start_snapshot,
                    self.end_snapshot,
                )
                .map_err(|e| DataFusionError::External(Box::new(e)))?
        };

        // Handle empty case
        if data_files.is_empty() && delete_files.is_empty() {
            use datafusion::physical_plan::empty::EmptyExec;
            return Ok(Arc::new(EmptyExec::new(proj_info.output_schema)));
        }

        let mut data_files = data_files;

        // Decide whether to take the correlated path (reading delete sources to
        // emit `delete` rows and pair an UPDATE's delete+insert into
        // preimage/postimage). Guards, all cheap and metadata-only:
        //
        //  1. Deletes-present OR rowid-requested: with neither, the window is
        //     plain inserts and needs no correlation — a plain-INSERT catalog
        //     does ZERO per-file parquet footer reads at plan time.
        //  2. Not encrypted: the correlated path reads parquet footers (to detect
        //     the embedded-rowid postimage) and the source rows of deletes, none
        //     of which it can decrypt (the delete-side change record carries no
        //     key). On a PME catalog we therefore stay on the insert-only path
        //     below — which IS encryption-aware — so CDC over inserts never
        //     fails; the tradeoff is that UPDATEs surface as plain inserts and
        //     pure deletes are missing there. See COMPATIBILITY.md. (A
        //     delete-only window carries no data file to detect encryption
        //     from, so on an encrypted catalog it fails at read rather than
        //     returning wrong results.)
        let any_encrypted = {
            #[cfg(feature = "encryption")]
            {
                data_files.iter().any(|d| d.encryption_key.is_some())
            }
            #[cfg(not(feature = "encryption"))]
            {
                false
            }
        };

        // Merged partial files whose window overlap comes only from
        // `partial_max` (begin_snapshot before the window) need per-row
        // attribution, which the encrypted insert-only path below cannot do
        // (it cannot read the embedded snapshot column). Drop them there —
        // matching the pre-partial_max behavior — rather than emitting rows
        // with wrong snapshots. See COMPATIBILITY.md. When that leaves no
        // data files at all, the feed is empty (deletes are already
        // documented-missing on encrypted catalogs), not a planning error.
        if any_encrypted {
            data_files.retain(|f| f.begin_snapshot >= self.start_snapshot);
            if data_files.is_empty() {
                use datafusion::physical_plan::empty::EmptyExec;
                return Ok(Arc::new(EmptyExec::new(proj_info.output_schema)));
            }
            // Without the footers there are no field ids to resolve columns by,
            // and matching by name is wrong once a column has been renamed or
            // dropped and re-added.
            #[cfg(feature = "encryption")]
            self.reject_evolved_encrypted_table(&columns, &data_files)?;
        }
        let any_partial = data_files.iter().any(|f| f.partial_max.is_some());

        if (proj_info.need_rowid || !delete_files.is_empty() || any_partial) && !any_encrypted {
            let mut layouts: Vec<Arc<ParquetFileLayout>> = Vec::with_capacity(data_files.len());
            for data_file in &data_files {
                layouts.push(
                    self.file_layout(state, &columns, &data_file.path, data_file.path_is_relative)
                        .await?,
                );
            }
            return self
                .build_correlated_changes(
                    state,
                    &columns,
                    &data_files,
                    &delete_files,
                    &layouts,
                    projection,
                )
                .await;
        }

        // Build encryption factory from file encryption keys (when encryption feature is enabled)
        #[cfg(feature = "encryption")]
        let encryption_factory: Option<Arc<dyn EncryptionFactory>> = {
            let mut builder = EncryptionFactoryBuilder::new();
            for data_file in &data_files {
                let resolved_path = resolve_path(
                    &self.table_path,
                    &data_file.path,
                    data_file.path_is_relative,
                )
                .map_err(|e| DataFusionError::External(Box::new(e)))?;
                builder.add_file(&resolved_path, data_file.encryption_key.as_deref());
            }
            let factory = builder.build();
            if factory.has_encrypted_files() {
                Some(Arc::new(factory) as Arc<dyn EncryptionFactory>)
            } else {
                None
            }
        };

        // Build execution plan for each file with projection pushdown. Each
        // file's columns are resolved by field id — except on an encrypted
        // catalog, whose footers this path cannot read (the query has already
        // been refused above if that table's columns evolved).
        let mut execs: Vec<Arc<dyn ExecutionPlan>> = Vec::with_capacity(data_files.len());
        for data_file in &data_files {
            let layout = if any_encrypted {
                None
            } else {
                Some(
                    self.file_layout(state, &columns, &data_file.path, data_file.path_is_relative)
                        .await?,
                )
            };
            #[cfg(feature = "encryption")]
            let exec = self
                .build_exec_for_file(
                    state,
                    data_file,
                    layout.as_deref(),
                    &proj_info,
                    &encryption_factory,
                )
                .await?;
            #[cfg(not(feature = "encryption"))]
            let exec = self
                .build_exec_for_file(state, data_file, layout.as_deref(), &proj_info)
                .await?;
            execs.push(exec);
        }

        // Combine with UnionExec if multiple files
        if execs.len() == 1 {
            Ok(execs.into_iter().next().unwrap())
        } else {
            UnionExec::try_new(execs)
        }
    }
}

// ---------------------------------------------------------------------------
// Correlated change feed (insert / delete / update_preimage / update_postimage)
// ---------------------------------------------------------------------------

/// One inserted data file added in the snapshot range, with enough context to
/// derive each row's rowid. When `embedded_col_idx` is `Some`, the scan's column
/// at that index is the file's embedded rowid (an UPDATE / compaction postimage);
/// otherwise the file is a plain INSERT whose scan carries a physical-position
/// column at `pos_col_idx` and whose rowid is `row_id_start + position`.
#[derive(Clone)]
struct InsertUnit {
    /// The file's `begin_snapshot`: every row's snapshot for an ordinary file;
    /// for a merged partial file (`snapshot_col_idx` set) it is only the
    /// minimum, and each row's actual snapshot comes from the embedded column.
    snapshot_id: i64,
    scan: Arc<dyn ExecutionPlan>,
    /// Column index of the embedded rowid, or `None` for a plain insert.
    embedded_col_idx: Option<usize>,
    /// Column index of the embedded per-row snapshot (merged partial files
    /// only). Rows outside the query window are filtered out on read.
    snapshot_col_idx: Option<usize>,
    /// Column index of the physical row-position (plain inserts only).
    pos_col_idx: Option<usize>,
    /// First rowid of the file (`None` if the catalog carries none). Required
    /// only for a plain insert, whose rowid is `row_id_start + pos`.
    row_id_start: Option<i64>,
}

/// One delete applied in the snapshot range: enough to read the newly-deleted
/// rows of the source data file (the delete positions minus the previous
/// generation's) together with each row's rowid.
#[derive(Clone)]
struct DeleteUnit {
    snapshot_id: i64,
    /// Positional scan of the source data file: `[table columns..., (embedded
    /// rowid), __ducklake_row_pos]`.
    data_scan: Arc<dyn ExecutionPlan>,
    /// Column index of the source file's embedded rowid, or `None` (rowids are
    /// then `row_id_start + position`).
    embedded_col_idx: Option<usize>,
    /// Scan of the current delete file, or `None` for a full-file delete.
    current_delete_scan: Option<Arc<dyn ExecutionPlan>>,
    /// Scan of the delete file this one superseded, if any.
    previous_delete_scan: Option<Arc<dyn ExecutionPlan>>,
    /// Whether the current delete file is cumulative (carries an embedded
    /// per-row delete-snapshot column as its third column): positions are then
    /// windowed per row and each deleted row keys/emits at its own snapshot.
    cumulative: bool,
    record_count: i64,
    /// First rowid of the source file (`None` if the catalog carries none).
    /// Required only when a deleted row has no embedded rowid.
    row_id_start: Option<i64>,
}

/// Rows carrying their `(snapshot_id, rowid)` correlation key alongside the
/// table columns, ready to be tagged once update pairs are known.
struct KeyedRows {
    snapshot_id: i64,
    table_batch: RecordBatch,
    rowid: Int64Array,
}

/// Execution plan for the correlated `ducklake_table_changes` feed. Collects the
/// inserted rows (with embedded rowids) and the newly-deleted rows (with
/// synthesized/embedded rowids), pairs those sharing a `(snapshot_id, rowid)`
/// into preimage/postimage, and emits the tagged rows. Single output partition.
#[derive(Debug)]
pub struct TableChangesExec {
    insert_units: Vec<InsertUnit>,
    delete_units: Vec<DeleteUnit>,
    #[allow(dead_code)]
    table_schema: SchemaRef,
    output_schema: SchemaRef,
    table_len: usize,
    /// Whether the rowid column is actually requested. When false, plain inserts
    /// skip rowid synthesis (emitting a placeholder dropped by the projection),
    /// so a non-rowid projection never needs a plain insert's row_id_start.
    need_rowid: bool,
    /// The query's `[start, end]` snapshot window (inclusive), used to filter
    /// rows of merged partial files by their embedded per-row snapshot.
    window: (i64, i64),
    properties: Arc<PlanProperties>,
}

impl std::fmt::Debug for InsertUnit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("InsertUnit")
            .field("snapshot_id", &self.snapshot_id)
            .field("embedded_col_idx", &self.embedded_col_idx)
            .finish_non_exhaustive()
    }
}

impl std::fmt::Debug for DeleteUnit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DeleteUnit")
            .field("snapshot_id", &self.snapshot_id)
            .field("embedded_col_idx", &self.embedded_col_idx)
            .finish_non_exhaustive()
    }
}

impl TableChangesExec {
    #[allow(clippy::too_many_arguments)]
    fn new(
        insert_units: Vec<InsertUnit>,
        delete_units: Vec<DeleteUnit>,
        table_schema: SchemaRef,
        output_schema: SchemaRef,
        table_len: usize,
        need_rowid: bool,
        window: (i64, i64),
    ) -> Self {
        let properties = Arc::new(PlanProperties::new(
            EquivalenceProperties::new(output_schema.clone()),
            datafusion::physical_expr::Partitioning::UnknownPartitioning(1),
            datafusion::physical_plan::execution_plan::EmissionType::Final,
            datafusion::physical_plan::execution_plan::Boundedness::Bounded,
        ));
        Self {
            insert_units,
            delete_units,
            table_schema,
            output_schema,
            table_len,
            need_rowid,
            window,
            properties,
        }
    }
}

impl DisplayAs for TableChangesExec {
    fn fmt_as(&self, t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match t {
            DisplayFormatType::Default
            | DisplayFormatType::Verbose
            | DisplayFormatType::TreeRender => {
                write!(
                    f,
                    "TableChangesExec: inserts={}, deletes={}",
                    self.insert_units.len(),
                    self.delete_units.len()
                )
            },
        }
    }
}

impl ExecutionPlan for TableChangesExec {
    fn name(&self) -> &str {
        "TableChangesExec"
    }

    fn properties(&self) -> &Arc<PlanProperties> {
        &self.properties
    }

    /// No DataFusion children: the per-file scans are internal and executed
    /// directly, so the optimizer never rewrites them.
    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
        vec![]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        if !children.is_empty() {
            return Err(DataFusionError::Internal(
                "TableChangesExec has no children".to_string(),
            ));
        }
        Ok(self)
    }

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

    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> DataFusionResult<SendableRecordBatchStream> {
        if partition != 0 {
            return Err(DataFusionError::Internal(format!(
                "TableChangesExec only supports partition 0, got {partition}"
            )));
        }

        let insert_units = self.insert_units.clone();
        let delete_units = self.delete_units.clone();
        let output_schema = self.output_schema.clone();
        let table_len = self.table_len;
        let need_rowid = self.need_rowid;
        let window = self.window;

        let fut = async move {
            correlate_changes(
                insert_units,
                delete_units,
                output_schema,
                table_len,
                need_rowid,
                window,
                context,
            )
            .await
        };

        let schema = self.output_schema.clone();
        let stream = futures::stream::once(fut)
            .map(|res: DataFusionResult<Vec<RecordBatch>>| match res {
                Ok(batches) => futures::stream::iter(batches.into_iter().map(Ok)).boxed(),
                Err(e) => futures::stream::iter(std::iter::once(Err(e))).boxed(),
            })
            .flatten();

        Ok(Box::pin(
            datafusion::physical_plan::stream::RecordBatchStreamAdapter::new(schema, stream),
        ))
    }
}

/// Collect the inserted and deleted rows, correlate update pairs by
/// `(snapshot_id, rowid)`, and produce the tagged output batches. `window` is
/// the query's inclusive `[start, end]` snapshot range, applied per row to
/// merged partial files (whose rows span several snapshots).
#[allow(clippy::too_many_arguments)]
async fn correlate_changes(
    insert_units: Vec<InsertUnit>,
    delete_units: Vec<DeleteUnit>,
    output_schema: SchemaRef,
    table_len: usize,
    need_rowid: bool,
    window: (i64, i64),
    context: Arc<TaskContext>,
) -> DataFusionResult<Vec<RecordBatch>> {
    // Inserted rows split into postimage candidates (embedded rowid — can pair
    // with a delete into an UPDATE) and plain inserts (fresh rowid — never pair).
    // A plain insert's rowid is synthesized only when actually requested; when it
    // is projected away it is a placeholder, so a non-rowid query never needs a
    // plain insert's row_id_start.
    let mut postimages: Vec<KeyedRows> = Vec::new();
    let mut plain_inserts: Vec<KeyedRows> = Vec::new();
    for unit in &insert_units {
        let batches =
            datafusion::physical_plan::collect(Arc::clone(&unit.scan), context.clone()).await?;
        for b in batches {
            let n = b.num_rows();
            if n == 0 {
                continue;
            }
            let table_batch = b.project(&(0..table_len).collect::<Vec<_>>())?;

            // Resolve each row's rowid: the embedded column when present, else
            // row_id_start + physical position (only when actually needed).
            let embedded_rowid = match unit.embedded_col_idx {
                Some(idx) => Some(int64_column(&b, idx, "embedded rowid")?.clone()),
                None => None,
            };
            let rowid: Int64Array = match (&embedded_rowid, unit.pos_col_idx) {
                (Some(arr), _) => (*arr).clone(),
                (None, Some(pos_idx)) => {
                    let row_id_start = unit.row_id_start.ok_or_else(|| {
                        DataFusionError::Internal(
                            "cannot synthesize rowid: inserted file has neither an embedded \
                             rowid nor a row_id_start"
                                .to_string(),
                        )
                    })?;
                    let pos = int64_column(&b, pos_idx, ROW_POS_COLUMN_NAME)?;
                    Int64Array::from(
                        (0..n)
                            .map(|i| row_id_start + pos.value(i))
                            .collect::<Vec<i64>>(),
                    )
                },
                // Rowid neither embedded nor resolved (not requested): a
                // placeholder dropped by the projection; never pairs.
                (None, None) => Int64Array::from(vec![0i64; n]),
            };

            match unit.snapshot_col_idx {
                // Merged partial file: each row belongs to its embedded origin
                // snapshot. Filter rows to the query window and split the batch
                // into one KeyedRows group per snapshot; their rowids are real,
                // so they participate in update pairing like any insertion.
                Some(snap_idx) => {
                    let snaps = int64_column(&b, snap_idx, "embedded snapshot_id")?;
                    let mut by_snapshot: std::collections::BTreeMap<i64, Vec<u32>> =
                        std::collections::BTreeMap::new();
                    for i in 0..n {
                        if snaps.is_null(i) {
                            return Err(DataFusionError::Internal(
                                "embedded snapshot_id column contains NULL".to_string(),
                            ));
                        }
                        let s = snaps.value(i);
                        if s >= window.0 && s <= window.1 {
                            by_snapshot.entry(s).or_default().push(i as u32);
                        }
                    }
                    for (snapshot, row_indices) in by_snapshot {
                        let indices = UInt32Array::from(row_indices);
                        let cols: Vec<ArrayRef> = table_batch
                            .columns()
                            .iter()
                            .map(|c| {
                                take(c.as_ref(), &indices, None)
                                    .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
                            })
                            .collect::<DataFusionResult<_>>()?;
                        let group_batch = RecordBatch::try_new(table_batch.schema(), cols)?;
                        let group_rowid = take(&rowid, &indices, None)
                            .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?
                            .as_any()
                            .downcast_ref::<Int64Array>()
                            .expect("take preserves Int64")
                            .clone();
                        postimages.push(KeyedRows {
                            snapshot_id: snapshot,
                            table_batch: group_batch,
                            rowid: group_rowid,
                        });
                    }
                },
                // Ordinary file: all rows belong to begin_snapshot. Embedded
                // rowids (UPDATE / rewrite outputs) can pair with deletes;
                // plain inserts carry fresh (or placeholder) rowids and never
                // pair.
                None => {
                    let keyed = KeyedRows {
                        snapshot_id: unit.snapshot_id,
                        table_batch,
                        rowid,
                    };
                    if embedded_rowid.is_some() {
                        postimages.push(keyed);
                    } else {
                        plain_inserts.push(keyed);
                    }
                },
            }
        }
    }

    // Deleted rows: the positions newly masked at this snapshot, with each row's
    // rowid (embedded column when the source file has one, else row_id_start +
    // physical position). The rowid is required only when it is output
    // (`need_rowid`) or when an update pair is possible (some postimage exists
    // to correlate against); with neither, every delete is a pure delete and
    // its rowid a placeholder — so a non-rowid projection over a delete-only
    // window never fails on a source file whose rowid cannot be synthesized
    // (no embedded rowid and a NULL row_id_start), mirroring
    // `ducklake_table_deletions`.
    let preimage_rowids_required = need_rowid || !postimages.is_empty();
    let mut preimages: Vec<KeyedRows> = Vec::new();
    for unit in &delete_units {
        // Cumulative delete files carry a per-row delete snapshot: only
        // in-window positions are collected, each remembering its snapshot;
        // no previous-file subtraction applies (the scan is None then).
        let (current, position_snapshots): (Option<HashSet<i64>>, HashMap<i64, i64>) =
            if unit.cumulative {
                let (set, map) = collect_windowed_delete_positions(
                    &unit.current_delete_scan,
                    window,
                    context.clone(),
                )
                .await?;
                (Some(set), map)
            } else {
                (
                    collect_delete_positions(&unit.current_delete_scan, context.clone()).await?,
                    HashMap::new(),
                )
            };
        let current: HashSet<i64> = match current {
            Some(set) => set,
            None => (0..unit.record_count).collect(),
        };
        let previous = collect_delete_positions(&unit.previous_delete_scan, context.clone())
            .await?
            .unwrap_or_default();

        let data_batches =
            datafusion::physical_plan::collect(Arc::clone(&unit.data_scan), context.clone())
                .await?;
        for b in data_batches {
            let n = b.num_rows();
            if n == 0 {
                continue;
            }
            let pos_idx = b.schema().index_of(ROW_POS_COLUMN_NAME)?;
            let pos = b
                .column(pos_idx)
                .as_any()
                .downcast_ref::<Int64Array>()
                .ok_or_else(|| {
                    DataFusionError::Internal(format!("{ROW_POS_COLUMN_NAME} column is not Int64"))
                })?;
            let embedded = match unit.embedded_col_idx {
                Some(idx) => Some(
                    b.column(idx)
                        .as_any()
                        .downcast_ref::<Int64Array>()
                        .ok_or_else(|| {
                            DataFusionError::Internal(
                                "embedded rowid column is not Int64".to_string(),
                            )
                        })?,
                ),
                None => None,
            };

            // With no embedded rowid, deleted rowids are row_id_start + position;
            // require row_id_start in that case rather than emitting wrong ids —
            // but only when the rowid is actually consumed (output or pairing).
            let synth_start: Option<i64> = if embedded.is_none() && preimage_rowids_required {
                Some(unit.row_id_start.ok_or_else(|| {
                    DataFusionError::Internal(
                        "cannot synthesize deleted rowid: source file has neither an embedded \
                         rowid nor a row_id_start"
                            .to_string(),
                    )
                })?)
            } else {
                None
            };

            // Group kept rows by their delete snapshot: constant for legacy
            // delete files, per-row for cumulative ones. Each group becomes
            // one KeyedRows so the (snapshot, rowid) pairing sees the right
            // snapshot for every deleted row.
            let mut by_snapshot: std::collections::BTreeMap<i64, (Vec<u32>, Vec<i64>)> =
                std::collections::BTreeMap::new();
            for i in 0..n {
                let p = pos.value(i);
                if current.contains(&p) && !previous.contains(&p) {
                    let rowid = match (embedded, synth_start) {
                        (Some(arr), _) => arr.value(i),
                        (None, Some(start)) => start + p,
                        // Unneeded rowid (not output, nothing to pair with):
                        // a placeholder that update_keys can never contain.
                        (None, None) => 0,
                    };
                    let snapshot = if unit.cumulative {
                        *position_snapshots.get(&p).unwrap_or(&unit.snapshot_id)
                    } else {
                        unit.snapshot_id
                    };
                    let entry = by_snapshot.entry(snapshot).or_default();
                    entry.0.push(i as u32);
                    entry.1.push(rowid);
                }
            }
            for (snapshot, (keep, rowids)) in by_snapshot {
                let indices = UInt32Array::from(keep);
                let table_cols: Vec<ArrayRef> = (0..table_len)
                    .map(|c| {
                        take(b.column(c).as_ref(), &indices, None)
                            .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
                    })
                    .collect::<DataFusionResult<_>>()?;
                let table_batch = RecordBatch::try_new(
                    Arc::new(Schema::new(
                        (0..table_len)
                            .map(|c| b.schema().field(c).clone())
                            .collect::<Vec<_>>(),
                    )),
                    table_cols,
                )?;
                preimages.push(KeyedRows {
                    snapshot_id: snapshot,
                    table_batch,
                    rowid: Int64Array::from(rowids),
                });
            }
        }
    }

    // Update pairs = a postimage (embedded, preserved rowid) whose (snapshot,
    // rowid) also appears as a delete. Only postimages seed the key set — plain
    // inserts carry fresh rowids that never match a delete (and, when rowid is
    // projected away, only a placeholder), so they must not participate.
    let post_keys: HashSet<(i64, i64)> = postimages
        .iter()
        .flat_map(|k| (0..k.rowid.len()).map(move |i| (k.snapshot_id, k.rowid.value(i))))
        .collect();
    let update_keys: HashSet<(i64, i64)> = preimages
        .iter()
        .flat_map(|k| (0..k.rowid.len()).map(move |i| (k.snapshot_id, k.rowid.value(i))))
        .filter(|key| post_keys.contains(key))
        .collect();

    let mut out: Vec<RecordBatch> = Vec::new();
    // Plain inserts are always `insert` (they never pair with a delete).
    for k in &plain_inserts {
        out.push(prepend_cdc_columns(
            &k.table_batch,
            k.rowid.clone(),
            k.snapshot_id,
            ChangeType::Insert,
            &output_schema,
        )?);
    }
    for k in &postimages {
        // Rows whose key is an update pair become postimages; the rest are plain
        // inserts (embedded file with no matching delete, e.g. compaction).
        if let Some(b) = filter_and_tag(
            k,
            &key_mask(k, &update_keys, true),
            ChangeType::UpdatePostimage,
            &output_schema,
        )? {
            out.push(b);
        }
        if let Some(b) = filter_and_tag(
            k,
            &key_mask(k, &update_keys, false),
            ChangeType::Insert,
            &output_schema,
        )? {
            out.push(b);
        }
    }
    for k in &preimages {
        // Rows paired with an insert surface as update preimages; the rest are
        // pure deletes, emitted as `delete` rows carrying the old values
        // (matching official DuckLake's table_changes).
        if let Some(b) = filter_and_tag(
            k,
            &key_mask(k, &update_keys, true),
            ChangeType::UpdatePreimage,
            &output_schema,
        )? {
            out.push(b);
        }
        if let Some(b) = filter_and_tag(
            k,
            &key_mask(k, &update_keys, false),
            ChangeType::Delete,
            &output_schema,
        )? {
            out.push(b);
        }
    }
    Ok(out)
}

/// Downcast a batch column to `Int64Array` with a descriptive error.
fn int64_column<'a>(
    batch: &'a RecordBatch,
    idx: usize,
    what: &str,
) -> DataFusionResult<&'a Int64Array> {
    batch
        .column(idx)
        .as_any()
        .downcast_ref::<Int64Array>()
        .ok_or_else(|| DataFusionError::Internal(format!("{what} column is not Int64")))
}

/// Collect the in-window `pos` set AND per-position delete snapshots from a
/// cumulative delete-file scan (`(file_path, pos, snapshot)` schema). Rows
/// whose snapshot falls outside the inclusive `window` are skipped.
async fn collect_windowed_delete_positions(
    scan: &Option<Arc<dyn ExecutionPlan>>,
    window: (i64, i64),
    context: Arc<TaskContext>,
) -> DataFusionResult<(HashSet<i64>, HashMap<i64, i64>)> {
    let Some(scan) = scan else {
        return Ok((HashSet::new(), HashMap::new()));
    };
    let batches = datafusion::physical_plan::collect(Arc::clone(scan), context).await?;
    let mut set = HashSet::new();
    let mut map = HashMap::new();
    for b in &batches {
        if b.num_columns() < 3 {
            return Err(DataFusionError::Internal(
                "cumulative delete file batch is missing its snapshot column".to_string(),
            ));
        }
        let pos = int64_column(b, 1, "delete `pos`")?;
        let snaps = int64_column(b, 2, "delete snapshot")?;
        for i in 0..pos.len() {
            if pos.is_null(i) {
                continue;
            }
            if snaps.is_null(i) {
                return Err(DataFusionError::Internal(
                    "cumulative delete file has a NULL per-row snapshot".to_string(),
                ));
            }
            let s = snaps.value(i);
            if s >= window.0 && s <= window.1 {
                let p = pos.value(i);
                set.insert(p);
                map.insert(p, s);
            }
        }
    }
    Ok((set, map))
}

/// Collect the `pos` set from a delete-file scan (`None` scan => `None`).
async fn collect_delete_positions(
    scan: &Option<Arc<dyn ExecutionPlan>>,
    context: Arc<TaskContext>,
) -> DataFusionResult<Option<HashSet<i64>>> {
    let Some(scan) = scan else {
        return Ok(None);
    };
    let batches = datafusion::physical_plan::collect(Arc::clone(scan), context).await?;
    let mut set = HashSet::new();
    for b in &batches {
        if b.num_columns() < 2 {
            continue;
        }
        let pos = b
            .column(1)
            .as_any()
            .downcast_ref::<Int64Array>()
            .ok_or_else(|| {
                DataFusionError::Internal("delete `pos` column is not Int64".to_string())
            })?;
        for i in 0..pos.len() {
            if !pos.is_null(i) {
                set.insert(pos.value(i));
            }
        }
    }
    Ok(Some(set))
}

/// Prepend the CDC `snapshot_id` + `rowid` + `change_type` columns to a
/// table-column batch (official DuckLake column order). `rowid` must have the
/// same length as `table_batch`.
fn prepend_cdc_columns(
    table_batch: &RecordBatch,
    rowid: Int64Array,
    snapshot_id: i64,
    change: ChangeType,
    output_schema: &SchemaRef,
) -> DataFusionResult<RecordBatch> {
    let n = table_batch.num_rows();
    let mut cols: Vec<ArrayRef> = Vec::with_capacity(table_batch.num_columns() + CDC_COLS);
    cols.push(Arc::new(Int64Array::from(vec![snapshot_id; n])));
    cols.push(Arc::new(rowid));
    cols.push(Arc::new(StringArray::from(vec![change.as_str(); n])));
    cols.extend(table_batch.columns().iter().cloned());
    RecordBatch::try_new(output_schema.clone(), cols)
        .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
}

/// A row-selection mask over `keyed`: `want_update` selects rows whose
/// `(snapshot_id, rowid)` is (or is not) an update pair.
fn key_mask(
    keyed: &KeyedRows,
    update_keys: &HashSet<(i64, i64)>,
    want_update: bool,
) -> BooleanArray {
    BooleanArray::from(
        (0..keyed.rowid.len())
            .map(|i| {
                let is_update = update_keys.contains(&(keyed.snapshot_id, keyed.rowid.value(i)));
                is_update == want_update
            })
            .collect::<Vec<bool>>(),
    )
}

/// Filter `keyed`'s table columns by `mask`, tag with `change`, and append the
/// CDC columns. Returns `None` when the mask selects no rows.
fn filter_and_tag(
    keyed: &KeyedRows,
    mask: &BooleanArray,
    change: ChangeType,
    output_schema: &SchemaRef,
) -> DataFusionResult<Option<RecordBatch>> {
    if mask.true_count() == 0 {
        return Ok(None);
    }
    let cols: Vec<ArrayRef> = keyed
        .table_batch
        .columns()
        .iter()
        .map(|c| {
            arrow::compute::filter(c.as_ref(), mask)
                .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))
        })
        .collect::<DataFusionResult<_>>()?;
    let filtered = RecordBatch::try_new(keyed.table_batch.schema(), cols)?;
    let rowid = arrow::compute::filter(&keyed.rowid, mask)
        .map_err(|e| DataFusionError::ArrowError(Box::new(e), None))?
        .as_any()
        .downcast_ref::<Int64Array>()
        .ok_or_else(|| DataFusionError::Internal("filtered rowid is not Int64".to_string()))?
        .clone();
    Ok(Some(prepend_cdc_columns(
        &filtered,
        rowid,
        keyed.snapshot_id,
        change,
        output_schema,
    )?))
}

// ---------------------------------------------------------------------------
// ducklake_table_insertions
// ---------------------------------------------------------------------------

/// `ducklake_table_insertions`: every row added in the snapshot window —
/// plain inserts, UPDATE postimages, and in-window rows of compaction-merged
/// partial files — with `(snapshot_id, rowid)` leading and NO `change_type`
/// column, matching official DuckLake's insertions feed surface (which has
/// none; it exposes rowid/snapshot_id as virtual columns).
///
/// A thin wrapper over the [`TableChangesTable`] machinery with the delete
/// side disabled: projections are translated to skip the inner feed's
/// `change_type` column.
#[derive(Debug)]
pub struct TableInsertionsTable {
    inner: TableChangesTable,
    output_schema: SchemaRef,
}

impl TableInsertionsTable {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        provider: Arc<dyn MetadataProvider>,
        table_id: i64,
        start_snapshot: i64,
        end_snapshot: i64,
        object_store_url: Arc<ObjectStoreUrl>,
        table_path: String,
        table_schema: SchemaRef,
    ) -> Self {
        let mut fields: Vec<Field> = Vec::with_capacity(table_schema.fields().len() + 2);
        fields.push(Field::new("snapshot_id", DataType::Int64, false));
        fields.push(Field::new("rowid", DataType::Int64, true));
        fields.extend(table_schema.fields().iter().map(|f| f.as_ref().clone()));
        let output_schema = Arc::new(Schema::new(fields));
        let inner = TableChangesTable::new(
            provider,
            table_id,
            start_snapshot,
            end_snapshot,
            object_store_url,
            table_path,
            table_schema,
        )
        .insertions_only();
        Self {
            inner,
            output_schema,
        }
    }

    /// Supply the table's columns as of the window's end snapshot; see
    /// [`TableChangesTable::with_columns`].
    pub fn with_columns(mut self, columns: Vec<DuckLakeTableColumn>) -> Self {
        self.inner = self.inner.with_columns(columns);
        self
    }

    /// Map an index of this feed's schema — `(snapshot_id, rowid, table
    /// columns...)` — onto the inner changes schema, which has `change_type`
    /// at [`CHANGE_TYPE_IDX`].
    fn inner_index(outer: usize) -> usize {
        if outer < CHANGE_TYPE_IDX {
            outer
        } else {
            outer + 1
        }
    }
}

#[async_trait]
impl TableProvider for TableInsertionsTable {
    fn schema(&self) -> SchemaRef {
        self.output_schema.clone()
    }

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

    async fn scan(
        &self,
        state: &dyn Session,
        projection: Option<&Vec<usize>>,
        filters: &[datafusion::prelude::Expr],
        limit: Option<usize>,
    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
        // The inner feed honors arbitrary projections in requested order, so a
        // translated projection yields exactly this feed's columns.
        let translated: Vec<usize> = match projection {
            Some(indices) => indices.iter().map(|&i| Self::inner_index(i)).collect(),
            None => (0..self.output_schema.fields().len())
                .map(Self::inner_index)
                .collect(),
        };
        self.inner
            .scan(state, Some(&translated), filters, limit)
            .await
    }
}