delta_kernel 0.28.0

//! Fluent builder for declarative [`Plan`]s: a source, chained transforms, then
//! [`PlanBuilder::build`] / [`PlanBuilder::build_opt`]. Each transform consumes `self` and returns
//! a new builder; clone a builder (cheap) to feed it into more than one transform. Validating
//! methods return [`DeltaResult`] and surface errors at the call site.
//!
//! A source with no rows (e.g. a scan over no files) is the *absent* relation. The builder treats
//! it as dead code: transforms validate as usual but propagate absence, eliminating the parts of
//! the plan that cannot contribute rows (a union drops absent arms, an anti-join with an absent
//! build forwards its probe, etc.). [`build`](PlanBuilder::build) collapses the plan to its
//! minimal correct shape, emitting a single empty [`Values`] node when the whole relation is
//! absent, so it always yields a runnable plan; [`build_opt`](PlanBuilder::build_opt) yields
//! `None` for it.
//!
//! ```
//! use std::sync::Arc;
//! use delta_kernel::PlanBuilder;
//! use delta_kernel::expressions::col;
//! use delta_kernel::plans::ir::nodes::Operator;
//! use delta_kernel::schema::{DataType, StructField, StructType};
//!
//! let schema = Arc::new(StructType::try_new([StructField::not_null("id", DataType::INTEGER)])?);
//!
//! // A source with no rows is the *absent* relation: it produces no rows, so the builder
//! // eliminates it as dead code, collapsing the plan to its minimal correct shape. Downstream
//! // transforms still validate and chain exactly as they would over a populated source.
//! let scan = PlanBuilder::values(schema, vec![])?
//!     .filter(col!("id").is_not_null())?;
//!
//! // `build_opt` reports the eliminated plan as `None` ...
//! assert!(scan.clone().build_opt()?.is_none());
//! // ... while `build` keeps the plan runnable, replacing the absent relation with a single
//! // empty `Values` node.
//! let plan = scan.build()?;
//! let [node] = plan.nodes.as_slice() else { panic!("expected one node") };
//! assert!(matches!(&node.op, Operator::Values(values) if values.rows.is_empty()));
//! # Ok::<(), delta_kernel::Error>(())
//! ```

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

use delta_kernel_derive::internal_api;

use super::ir::nodes::{
    Aggregate, AggregateBuilder, DynamicScan, FileType, Filter, Operator, Project, ScanFile,
    ScanJson, ScanParquet, SemiJoin, UnionAll, Values,
};
use super::ir::plan::{Plan, PlanNode};
use crate::expressions::{ColumnName, ExpressionRef, PredicateRef, Scalar, StructData};
use crate::schema::{SchemaRef, ToSchema};
use crate::struct_patch::ProjectionStructPatchBuilder;
use crate::utils::CollectInto;
use crate::{DeltaResult, Error};

/// One node of a plan DAG: an operator, its inputs, and its output schema. Node identity is the
/// `Arc`'s address ([`PlanBuilder::build`] dedups shared subgraphs by pointer). Inputs are always
/// present nodes -- combinators drop or forward an absent input.
#[derive(Debug)]
struct BuilderNode {
    op: Operator,
    inputs: Vec<BuilderNodeRef>,
    schema: SchemaRef,
}

type BuilderNodeRef = Arc<BuilderNode>;

/// A present node, or the *absent* (uninhabited) relation, which still carries its output schema.
#[derive(Clone, Debug)]
enum PlanBuilderRoot {
    Present(BuilderNodeRef),
    Absent(SchemaRef),
}

/// A [`Plan`] under construction. See the module docs.
#[derive(Clone, Debug)]
#[must_use]
pub struct PlanBuilder(PlanBuilderRoot);

impl PlanBuilder {
    /// A present relation applying `op` (output `schema`) over `inputs`.
    fn present(schema: SchemaRef, op: impl Into<Operator>, inputs: Vec<BuilderNodeRef>) -> Self {
        PlanBuilder(PlanBuilderRoot::Present(Arc::new(BuilderNode {
            op: op.into(),
            inputs,
            schema,
        })))
    }

    /// The uninhabited relation over `schema`.
    fn absent(schema: SchemaRef) -> Self {
        PlanBuilder(PlanBuilderRoot::Absent(schema))
    }

    /// The output schema of this relation -- carried whether present or absent.
    fn schema(&self) -> &SchemaRef {
        match &self.0 {
            PlanBuilderRoot::Present(node) => &node.schema,
            PlanBuilderRoot::Absent(schema) => schema,
        }
    }

    /// Apply a single-input transform: wrap `op` (output `schema`) over a present input, or stay
    /// absent. Callers validate against [`Self::schema`] first.
    fn unary_op_or_absent(self, schema: SchemaRef, op: impl Into<Operator>) -> Self {
        match self.0 {
            PlanBuilderRoot::Present(node) => Self::present(schema, op, vec![node]),
            PlanBuilderRoot::Absent(_) => Self::absent(schema),
        }
    }

    /// A Parquet scan source over `files` producing rows matching `schema`.
    ///
    /// `file_constant_columns` names the per-file-constant output columns (see [`ScanParquet`]);
    /// pass `&[]` when there are none. An empty `files` yields the absent relation.
    ///
    /// Produces an error when any file's constant count differs from `file_constant_columns`'s
    /// length, or a `file_constant_columns` entry is absent from `schema`.
    pub fn scan_parquet(
        files: impl IntoIterator<Item = impl Into<ScanFile>>,
        file_constant_columns: &[&str],
        schema: impl Into<SchemaRef>,
    ) -> DeltaResult<Self> {
        Self::scan_source(FileType::Parquet, files, file_constant_columns, schema)
    }

    /// A newline-delimited JSON scan source over `files` producing rows matching `schema`. See
    /// [`ScanJson`]. Exhibits same behaviour as [`Self::scan_parquet`].
    pub fn scan_json(
        files: impl IntoIterator<Item = impl Into<ScanFile>>,
        file_constant_columns: &[&str],
        schema: impl Into<SchemaRef>,
    ) -> DeltaResult<Self> {
        Self::scan_source(FileType::Json, files, file_constant_columns, schema)
    }

    /// Shared body of [`Self::scan_parquet`] / [`Self::scan_json`]. Normalizes and validates the
    /// inputs, then yields the absent relation for an empty file set, else the scan source node for
    /// `file_type`.
    fn scan_source(
        file_type: FileType,
        files: impl IntoIterator<Item = impl Into<ScanFile>>,
        file_constant_columns: &[&str],
        schema: impl Into<SchemaRef>,
    ) -> DeltaResult<Self> {
        let schema = schema.into();
        let files = Vec::from_iter(files.into_iter().map(Into::into));
        let cols = Vec::from_iter(file_constant_columns.iter().map(|&c| c.to_string()));
        for (i, file) in files.iter().enumerate() {
            if file.file_constants.len() != cols.len() {
                return Err(Error::generic(format!(
                    "scan: file {i} has {} constant value(s) {:?} for {} file-constant column(s) {:?}",
                    file.file_constants.len(),
                    file.file_constants,
                    cols.len(),
                    cols,
                )));
            }
        }
        check_file_constant_columns(&schema, &cols, "scan file_constant_columns")?;
        if files.is_empty() {
            return Ok(Self::absent(schema));
        }
        let file_constant_columns = cols;
        let op = match file_type {
            FileType::Parquet => Operator::from(ScanParquet {
                files,
                file_constant_columns,
                schema: Arc::clone(&schema),
            }),
            FileType::Json => Operator::from(ScanJson {
                files,
                file_constant_columns,
                schema: Arc::clone(&schema),
            }),
        };
        Ok(Self::present(schema, op, vec![]))
    }

    /// Inline literal rows. See [`Values`] for the row encoding. Empty `rows` yields the absent
    /// relation.
    ///
    /// Produces an error when any row's width differs from `schema`'s top-level field count.
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let schema = Arc::new(StructType::try_new([StructField::not_null("id", DataType::INTEGER)])?);
    /// let plan = PlanBuilder::values(schema, vec![vec![1.into()], vec![2.into()]])?.build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn values(schema: impl Into<SchemaRef>, rows: Vec<Vec<Scalar>>) -> DeltaResult<Self> {
        let schema = schema.into();
        let width = schema.fields().count();
        for (i, row) in rows.iter().enumerate() {
            if row.len() != width {
                return Err(Error::generic(format!(
                    "values: row {i} has {} value(s) {row:?} but schema has {width} field(s) {:?}",
                    row.len(),
                    Vec::from_iter(schema.fields().map(|f| f.name())),
                )));
            }
        }
        Ok(Values::new(schema, rows).into())
    }

    /// Infallible sibling of [`Self::values`] containing rows of `T` converted to scalar data.
    ///
    /// Schema is [`ToSchema::to_schema`] for `T`. Each row converts via [`Into<StructData>`] and is
    /// peeled into top-level field scalars (nested fields remain [`Scalar::Struct`]); see
    /// [`Values`]'s [`FromIterator`]. Empty `rows` yields the absent relation.
    ///
    /// # Example
    /// ```
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::expressions::Scalar;
    /// # use delta_kernel::plans::ir::nodes::Operator;
    /// # use delta_kernel_derive::{IntoStructData, ToSchema};
    /// #
    /// #[derive(ToSchema, IntoStructData)]
    /// struct Row {
    ///     id: i32,
    /// }
    ///
    /// let plan = PlanBuilder::values_from([Row { id: 1 }, Row { id: 2 }]).build()?;
    /// let Operator::Values(values) = &plan.nodes[0].op else { panic!("expected Values") };
    /// assert_eq!(
    ///     values.rows,
    ///     vec![vec![Scalar::Integer(1)], vec![Scalar::Integer(2)]],
    /// );
    /// assert!(PlanBuilder::values_from(std::iter::empty::<Row>())
    ///     .build_opt()?
    ///     .is_none());
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    #[internal_api]
    pub(crate) fn values_from<T>(rows: impl IntoIterator<Item = T>) -> Self
    where
        T: Into<StructData> + ToSchema,
    {
        Values::from_iter(rows).into()
    }

    /// Keep rows where `predicate` holds. Output schema is unchanged. See [`Filter`].
    ///
    /// Produces an error when `predicate` references a column absent from the input schema.
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::expressions::col;
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let schema = Arc::new(StructType::try_new([StructField::not_null("id", DataType::INTEGER)])?);
    /// let plan = PlanBuilder::values(schema, vec![vec![1.into()], vec![2.into()]])?
    ///     .filter(col!("id").is_not_null())?
    ///     .build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn filter(self, predicate: impl Into<PredicateRef>) -> DeltaResult<Self> {
        let predicate = predicate.into();
        check_columns_resolve(self.schema(), predicate.references(), "filter")?;
        let schema = Arc::clone(self.schema());
        Ok(self.unary_op_or_absent(schema, Filter { predicate }))
    }

    /// Project `self` through `expr` into rows of the caller-declared `schema`. `expr` must be a
    /// struct constructor / patch matching `schema`. See [`Project`].
    ///
    /// Produces an error when `expr` references a column absent from the input schema. References
    /// inside a `StructPatch` are validated by [`ProjectionStructPatchBuilder`] when the patch is
    ///
    /// [`ProjectionStructPatchBuilder`]: crate::struct_patch::ProjectionStructPatchBuilder
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::expressions::{col, Expression};
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let input = Arc::new(StructType::try_new([
    ///     StructField::not_null("id", DataType::INTEGER),
    ///     StructField::nullable("name", DataType::STRING),
    /// ])?);
    /// let out = Arc::new(StructType::try_new([StructField::not_null("id", DataType::INTEGER)])?);
    /// // Keep only `id`.
    /// let plan = PlanBuilder::values(input, vec![vec![1.into(), "a".into()]])?
    ///     .project(Expression::struct_from([col!("id")]), out)?
    ///     .build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn project(
        self,
        expr: impl Into<ExpressionRef>,
        schema: impl Into<SchemaRef>,
    ) -> DeltaResult<Self> {
        let schema = schema.into();
        let expr = expr.into();
        check_columns_resolve(self.schema(), expr.references(), "project")?;
        Ok(self.unary_op_or_absent(Arc::clone(&schema), Project { expr, schema }))
    }

    /// Project `self` by editing its columns. `edit` receives a [`ProjectionStructPatchBuilder`]
    /// rooted at `self`'s schema and records field edits (replace/drop/append); the lowered patch
    /// and the resulting output schema feed the [`Project`] node together. See [`Project`].
    ///
    /// Produces an error when the patch fails to build -- e.g. a replaced or dropped field is
    /// absent from the input schema, or a nested path does not resolve.
    pub fn project_patch(
        self,
        patch: impl FnOnce(ProjectionStructPatchBuilder<'_>) -> ProjectionStructPatchBuilder<'_>,
    ) -> DeltaResult<Self> {
        let (out, expr) = patch(ProjectionStructPatchBuilder::new(self.schema())).build()?;
        self.project(expr, out)
    }

    /// Read data files named by `self`'s rows. Output schema is `dynamic_scan.schema`. See
    /// [`DynamicScan`].
    ///
    /// Applies [`DynamicScan::validate_input`]'s validation against `self`'s schema.
    pub fn dynamic_scan(self, dynamic_scan: DynamicScan) -> DeltaResult<Self> {
        dynamic_scan.validate_input(self.schema())?;
        let schema = Arc::clone(&dynamic_scan.schema);
        Ok(self.unary_op_or_absent(schema, dynamic_scan))
    }

    /// Aggregate `self` into `aggregate` (build one with [`Aggregate::group_by`]). The output
    /// schema is the group keys followed by the aggregate columns. See [`Aggregate`].
    ///
    /// Over an absent input the result is group-arity dependent: a **grouped** aggregate yields
    /// zero groups and so is absent; a **global** aggregate (no group keys) still yields one row,
    /// so it aggregates an empty [`Values`] input and lets the engine produce that row.
    ///
    /// Produces an error when building `aggregate` fails: a group key or an aggregate's operand
    /// column is absent from its input schema, or two output columns would share a name.
    pub fn aggregate(self, aggregate: impl TryInto<Aggregate, Error = Error>) -> DeltaResult<Self> {
        let aggregate = aggregate.try_into()?;
        let schema = Arc::clone(&aggregate.schema);
        match self.0 {
            PlanBuilderRoot::Present(node) => Ok(Self::present(schema, aggregate, vec![node])),
            // A grouped aggregate over the empty relation has zero groups, hence absent.
            PlanBuilderRoot::Absent(_) if !aggregate.group_by.is_empty() => {
                Ok(Self::absent(schema))
            }
            // A global aggregate always emits one row, even over no input: aggregate an empty
            // Values relation and let the engine produce that row.
            PlanBuilderRoot::Absent(input_schema) => {
                let input = Self::present(
                    Arc::clone(&input_schema),
                    Values::new(input_schema, vec![]),
                    vec![],
                );
                Ok(input.unary_op_or_absent(schema, aggregate))
            }
        }
    }

    /// Aggregate `self`, grouped by `keys`. `aggs` receives an [`AggregateBuilder`] rooted at
    /// `self`'s schema and adds the aggregate columns (see e.g. [`AggregateBuilder::max`]).
    ///
    /// Mirrors [`Self::aggregate`] but spares the caller from naming the input schema, just as
    /// [`Self::project_patch`] does for [`Self::project`]. See [`Aggregate`].
    ///
    /// Produces an error under the same conditions as [`Self::aggregate`].
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::expressions::column_name;
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let schema = Arc::new(StructType::try_new([
    ///     StructField::not_null("id", DataType::INTEGER),
    ///     StructField::nullable("version", DataType::LONG),
    /// ])?);
    /// // Latest `version` per `id`.
    /// let plan = PlanBuilder::values(schema, vec![vec![1.into(), 7i64.into()]])?
    ///     .aggregate_by([column_name!("id")], |a| a.max(column_name!("version")))?
    ///     .build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn aggregate_by(
        self,
        keys: impl CollectInto<Vec<ColumnName>>,
        aggs: impl FnOnce(AggregateBuilder) -> AggregateBuilder,
    ) -> DeltaResult<Self> {
        let builder = Aggregate::group_by(Arc::clone(self.schema()), keys);
        self.aggregate(aggs(builder))
    }

    /// Aggregate `self` without grouping. `aggs` receives an [`AggregateBuilder`] rooted at
    /// `self`'s schema and adds aggregate columns (see e.g. [`AggregateBuilder::max`]).
    ///
    /// Mirrors [`Self::aggregate`] but infers the input schema and uses
    /// [`Aggregate::ungrouped`] as the builder root.
    ///
    /// Produces an error under the same conditions as [`Self::aggregate`].
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::expressions::column_name;
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let schema = Arc::new(StructType::try_new([
    ///     StructField::not_null("id", DataType::INTEGER),
    ///     StructField::nullable("version", DataType::LONG),
    /// ])?);
    /// // Latest version across all rows.
    /// let plan = PlanBuilder::values(schema, vec![vec![1.into(), 7i64.into()]])?
    ///     .aggregate_ungrouped(|a| a.max(column_name!("version")))?
    ///     .build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn aggregate_ungrouped(
        self,
        aggs: impl FnOnce(AggregateBuilder) -> AggregateBuilder,
    ) -> DeltaResult<Self> {
        let builder = Aggregate::ungrouped(Arc::clone(self.schema()));
        self.aggregate(aggs(builder))
    }

    /// Semi join: emit the `self` (probe) rows that have a match in `build` on the join keys.
    /// Output schema mirrors `self`. Inputs are recorded as `[probe, build]`. See [`SemiJoin`].
    ///
    /// Keys are columns (e.g. [`column_name!`](crate::expressions::column_name)), matched pairwise.
    ///
    /// Produces an error when the probe and build key counts differ, or a probe/build key is absent
    /// from its input schema.
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::expressions::column_name;
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let schema = Arc::new(StructType::try_new([StructField::not_null("id", DataType::INTEGER)])?);
    /// let probe = PlanBuilder::values(Arc::clone(&schema), vec![vec![1.into()], vec![2.into()]])?;
    /// let allow = PlanBuilder::values(schema, vec![vec![2.into()]])?;
    /// // Probe rows whose `id` appears in `allow`.
    /// let plan = probe.semi_join(allow, [column_name!("id")], [column_name!("id")])?.build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn semi_join(
        self,
        build: PlanBuilder,
        probe_keys: impl IntoIterator<Item = ColumnName>,
        build_keys: impl IntoIterator<Item = ColumnName>,
    ) -> DeltaResult<Self> {
        self.semi_join_impl(build, false, probe_keys, build_keys)
    }

    /// Anti join: emit the `self` (probe) rows that have *no* match in `build` on the join keys.
    /// An inverted [`SemiJoin`]; otherwise as [`Self::semi_join`].
    pub fn anti_join(
        self,
        build: PlanBuilder,
        probe_keys: impl IntoIterator<Item = ColumnName>,
        build_keys: impl IntoIterator<Item = ColumnName>,
    ) -> DeltaResult<Self> {
        self.semi_join_impl(build, true, probe_keys, build_keys)
    }

    fn semi_join_impl(
        self,
        build: PlanBuilder,
        inverted: bool,
        probe_keys: impl IntoIterator<Item = ColumnName>,
        build_keys: impl IntoIterator<Item = ColumnName>,
    ) -> DeltaResult<Self> {
        let probe_keys = Vec::from_iter(probe_keys);
        let build_keys = Vec::from_iter(build_keys);
        if probe_keys.len() != build_keys.len() {
            return Err(Error::generic(format!(
                "join: {} probe key(s) {probe_keys:?} but {} build key(s) {build_keys:?}; \
                 they must match in length.",
                probe_keys.len(),
                build_keys.len(),
            )));
        }
        check_columns_resolve(self.schema(), &probe_keys, "join probe")?;
        check_columns_resolve(build.schema(), &build_keys, "join build")?;
        // An uninhabited probe always produces an uninhabited result; forward it unchanged.
        let PlanBuilderRoot::Present(probe) = self.0 else {
            return Ok(self);
        };
        let build = match build.0 {
            PlanBuilderRoot::Present(build) => build,
            // An absent build side matches nothing, so an anti-join forwards an unfiltered
            // probe side while a semi-join is uninhabited.
            PlanBuilderRoot::Absent(_) if inverted => {
                return Ok(PlanBuilder(PlanBuilderRoot::Present(probe)))
            }
            PlanBuilderRoot::Absent(_) => return Ok(Self::absent(Arc::clone(&probe.schema))),
        };
        let node = SemiJoin {
            inverted,
            probe_keys,
            build_keys,
        };
        let schema = Arc::clone(&probe.schema);
        Ok(Self::present(schema, node, vec![probe, build]))
    }

    /// Unordered bag union of `inputs` into a [`UnionAll`]. All inputs must share the same schema;
    /// absent inputs are dropped, and a lone present input is forwarded unchanged.
    ///
    /// Produces an error when `inputs` is empty, or two inputs have differing schemas.
    ///
    /// # Example
    /// ```
    /// # use std::sync::Arc;
    /// # use delta_kernel::PlanBuilder;
    /// # use delta_kernel::schema::{DataType, StructField, StructType};
    /// let schema = Arc::new(StructType::try_new([StructField::not_null("id", DataType::INTEGER)])?);
    /// let a = PlanBuilder::values(Arc::clone(&schema), vec![vec![1.into()]])?;
    /// let b = PlanBuilder::values(schema, vec![vec![2.into()]])?;
    /// let plan = PlanBuilder::union_all([a, b])?.build()?;
    /// # Ok::<(), delta_kernel::Error>(())
    /// ```
    pub fn union_all(inputs: impl IntoIterator<Item = PlanBuilder>) -> DeltaResult<Self> {
        let inputs = Vec::from_iter(inputs);
        let Some(schema) = inputs.first().map(|b| Arc::clone(b.schema())) else {
            return Err(Error::generic("union_all: requires at least one input"));
        };
        if let Some(i) = inputs.iter().position(|b| b.schema() != &schema) {
            return Err(Error::generic(format!(
                "union_all: input {i} has a schema differing from input 0: {:?} vs {:?}",
                inputs[i].schema(),
                schema,
            )));
        }
        let mut present = Vec::from_iter(inputs.into_iter().filter_map(|b| match b.0 {
            PlanBuilderRoot::Present(node) => Some(node),
            PlanBuilderRoot::Absent(_) => None,
        }));
        Ok(match present.len() {
            0 => Self::absent(schema),
            1 => PlanBuilder(PlanBuilderRoot::Present(present.swap_remove(0))),
            _ => Self::present(schema, UnionAll, present),
        })
    }

    /// Linearize the DAG reachable from `self` into a [`Plan`]. An absent relation builds to a
    /// single empty [`Values`] node carrying its schema, so the result is always a runnable plan.
    pub fn build(&self) -> DeltaResult<Plan> {
        Ok(match &self.0 {
            PlanBuilderRoot::Present(root) => Self::build_plan(root),
            PlanBuilderRoot::Absent(schema) => Plan {
                nodes: vec![PlanNode::new(
                    Values::new(Arc::clone(schema), vec![]),
                    vec![],
                )],
            },
        })
    }

    /// Like [`Self::build`], but yields `None` for an absent relation instead of an empty plan.
    pub fn build_opt(&self) -> DeltaResult<Option<Plan>> {
        Ok(match &self.0 {
            PlanBuilderRoot::Present(root) => Some(Self::build_plan(root)),
            PlanBuilderRoot::Absent(_) => None,
        })
    }

    /// Linearize the DAG rooted at `root` into a topologically sorted [`Plan`] with `root` as the
    /// last (terminal) node. A node's inputs always precede it, referenced by their index in
    /// [`Plan::nodes`]. Shared subgraphs are emitted once; multiple nodes can reference them.
    fn build_plan(root: &BuilderNodeRef) -> Plan {
        // Maps a node's Arc pointer to its index, deduping shared subgraphs. Pointers are stable
        // because `root` pins all reachable nodes for the lifetime of `build_plan`.
        fn emit(
            node: &BuilderNodeRef,
            nodes: &mut Vec<PlanNode>,
            emitted: &mut HashMap<*const BuilderNode, usize>,
        ) -> usize {
            let key = Arc::as_ptr(node);
            if let Some(&index) = emitted.get(&key) {
                return index;
            }
            // Recurse before `entry` so the closure borrows only `nodes`, not `emitted`.
            let inputs = Vec::from_iter(node.inputs.iter().map(|i| emit(i, nodes, emitted)));
            *emitted.entry(key).or_insert_with(|| {
                nodes.push(PlanNode::new(node.op.clone(), inputs));
                nodes.len() - 1
            })
        }
        let mut nodes = Vec::new();
        emit(root, &mut nodes, &mut HashMap::new());
        Plan { nodes }
    }
}

/// A literal relation. Empty rows is the absent relation, which the builder eliminates as dead
/// code (see the module docs).
impl From<Values> for PlanBuilder {
    fn from(values: Values) -> Self {
        match values.rows.is_empty() {
            true => Self::absent(values.schema),
            false => Self::present(values.schema.clone(), values, vec![]),
        }
    }
}

/// Error if any of `cols` fails to resolve against `schema` (nested paths supported).
fn check_columns_resolve<'a>(
    schema: &SchemaRef,
    cols: impl IntoIterator<Item = &'a ColumnName>,
    ctx: &str,
) -> DeltaResult<()> {
    for col in cols {
        schema.field_at(col).map_err(|_| {
            Error::generic(format!(
                "{ctx}: column `{col}` not found; schema has {:?}",
                Vec::from_iter(schema.fields().map(|f| f.name())),
            ))
        })?;
    }
    Ok(())
}

/// Error unless each of `names` is a top-level, non-metadata field of `schema`. A file-constant is
/// one scalar broadcast into a top-level field, so nested paths and engine-generated metadata
/// columns are rejected.
fn check_file_constant_columns<'a>(
    schema: &SchemaRef,
    names: impl IntoIterator<Item = &'a String>,
    ctx: &str,
) -> DeltaResult<()> {
    for name in names {
        let Some(field) = schema.field(name.as_str()) else {
            return Err(Error::generic(format!(
                "{ctx}: column `{name}` not found; schema has {:?}",
                Vec::from_iter(schema.fields().map(|f| f.name())),
            )));
        };
        if field.is_metadata_column() {
            return Err(Error::generic(format!(
                "{ctx}: column `{name}` is a metadata column"
            )));
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use test_utils::assert_result_error_with_message;

    use super::*;
    use crate::actions::deletion_vector::DeletionVectorDescriptor;
    use crate::expressions::{col, column_name, lit, Expression};
    use crate::plans::ir::nodes::FileType;
    use crate::schema::{
        schema, schema_ref, DataType, MetadataColumnSpec, StructField, StructType,
    };
    use crate::FileMeta;

    /// A single-file scan (present), no file-constant columns -- the trivial scan fixture.
    fn scan(schema: SchemaRef) -> PlanBuilder {
        PlanBuilder::scan_parquet([test_file("file:///a.parquet")], &[], schema).unwrap()
    }

    fn test_file(path: &str) -> FileMeta {
        FileMeta {
            location: url::Url::parse(path).unwrap(),
            last_modified: 0,
            size: 0,
        }
    }

    fn id_schema() -> SchemaRef {
        schema_ref! {
            nullable "id": STRING,
        }
    }

    fn x_schema() -> SchemaRef {
        schema_ref! {
            nullable "x": LONG,
        }
    }

    /// A one-row (all-null) `Values` source over `schema` -- the common present-source fixture.
    /// (Empty rows would collapse to absent.)
    fn vals(schema: SchemaRef) -> PlanBuilder {
        let row = Vec::from_iter(schema.fields().map(|f| Scalar::null(f.data_type().clone())));
        PlanBuilder::values(schema, vec![row]).unwrap()
    }

    /// The absent (uninhabited) relation over `schema`.
    fn absent_over(schema: SchemaRef) -> PlanBuilder {
        PlanBuilder::values(schema, vec![]).unwrap()
    }

    /// The absent (uninhabited) relation over `id_schema`.
    fn absent_src() -> PlanBuilder {
        absent_over(id_schema())
    }

    /// Build `builder` and assert its nodes match `expected` one-for-one in order, each described
    /// by its `(inputs, op)`: the indices of the nodes it references and its operator
    /// tag. A node is identified by its index in `nodes`. Returns the built [`Plan`] for further
    /// inspection.
    fn assert_plan(builder: PlanBuilder, expected: &[(&[usize], &str)]) -> Plan {
        let plan = builder.build().unwrap();
        assert_eq!(plan.nodes.len(), expected.len(), "node count");
        for (i, (node, &(inputs, op))) in plan.nodes.iter().zip(expected).enumerate() {
            assert_eq!(node.inputs.as_slice(), inputs, "node {i} inputs");
            assert_eq!(node.op.to_string(), op, "node {i} op");
        }
        plan
    }

    /// A source records its declared schema and builds to a single node.
    #[test]
    fn source_builds_single_node() {
        let src = scan(id_schema());
        assert_eq!(src.schema(), &id_schema());
        assert_plan(src, &[(&[], "scan_parquet")]);
    }

    /// `{ id, part }`, with `part` used as a file-constant column.
    fn part_schema() -> SchemaRef {
        schema_ref! {
            nullable "id": STRING,
            nullable "part": STRING,
        }
    }

    /// A single-file scan with one file constant `"p1"` for the `part` column.
    fn one_constant_file(path: &str) -> ScanFile {
        ScanFile {
            meta: test_file(path),
            file_constants: vec!["p1".into()],
        }
    }

    /// Both scan sources carry file-constant columns and record them on the node.
    #[test]
    fn scans_record_file_constant_columns() -> DeltaResult<()> {
        let part = vec!["part".to_string()];

        let parquet = PlanBuilder::scan_parquet(
            vec![one_constant_file("file:///a.parquet")],
            &["part"],
            part_schema(),
        )?;
        assert_eq!(parquet.schema(), &part_schema());
        let plan = assert_plan(parquet, &[(&[], "scan_parquet")]);
        let Operator::ScanParquet(node) = &plan.nodes[0].op else {
            panic!("expected ScanParquet");
        };
        assert_eq!(node.file_constant_columns, part);

        let json = PlanBuilder::scan_json(
            vec![one_constant_file("file:///a.json")],
            &["part"],
            part_schema(),
        )?;
        let plan = assert_plan(json, &[(&[], "scan_json")]);
        let Operator::ScanJson(node) = &plan.nodes[0].op else {
            panic!("expected ScanJson");
        };
        assert_eq!(node.file_constant_columns, part);
        Ok(())
    }

    /// Filter passes the input schema through unchanged.
    #[test]
    fn filter_preserves_schema() -> DeltaResult<()> {
        let filtered = vals(id_schema()).filter(col!("id").is_not_null())?;
        assert_eq!(filtered.schema(), &id_schema());
        Ok(())
    }

    /// A method chain linearizes in topological order with the terminal last.
    #[test]
    fn monadic_chain_is_topological() -> DeltaResult<()> {
        assert_plan(
            scan(id_schema())
                .filter(col!("id").is_not_null())?
                .filter(col!("id").is_null())?,
            &[(&[], "scan_parquet"), (&[0], "filter"), (&[1], "filter")],
        );
        Ok(())
    }

    /// A value reused by two consumers (SSA style) is emitted exactly once: the shared source
    /// node appears a single time and both consumers reference its index.
    #[test]
    fn shared_source_emitted_once() -> DeltaResult<()> {
        let a = vals(id_schema());
        let b = a.clone().filter(col!("id").is_not_null())?;
        let c = a.filter(col!("id").is_null())?;

        // `a` is emitted once as node 0; both filters reference it.
        assert_plan(
            PlanBuilder::union_all([b, c])?,
            &[
                (&[], "values"),
                (&[0], "filter"),
                (&[0], "filter"),
                (&[1, 2], "union_all"),
            ],
        );
        Ok(())
    }

    /// Nodes not reachable from the terminal are dropped (dead-code elimination).
    #[test]
    fn unreachable_nodes_are_dropped() -> DeltaResult<()> {
        let src = vals(id_schema());
        let _dead = src.clone().filter(col!("id").is_null())?; // never reaches the terminal
        let kept = src.filter(col!("id").is_not_null())?;

        // Only src(0) and the kept filter(1) survive.
        assert_plan(kept, &[(&[], "values"), (&[0], "filter")]);
        Ok(())
    }

    /// A shared *intermediate* (a node with its own input) feeding two consumers is emitted once,
    /// and both consumers reference its single index.
    #[test]
    fn shared_intermediate_emitted_once_diamond() -> DeltaResult<()> {
        let mid = vals(id_schema()).filter(col!("id").is_not_null())?;
        let left = mid.clone().filter(col!("id").is_not_null())?;
        let right = mid.filter(col!("id").is_null())?;

        // src(0) and mid(1) each emitted once; left(2) and right(3) both reference mid.
        assert_plan(
            PlanBuilder::union_all([left, right])?,
            &[
                (&[], "values"),
                (&[0], "filter"),
                (&[1], "filter"),
                (&[1], "filter"),
                (&[2, 3], "union_all"),
            ],
        );
        Ok(())
    }

    /// `union_all` over two or more present inputs keeps `self`'s schema and records all inputs.
    #[test]
    fn union_all_records_present_inputs() -> DeltaResult<()> {
        let inputs = Vec::from_iter((0..3).map(|_| vals(id_schema())));
        assert_plan(
            PlanBuilder::union_all(inputs)?,
            &[
                (&[], "values"),
                (&[], "values"),
                (&[], "values"),
                (&[0, 1, 2], "union_all"),
            ],
        );
        Ok(())
    }

    /// `union_all` with a single present input forwards it unchanged -- no `UnionAll` node.
    #[test]
    fn union_all_of_one_forwards() -> DeltaResult<()> {
        assert_plan(
            PlanBuilder::union_all([vals(id_schema())])?,
            &[(&[], "values")],
        );
        Ok(())
    }

    // === Absent (uninhabited) relation =======================================

    /// Empty sources and transforms over an absent input all collapse to the absent relation, which
    /// `build_opt` reports as `None`.
    #[rstest::rstest]
    #[case::empty_parquet(PlanBuilder::scan_parquet(Vec::<FileMeta>::new(), &[], id_schema()))]
    #[case::empty_json(PlanBuilder::scan_json(Vec::<FileMeta>::new(), &[], id_schema()))]
    #[case::empty_values(PlanBuilder::values(id_schema(), vec![]))]
    #[case::filter(absent_src().filter(col!("id").is_not_null()))]
    #[case::project(absent_src().project(Expression::struct_from([col!("id")]), id_schema()))]
    #[case::project_patch(absent_src().project_patch(|p| p.drop("id")))]
    #[case::dynamic_scan({
        let input = dynamic_scan_input_schema();
        dynamic_scan_node(&input, dynamic_scan_output_schema())
            .and_then(|dynamic_scan| absent_over(input).dynamic_scan(dynamic_scan))
    })]
    #[case::grouped_aggregate(absent_src().aggregate(
        Aggregate::group_by(part_schema(), [column_name!("id")]).max(column_name!("part"))))]
    #[case::semi_join_absent_build(
        vals(id_schema()).semi_join(absent_src(), [column_name!("id")], [column_name!("id")]))]
    #[case::semi_join_absent_probe(
        absent_src().semi_join(vals(x_schema()), [column_name!("id")], [column_name!("x")]))]
    #[case::anti_join_absent_probe(
        absent_src().anti_join(vals(x_schema()), [column_name!("id")], [column_name!("x")]))]
    #[case::union_all_of_absent(PlanBuilder::union_all([absent_src(), absent_src()]))]
    fn collapses_to_absent(#[case] builder: DeltaResult<PlanBuilder>) -> DeltaResult<()> {
        assert!(builder?.build_opt()?.is_none()); // absent has no plan to run
        Ok(())
    }

    /// An absent relation still carries its schema, and `build` materializes it as an empty
    /// `Values` node.
    #[test]
    fn absent_carries_schema_and_builds_empty_values() -> DeltaResult<()> {
        let absent = absent_src();
        assert_eq!(absent.schema(), &id_schema());
        let plan = assert_plan(absent, &[(&[], "values")]);
        let Operator::Values(values) = &plan.nodes[0].op else {
            panic!("expected Values");
        };
        assert!(values.rows.is_empty());
        assert_eq!(values.schema, id_schema());
        Ok(())
    }

    /// `union_all` drops absent arms, so a lone present arm survives and is forwarded unchanged --
    /// no `UnionAll` node.
    #[test]
    fn union_all_forwards_lone_present_arm() -> DeltaResult<()> {
        assert_plan(
            PlanBuilder::union_all([absent_src(), vals(id_schema())])?,
            &[(&[], "values")],
        );
        Ok(())
    }

    /// An anti join with an absent build forwards its probe unchanged: it subtracts nothing.
    #[test]
    fn anti_join_over_absent_build_forwards_probe() -> DeltaResult<()> {
        let anti = vals(id_schema()).anti_join(
            absent_src(),
            [column_name!("id")],
            [column_name!("id")],
        )?;
        assert_plan(anti, &[(&[], "values")]);
        Ok(())
    }

    /// `aggregate` outputs the group keys followed by the aggregate columns.
    #[test]
    fn aggregate_outputs_keys_then_aggs() -> DeltaResult<()> {
        // Group by `id`, take the latest `part` by version -- output is `{id, part}`.
        let agg = vals(part_schema()).aggregate(
            Aggregate::group_by(part_schema(), [column_name!("id")]).max_non_null_by(
                column_name!("part"),
                column_name!("part"),
                column_name!("id"),
            ),
        )?;
        assert_eq!(agg.schema(), &part_schema());
        assert_plan(agg, &[(&[], "values"), (&[0], "aggregate")]);
        Ok(())
    }

    /// `aggregate_by` roots the `AggregateBuilder` at the input schema, yielding the same plan as
    /// `aggregate` with an explicitly schema'd builder.
    #[test]
    fn aggregate_by_infers_input_schema() -> DeltaResult<()> {
        let agg = vals(part_schema()).aggregate_by([column_name!("id")], |a| {
            a.max_non_null_by(
                column_name!("part"),
                column_name!("part"),
                column_name!("id"),
            )
        })?;
        assert_eq!(agg.schema(), &part_schema());
        assert_plan(agg, &[(&[], "values"), (&[0], "aggregate")]);
        Ok(())
    }

    /// `aggregate_ungrouped` roots the `AggregateBuilder` at the input schema without grouping.
    #[test]
    fn aggregate_ungrouped_infers_input_schema() -> DeltaResult<()> {
        let agg = vals(part_schema()).aggregate_ungrouped(|a| a.max(column_name!("part")))?;
        let schema = agg.schema();
        assert_eq!(schema.fields().count(), 1);
        assert!(schema.field("part").is_some());
        assert_plan(agg, &[(&[], "values"), (&[0], "aggregate")]);
        Ok(())
    }

    /// A global (no group keys) aggregate over an absent input aggregates an empty `Values`
    /// relation, so the engine produces the single output row.
    #[test]
    fn aggregate_global_over_absent_aggregates_empty_values() -> DeltaResult<()> {
        let agg = absent_src().aggregate(
            Aggregate::group_by(id_schema(), Vec::<ColumnName>::new()).max(column_name!("id")),
        )?;
        let plan = assert_plan(agg, &[(&[], "values"), (&[0], "aggregate")]);
        let Operator::Values(values) = &plan.nodes[0].op else {
            panic!("expected Values input");
        };
        assert!(values.rows.is_empty());
        Ok(())
    }

    /// A global (no group keys) aggregate over a present input outputs just the aggregate columns.
    #[test]
    fn aggregate_global_over_present_outputs_aggs_only() -> DeltaResult<()> {
        let agg = vals(part_schema()).aggregate(
            Aggregate::group_by(part_schema(), Vec::<ColumnName>::new()).max(column_name!("part")),
        )?;
        let schema = agg.schema();
        assert_eq!(schema.fields().count(), 1);
        assert!(schema.field("part").is_some());
        let plan = assert_plan(agg, &[(&[], "values"), (&[0], "aggregate")]);
        let Operator::Aggregate(node) = &plan.nodes[1].op else {
            panic!("expected Aggregate");
        };
        assert!(node.group_by.is_empty());
        Ok(())
    }

    /// `{ outer: { a, b }, c }` -- a nested schema for exercising `project_patch`.
    fn nested_ab_c() -> SchemaRef {
        schema_ref! {
            nullable "outer": {
                nullable "a": LONG,
                nullable "b": STRING,
            },
            nullable "c": LONG,
        }
    }

    /// `project_patch` lowers field edits and the output schema together: a nested replace, a
    /// top-level append, and a drop are all reflected in the resulting schema.
    #[test]
    fn project_patch_edits_track_schema() -> DeltaResult<()> {
        let patched = vals(nested_ab_c()).project_patch(|p| {
            p.replace_at(
                ["outer"],
                "b",
                StructField::nullable("b2", DataType::LONG),
                lit(0i64),
            )
            .append(StructField::nullable("d", DataType::LONG), lit(1i64))
            .drop("c")
        })?;

        let expected: SchemaRef = schema_ref! {
            nullable "outer": {
                nullable "a": LONG,
                nullable "b2": LONG,
            },
            nullable "d": LONG,
        };
        assert_eq!(patched.schema(), &expected);
        assert_plan(patched, &[(&[], "values"), (&[0], "project")]);
        Ok(())
    }

    /// A shared edit `fn` can be reused across multiple `project_patch` sites: it is higher-ranked
    /// over the patch-builder lifetime, which a `let` closure cannot express.
    #[test]
    fn project_patch_accepts_shared_fn() -> DeltaResult<()> {
        fn drop_c(p: ProjectionStructPatchBuilder<'_>) -> ProjectionStructPatchBuilder<'_> {
            p.drop("c")
        }
        let a = vals(nested_ab_c()).project_patch(drop_c)?;
        let b = vals(nested_ab_c()).project_patch(drop_c)?;
        assert_eq!(a.schema(), b.schema());
        Ok(())
    }

    /// `project` records the caller's declared output schema verbatim.
    #[test]
    fn project_records_declared_schema() -> DeltaResult<()> {
        let out = x_schema();
        let p = vals(id_schema()).project(Expression::struct_from([col!("id")]), out.clone())?;
        assert_eq!(p.schema(), &out);
        assert_plan(p, &[(&[], "values"), (&[0], "project")]);
        Ok(())
    }

    /// `semi_join`/`anti_join` mirror the probe schema, record `[probe, build]` inputs in order,
    /// and set `inverted` accordingly.
    #[rstest::rstest]
    fn join_mirrors_probe_and_orders_inputs(
        #[values(false, true)] inverted: bool,
    ) -> DeltaResult<()> {
        let probe = vals(id_schema());
        let build = vals(x_schema());
        let joined = if inverted {
            probe.anti_join(build, [column_name!("id")], [column_name!("x")])?
        } else {
            probe.semi_join(build, [column_name!("id")], [column_name!("x")])?
        };
        assert_eq!(joined.schema(), &id_schema());

        let plan = assert_plan(
            joined,
            &[(&[], "values"), (&[], "values"), (&[0, 1], "semi_join")],
        );
        let Operator::SemiJoin(node) = &plan.nodes[2].op else {
            panic!("expected SemiJoin");
        };
        assert_eq!(node.inverted, inverted);
        Ok(())
    }

    fn dynamic_scan_node(
        input_schema: &SchemaRef,
        out_schema: SchemaRef,
    ) -> DeltaResult<DynamicScan> {
        DynamicScan::try_new(
            input_schema,
            out_schema,
            FileType::Parquet,
            url::Url::parse("memory:///").unwrap(),
            ["version"],
            column_name!("path"),
            column_name!("size"),
            column_name!("filemod"),
            column_name!("dv"),
        )
    }

    fn dynamic_scan_input_schema() -> SchemaRef {
        schema_ref! {
            not_null "path": STRING,
            not_null "size": LONG,
            not_null "filemod": LONG,
            nullable "dv": (DeletionVectorDescriptor::to_schema()),
            nullable "version": LONG,
        }
    }

    fn dynamic_scan_output_schema() -> SchemaRef {
        schema_ref! {
            nullable "id": STRING,
            nullable "version": LONG,
        }
    }

    #[test]
    fn dynamic_scan_sets_output_schema_and_records_input() -> DeltaResult<()> {
        let input = dynamic_scan_input_schema();
        let out = dynamic_scan_output_schema();
        let node = dynamic_scan_node(&input, out.clone())?;
        let dynamic_scan = vals(input).dynamic_scan(node)?;
        assert_eq!(dynamic_scan.schema(), &out);
        assert_plan(dynamic_scan, &[(&[], "values"), (&[0], "dynamic_scan")]);
        Ok(())
    }

    fn dynamic_scan_input_missing(omit: &str) -> SchemaRef {
        let schema = dynamic_scan_input_schema();
        let fields = schema
            .fields()
            .filter(|field| field.name() != omit)
            .cloned();
        Arc::new(StructType::new_unchecked(fields))
    }

    fn schema_with_field(schema: SchemaRef, replacement: StructField) -> SchemaRef {
        let fields = schema.fields().map(|field| {
            if field.name() == replacement.name() {
                replacement.clone()
            } else {
                field.clone()
            }
        });
        Arc::new(StructType::new_unchecked(fields))
    }

    fn dynamic_scan_input_with(field: StructField) -> SchemaRef {
        schema_with_field(dynamic_scan_input_schema(), field)
    }

    fn construct_dynamic_scan(input: SchemaRef) -> DeltaResult<DynamicScan> {
        dynamic_scan_node(&input, dynamic_scan_output_schema())
    }

    fn build_dynamic_scan_with_schemas(
        input: SchemaRef,
        output: SchemaRef,
    ) -> DeltaResult<PlanBuilder> {
        let dynamic_scan = dynamic_scan_node(&input, output)?;
        vals(input).dynamic_scan(dynamic_scan)
    }

    #[rstest::rstest]
    #[case::path("path", 0)]
    #[case::size("size", 1)]
    #[case::filemod("filemod", 2)]
    fn dynamic_scan_validates_nested_metadata(
        #[case] column: &str,
        #[case] index: usize,
        #[values(false, true)] parent_nullable: bool,
    ) {
        let base_schema = dynamic_scan_input_schema();
        let metadata = StructField::new(
            "metadata",
            schema! {
                not_null "path": STRING,
                not_null "size": LONG,
                not_null "filemod": LONG,
            },
            parent_nullable,
        );
        let input = schema_ref! {
            ..(base_schema.fields()),
            (metadata),
        };
        let mut columns = [
            column_name!("path"),
            column_name!("size"),
            column_name!("filemod"),
        ];
        columns[index] = ColumnName::new(["metadata", column]);
        let [path_column, file_size_column, last_modified_column] = columns;
        let result = DynamicScan::try_new(
            &input,
            dynamic_scan_output_schema(),
            FileType::Parquet,
            url::Url::parse("memory:///").unwrap(),
            ["version"],
            path_column,
            file_size_column,
            last_modified_column,
            column_name!("dv"),
        )
        .and_then(|dynamic_scan| vals(input).dynamic_scan(dynamic_scan));

        if parent_nullable {
            assert_result_error_with_message(result, &format!("`metadata.{column}` is nullable"));
        } else {
            let plan = result.unwrap();
            assert_eq!(plan.schema(), &dynamic_scan_output_schema());
        }
    }

    #[rstest::rstest]
    fn dynamic_scan_accepts_dv_with_nullable_ancestor(
        #[values(false, true)] parent_nullable: bool,
    ) {
        let metadata = StructField::new(
            "metadata",
            schema! { nullable "dv": (DeletionVectorDescriptor::to_schema()) },
            parent_nullable,
        );
        let base_schema = dynamic_scan_input_schema();
        let input = schema_ref! {
            ..(base_schema.fields()),
            (metadata),
        };

        DynamicScan::try_new(
            &input,
            dynamic_scan_output_schema(),
            FileType::Parquet,
            url::Url::parse("memory:///").unwrap(),
            ["version"],
            column_name!("path"),
            column_name!("size"),
            column_name!("filemod"),
            column_name!("metadata.dv"),
        )
        .unwrap();
    }

    /// Every validating method surfaces its malformed input as an error at the call site, with a
    /// message that names the offending column or mismatch.
    #[rstest::rstest]
    // sources
    #[case::scan_file_constant_arity("constant value",
        || PlanBuilder::scan_parquet(vec![one_constant_file("file:///a.parquet")], &[], part_schema()))]
    #[case::scan_unknown_file_constant("`nope` not found",
        || PlanBuilder::scan_parquet(vec![one_constant_file("file:///a.parquet")], &["nope"], part_schema()))]
    #[case::values_row_width("schema has 1 field",
        || PlanBuilder::values(id_schema(), vec![vec!["a".into(), "b".into()]]))]
    // transforms referencing an absent column
    #[case::filter_unknown_column("`nope` not found",
        || vals(id_schema()).filter(col!("nope").is_not_null()))]
    #[case::project_unknown_column("`nope` not found",
        || vals(id_schema()).project(Expression::struct_from([col!("nope")]), id_schema()))]
    #[case::project_patch_missing_field("does not exist",
        || vals(id_schema()).project_patch(|p| p.drop("nope")))]
    #[case::aggregate_unknown_value("not found in schema",
        || vals(id_schema()).aggregate(Aggregate::group_by(id_schema(), Vec::<ColumnName>::new()).max(column_name!("nope"))))]
    #[case::aggregate_unknown_group_by("not found in schema",
        || vals(id_schema()).aggregate(Aggregate::group_by(id_schema(), [column_name!("nope")]).max(column_name!("id"))))]
    // union
    #[case::union_schema_disagrees("differing",
        || PlanBuilder::union_all([vals(id_schema()), vals(x_schema())]))]
    #[case::union_empty("at least one input", || PlanBuilder::union_all([]))]
    // semi-join keys
    #[case::join_unknown_probe_key("join probe: column `nope`",
        || vals(id_schema()).semi_join(vals(x_schema()), [column_name!("nope")], [column_name!("x")]))]
    #[case::join_unknown_build_key("join build: column `nope`",
        || vals(id_schema()).anti_join(vals(x_schema()), [column_name!("id")], [column_name!("nope")]))]
    #[case::join_key_arity("must match",
        || vals(id_schema()).semi_join(vals(x_schema()), [column_name!("id")], [column_name!("x"), column_name!("x")]))]
    // dynamic scan file constants
    #[case::dynamic_scan_missing_source("`version`", || build_dynamic_scan_with_schemas(
        dynamic_scan_input_missing("version"),
        dynamic_scan_output_schema(),
    ))]
    #[case::dynamic_scan_absent_from_output("`version`", || {
        let input = dynamic_scan_input_schema();
        let dynamic_scan = dynamic_scan_node(&input, id_schema())?;
        vals(input).dynamic_scan(dynamic_scan)
    })]
    #[case::dynamic_scan_type_mismatch("same type and nullability", || {
        let output = schema_with_field(
            dynamic_scan_output_schema(),
            StructField::nullable("version", DataType::STRING),
        );
        build_dynamic_scan_with_schemas(dynamic_scan_input_schema(), output)
    })]
    #[case::dynamic_scan_input_nullable("same type and nullability", || {
        let output = schema_with_field(
            dynamic_scan_output_schema(),
            StructField::not_null("version", DataType::LONG),
        );
        build_dynamic_scan_with_schemas(dynamic_scan_input_schema(), output)
    })]
    #[case::dynamic_scan_output_nullable("same type and nullability", || {
        let input = dynamic_scan_input_with(StructField::not_null("version", DataType::LONG));
        build_dynamic_scan_with_schemas(input, dynamic_scan_output_schema())
    })]
    #[case::dynamic_scan_metadata_source("metadata column", || {
        let input = dynamic_scan_input_with(
            StructField::create_metadata_column("version", MetadataColumnSpec::RowIndex),
        );
        build_dynamic_scan_with_schemas(input, dynamic_scan_output_schema())
    })]
    #[case::dynamic_scan_metadata_output("metadata column", || {
        let output = schema_with_field(
            dynamic_scan_output_schema(),
            StructField::create_metadata_column("version", MetadataColumnSpec::RowIndex),
        );
        build_dynamic_scan_with_schemas(dynamic_scan_input_schema(), output)
    })]
    fn rejects(#[case] needle: &str, #[case] make: impl Fn() -> DeltaResult<PlanBuilder>) {
        assert_result_error_with_message(make(), needle);
    }

    #[rstest::rstest]
    #[case::missing_path("path", || construct_dynamic_scan(dynamic_scan_input_missing("path")))]
    #[case::missing_size("size", || construct_dynamic_scan(dynamic_scan_input_missing("size")))]
    #[case::missing_filemod("filemod", || construct_dynamic_scan(dynamic_scan_input_missing("filemod")))]
    #[case::nullable_path("required column `path` is nullable", || construct_dynamic_scan(dynamic_scan_input_with(StructField::nullable("path", DataType::STRING))))]
    #[case::nullable_size("required column `size` is nullable", || construct_dynamic_scan(dynamic_scan_input_with(StructField::nullable("size", DataType::LONG))))]
    #[case::nullable_filemod("required column `filemod` is nullable", || construct_dynamic_scan(dynamic_scan_input_with(StructField::nullable("filemod", DataType::LONG))))]
    #[case::wrong_path_type("must have type string", || construct_dynamic_scan(dynamic_scan_input_with(StructField::not_null("path", DataType::BOOLEAN))))]
    #[case::wrong_size_type("must have type long", || construct_dynamic_scan(dynamic_scan_input_with(StructField::not_null("size", DataType::STRING))))]
    #[case::wrong_filemod_type("must have type long", || construct_dynamic_scan(dynamic_scan_input_with(StructField::not_null("filemod", DataType::STRING))))]
    #[case::missing_dv("`dv`", || construct_dynamic_scan(dynamic_scan_input_missing("dv")))]
    #[case::wrong_dv_type("deletion-vector column `dv` must have type", || construct_dynamic_scan(dynamic_scan_input_with(StructField::nullable("dv", DataType::STRING))))]
    #[case::non_nullable_dv("deletion-vector column `dv` must be nullable", || construct_dynamic_scan(dynamic_scan_input_with(StructField::not_null("dv", DeletionVectorDescriptor::to_schema()))))]
    #[case::opaque_base_url("must be hierarchical and end in `/`", || {
        let input = dynamic_scan_input_schema();
        DynamicScan::try_new(
            &input,
            dynamic_scan_output_schema(),
            FileType::Parquet,
            url::Url::parse("mailto:user@example.com").unwrap(),
            ["version"],
            column_name!("path"),
            column_name!("size"),
            column_name!("filemod"),
            column_name!("dv"),
        )
    })]
    #[case::base_url_without_trailing_slash("must be hierarchical and end in `/`", || {
        let input = dynamic_scan_input_schema();
        DynamicScan::try_new(
            &input,
            dynamic_scan_output_schema(),
            FileType::Parquet,
            url::Url::parse("s3://bucket/table/data").unwrap(),
            ["version"],
            column_name!("path"),
            column_name!("size"),
            column_name!("filemod"),
            column_name!("dv"),
        )
    })]
    fn dynamic_scan_constructor_rejects_invalid_configuration(
        #[case] needle: &str,
        #[case] make: impl Fn() -> DeltaResult<DynamicScan>,
    ) {
        assert_result_error_with_message(make(), needle);
    }

    #[test]
    fn dynamic_scan_attachment_revalidates_input_schema() {
        let input = dynamic_scan_input_with(StructField::nullable("size", DataType::LONG));
        let valid_input = dynamic_scan_input_schema();
        let dynamic_scan = dynamic_scan_node(&valid_input, dynamic_scan_output_schema()).unwrap();
        assert_result_error_with_message(
            vals(input).dynamic_scan(dynamic_scan),
            "required column `size` is nullable",
        );
    }

    #[test]
    fn dynamic_scan_attachment_revalidates_base_url() {
        let input = dynamic_scan_input_schema();
        let mut dynamic_scan = dynamic_scan_node(&input, dynamic_scan_output_schema()).unwrap();
        dynamic_scan.base_url = url::Url::parse("s3://bucket/table/data").unwrap();

        assert_result_error_with_message(
            vals(input).dynamic_scan(dynamic_scan),
            "must be hierarchical and end in `/`",
        );
    }
}