datafusion_physical_plan/

projection.rs

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

//! Defines the projection execution plan. A projection determines which columns or expressions
//! are returned from a query. The SQL statement `SELECT a, b, a+b FROM t1` is an example
//! of a projection on table `t1` where the expressions `a`, `b`, and `a+b` are the
//! projection expressions. `SELECT` without `FROM` will only evaluate expressions.

use super::expressions::Column;
use super::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
use super::{
    DisplayAs, ExecutionPlanProperties, PlanProperties, RecordBatchStream,
    SendableRecordBatchStream, SortOrderPushdownResult, Statistics,
};
use crate::column_rewriter::PhysicalColumnRewriter;
use crate::execution_plan::{CardinalityEffect, replace_children_if_necessary};
use crate::filter_pushdown::{
    ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
    FilterPushdownPropagation, FilterRemapper, PushedDownPredicate,
};
use crate::joins::utils::{ColumnIndex, JoinFilter, JoinOn, JoinOnRef};
use crate::statistics::{ChildStats, StatisticsArgs};
use crate::{
    ChildrenPropertiesMode, DisplayFormatType, ExecutionPlan, PhysicalExpr,
    ReplaceChildrenOptions, validate_child_count,
};
use std::collections::HashMap;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};

use arrow::datatypes::{Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use datafusion_common::config::ConfigOptions;
use datafusion_common::tree_node::{
    Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
};
use datafusion_common::{DataFusionError, JoinSide, Result, internal_err, plan_err};
use datafusion_execution::TaskContext;
use datafusion_expr::ExpressionPlacement;
use datafusion_physical_expr::equivalence::ProjectionMapping;
use datafusion_physical_expr::projection::Projector;
use datafusion_physical_expr_common::physical_expr::{PhysicalExprRef, fmt_sql};
use datafusion_physical_expr_common::sort_expr::{
    LexOrdering, LexRequirement, PhysicalSortExpr,
};
// Re-exported from datafusion-physical-expr for backwards compatibility
// We recommend updating your imports to use datafusion-physical-expr directly
pub use datafusion_physical_expr::projection::{
    ProjectionExpr, ProjectionExprs, update_expr,
};

use futures::stream::{Stream, StreamExt};
use log::trace;

/// [`ExecutionPlan`] for a projection
///
/// Computes a set of scalar value expressions for each input row, producing one
/// output row for each input row.
#[derive(Debug, Clone)]
pub struct ProjectionExec {
    /// A projector specialized to apply the projection to the input schema from the child node
    /// and produce [`RecordBatch`]es with the output schema of this node.
    projector: Projector,
    /// The input plan
    input: Arc<dyn ExecutionPlan>,
    /// Execution metrics
    metrics: ExecutionPlanMetricsSet,
    /// Cache holding plan properties like equivalences, output partitioning etc.
    cache: Arc<PlanProperties>,
}

impl ProjectionExec {
    /// Create a projection on an input
    ///
    /// # Example:
    /// Create a `ProjectionExec` to crate `SELECT a, a+b AS sum_ab FROM t1`:
    ///
    /// ```
    /// # use std::sync::Arc;
    /// # use arrow_schema::{Schema, Field, DataType};
    /// # use datafusion_expr::Operator;
    /// # use datafusion_physical_plan::ExecutionPlan;
    /// # use datafusion_physical_expr::expressions::{col, binary};
    /// # use datafusion_physical_plan::empty::EmptyExec;
    /// # use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
    /// # fn schema() -> Arc<Schema> {
    /// #  Arc::new(Schema::new(vec![
    /// #   Field::new("a", DataType::Int32, false),
    /// #   Field::new("b", DataType::Int32, false),
    /// # ]))
    /// # }
    /// #
    /// # fn input() -> Arc<dyn ExecutionPlan> {
    /// #  Arc::new(EmptyExec::new(schema()))
    /// # }
    /// #
    /// # fn main() {
    /// let schema = schema();
    /// // Create PhysicalExprs
    /// let a = col("a", &schema).unwrap();
    /// let b = col("b", &schema).unwrap();
    /// let a_plus_b = binary(Arc::clone(&a), Operator::Plus, b, &schema).unwrap();
    /// // create ProjectionExec
    /// let proj = ProjectionExec::try_new(
    ///     [
    ///         ProjectionExpr {
    ///             // expr a produces the column named "a"
    ///             expr: a,
    ///             alias: "a".to_string(),
    ///         },
    ///         ProjectionExpr {
    ///             // expr: a + b produces the column named "sum_ab"
    ///             expr: a_plus_b,
    ///             alias: "sum_ab".to_string(),
    ///         },
    ///     ],
    ///     input(),
    /// )
    /// .unwrap();
    /// # }
    /// ```
    pub fn try_new<I, E>(expr: I, input: Arc<dyn ExecutionPlan>) -> Result<Self>
    where
        I: IntoIterator<Item = E>,
        E: Into<ProjectionExpr>,
    {
        let input_schema = input.schema();
        let expr_arc = expr.into_iter().map(Into::into).collect::<Arc<_>>();
        let projection = ProjectionExprs::from_expressions(expr_arc);
        let projector = projection.make_projector(&input_schema)?;
        Self::try_from_projector(projector, input)
    }

    /// Create a projection using field and schema metadata from
    /// `projected_schema`.
    ///
    /// Field names, data types, and nullability are still derived from the physical
    /// projection expressions and the input plan; only field and schema metadata are
    /// taken from `projected_schema`.
    ///
    /// # Errors
    ///
    /// Returns an error if the projection cannot be applied to the input plan, or if
    /// `projected_schema` has a different number of fields than the projection.
    pub fn try_new_with_schema_metadata<I, E>(
        expr: I,
        input: Arc<dyn ExecutionPlan>,
        projected_schema: &Schema,
    ) -> Result<Self>
    where
        I: IntoIterator<Item = E>,
        E: Into<ProjectionExpr>,
    {
        let input_schema = input.schema();
        let expr_arc = expr.into_iter().map(Into::into).collect::<Arc<_>>();
        let projection = ProjectionExprs::from_expressions(expr_arc);
        let projector = projection
            .make_projector_with_schema_metadata(&input_schema, projected_schema)?;
        Self::try_from_projector(projector, input)
    }

    fn try_from_projector(
        projector: Projector,
        input: Arc<dyn ExecutionPlan>,
    ) -> Result<Self> {
        // Construct a map from the input expressions to the output expression of the Projection
        let projection_mapping =
            projector.projection().projection_mapping(&input.schema())?;
        let cache = Self::compute_properties(
            &input,
            &projection_mapping,
            Arc::clone(projector.output_schema()),
        )?;
        Ok(Self {
            projector,
            input,
            metrics: ExecutionPlanMetricsSet::new(),
            cache: Arc::new(cache),
        })
    }

    /// The projection expressions stored as tuples of (expression, output column name)
    pub fn expr(&self) -> &[ProjectionExpr] {
        self.projector.projection().as_ref()
    }

    /// The projection expressions as a [`ProjectionExprs`].
    pub fn projection_expr(&self) -> &ProjectionExprs {
        self.projector.projection()
    }

    /// The input plan
    pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
        &self.input
    }

    /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
    fn compute_properties(
        input: &Arc<dyn ExecutionPlan>,
        projection_mapping: &ProjectionMapping,
        schema: SchemaRef,
    ) -> Result<PlanProperties> {
        // Calculate equivalence properties:
        let input_eq_properties = input.equivalence_properties();
        let eq_properties = input_eq_properties.project(projection_mapping, schema);
        // Calculate output partitioning, which needs to respect aliases:
        let output_partitioning = input
            .output_partitioning()
            .project(projection_mapping, input_eq_properties);

        Ok(PlanProperties::new(
            eq_properties,
            output_partitioning,
            input.pipeline_behavior(),
            input.boundedness(),
        ))
    }

    /// Collect reverse alias mapping from projection expressions.
    /// The result hash map is a map from aliased Column in parent to original expr.
    fn collect_reverse_alias(
        &self,
    ) -> Result<datafusion_common::HashMap<Column, Arc<dyn PhysicalExpr>>> {
        let mut alias_map = datafusion_common::HashMap::new();
        for projection in self.projection_expr().iter() {
            let (aliased_index, _output_field) = self
                .projector
                .output_schema()
                .column_with_name(&projection.alias)
                .ok_or_else(|| {
                    DataFusionError::Internal(format!(
                        "Expr {} with alias {} not found in output schema",
                        projection.expr, projection.alias
                    ))
                })?;
            let aliased_col = Column::new(&projection.alias, aliased_index);
            alias_map.insert(aliased_col, Arc::clone(&projection.expr));
        }
        Ok(alias_map)
    }
}

impl DisplayAs for ProjectionExec {
    fn fmt_as(
        &self,
        t: DisplayFormatType,
        f: &mut std::fmt::Formatter,
    ) -> std::fmt::Result {
        match t {
            DisplayFormatType::Default | DisplayFormatType::Verbose => {
                let expr: Vec<String> = self
                    .projector
                    .projection()
                    .as_ref()
                    .iter()
                    .map(|proj_expr| {
                        let e = proj_expr.expr.to_string();
                        if e != proj_expr.alias {
                            format!("{e} as {}", proj_expr.alias)
                        } else {
                            e
                        }
                    })
                    .collect();

                write!(f, "ProjectionExec: expr=[{}]", expr.join(", "))
            }
            DisplayFormatType::TreeRender => {
                for (i, proj_expr) in self.expr().iter().enumerate() {
                    let expr_sql = fmt_sql(proj_expr.expr.as_ref());
                    if proj_expr.expr.to_string() == proj_expr.alias {
                        writeln!(f, "expr{i}={expr_sql}")?;
                    } else {
                        writeln!(f, "{}={expr_sql}", proj_expr.alias)?;
                    }
                }

                Ok(())
            }
        }
    }
}

impl ExecutionPlan for ProjectionExec {
    fn name(&self) -> &'static str {
        "ProjectionExec"
    }

    /// Return a reference to Any that can be used for downcasting
    fn properties(&self) -> &Arc<PlanProperties> {
        &self.cache
    }

    fn maintains_input_order(&self) -> Vec<bool> {
        // Tell optimizer this operator doesn't reorder its input
        vec![true]
    }

    fn benefits_from_input_partitioning(&self) -> Vec<bool> {
        let all_simple_exprs =
            self.projector
                .projection()
                .as_ref()
                .iter()
                .all(|proj_expr| {
                    !matches!(
                        proj_expr.expr.placement(),
                        ExpressionPlacement::KeepInPlace
                    )
                });
        // If expressions are all either column_expr or Literal (or other cheap expressions),
        // then all computations in this projection are reorder or rename,
        // and projection would not benefit from the repartition, benefits_from_input_partitioning will return false.
        vec![!all_simple_exprs]
    }

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

    fn apply_expressions(
        &self,
        f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
    ) -> Result<TreeNodeRecursion> {
        crate::apply_expression_roots(self.projector.projection().as_ref().iter(), f)
    }

    fn replace_children(
        self: Arc<Self>,
        mut children: Vec<Arc<dyn ExecutionPlan>>,
        options: ReplaceChildrenOptions,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        validate_child_count!(self, children);
        match options.children_properties {
            ChildrenPropertiesMode::Keep => Ok(Arc::new(Self {
                input: children.swap_remove(0),
                metrics: ExecutionPlanMetricsSet::new(),
                ..Self::clone(&*self)
            })),
            ChildrenPropertiesMode::Recompute => ProjectionExec::try_from_projector(
                self.projector.clone(),
                children.swap_remove(0),
            )
            .map(|p| Arc::new(p) as _),
        }
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        self.replace_children(
            children,
            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
        )
    }

    fn with_new_children_and_same_properties(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        self.replace_children(
            children,
            ReplaceChildrenOptions::new(ChildrenPropertiesMode::Keep),
        )
    }

    fn execute(
        &self,
        partition: usize,
        context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        trace!(
            "Start ProjectionExec::execute for partition {} of context session_id {} and task_id {:?}",
            partition,
            context.session_id(),
            context.task_id()
        );

        let projector = self.projector.with_metrics(&self.metrics, partition);
        Ok(Box::pin(ProjectionStream::new(
            projector,
            self.input.execute(partition, context)?,
            BaselineMetrics::new(&self.metrics, partition),
        )?))
    }

    fn metrics(&self) -> Option<MetricsSet> {
        Some(self.metrics.clone_inner())
    }

    fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
        vec![ChildStats::At(partition)]
    }

    fn statistics_from_inputs(
        &self,
        input_stats: &[Arc<Statistics>],
        _args: &StatisticsArgs,
    ) -> Result<Arc<Statistics>> {
        let input_stats = input_stats[0].as_ref().clone();
        let output_schema = self.schema();
        Ok(Arc::new(
            self.projector
                .projection()
                .project_statistics(input_stats, &output_schema)?,
        ))
    }

    fn supports_limit_pushdown(&self) -> bool {
        true
    }

    fn cardinality_effect(&self) -> CardinalityEffect {
        CardinalityEffect::Equal
    }

    fn try_swapping_with_projection(
        &self,
        projection: &ProjectionExec,
    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
        match try_collapse_projection_chain(projection)? {
            Some(plan) => Ok(Some(plan)),
            None => Ok(Some(Arc::new(projection.clone()))),
        }
    }

    fn gather_filters_for_pushdown(
        &self,
        _phase: FilterPushdownPhase,
        parent_filters: Vec<Arc<dyn PhysicalExpr>>,
        _config: &ConfigOptions,
    ) -> Result<FilterDescription> {
        // expand alias column to original expr in parent filters
        let invert_alias_map = self.collect_reverse_alias()?;
        let output_schema = self.schema();
        let remapper = FilterRemapper::new(output_schema);
        let mut child_parent_filters = Vec::with_capacity(parent_filters.len());

        for filter in parent_filters {
            // Check that column exists in child, then reassign column indices to match child schema
            if let Some(reassigned) = remapper.try_remap(&filter)? {
                // rewrite filter expression using invert alias map
                let mut rewriter = PhysicalColumnRewriter::new(&invert_alias_map);
                let rewritten = reassigned.rewrite(&mut rewriter)?.data;
                child_parent_filters.push(PushedDownPredicate::supported(rewritten));
            } else {
                child_parent_filters.push(PushedDownPredicate::unsupported(filter));
            }
        }

        Ok(FilterDescription::new().with_child(ChildFilterDescription {
            parent_filters: child_parent_filters,
            self_filters: vec![],
        }))
    }

    fn handle_child_pushdown_result(
        &self,
        _phase: FilterPushdownPhase,
        child_pushdown_result: ChildPushdownResult,
        _config: &ConfigOptions,
    ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
        Ok(FilterPushdownPropagation::if_all(child_pushdown_result))
    }

    fn try_pushdown_sort(
        &self,
        order: &[PhysicalSortExpr],
    ) -> Result<SortOrderPushdownResult<Arc<dyn ExecutionPlan>>> {
        let child = self.input();
        let mut child_order = Vec::new();

        // Check and transform sort expressions
        for sort_expr in order {
            // Recursively transform the expression
            let mut can_pushdown = true;
            let transformed = Arc::clone(&sort_expr.expr).transform(|expr| {
                if let Some(col) = expr.downcast_ref::<Column>() {
                    // Check if column index is valid.
                    // This should always be true but fail gracefully if it's not.
                    if col.index() >= self.expr().len() {
                        can_pushdown = false;
                        return Ok(Transformed::no(expr));
                    }

                    let proj_expr = &self.expr()[col.index()];

                    // Check if projection expression is a simple column
                    // We cannot push down order by clauses that depend on
                    // projected computations as they would have nothing to reference.
                    if let Some(child_col) = proj_expr.expr.downcast_ref::<Column>() {
                        // Replace with the child column
                        Ok(Transformed::yes(Arc::new(child_col.clone()) as _))
                    } else {
                        // Projection involves computation, cannot push down
                        can_pushdown = false;
                        Ok(Transformed::no(expr))
                    }
                } else {
                    Ok(Transformed::no(expr))
                }
            })?;

            if !can_pushdown {
                return Ok(SortOrderPushdownResult::Unsupported);
            }

            child_order.push(PhysicalSortExpr {
                expr: transformed.data,
                options: sort_expr.options,
            });
        }

        // Recursively push down to child node
        match child.try_pushdown_sort(&child_order)? {
            SortOrderPushdownResult::Exact { inner } => {
                let new_exec =
                    replace_children_if_necessary(Arc::new(self.clone()), vec![inner])?;
                Ok(SortOrderPushdownResult::Exact { inner: new_exec })
            }
            SortOrderPushdownResult::Inexact { inner } => {
                let new_exec =
                    replace_children_if_necessary(Arc::new(self.clone()), vec![inner])?;
                Ok(SortOrderPushdownResult::Inexact { inner: new_exec })
            }
            SortOrderPushdownResult::Unsupported => {
                Ok(SortOrderPushdownResult::Unsupported)
            }
        }
    }

    fn with_preserve_order(
        &self,
        preserve_order: bool,
    ) -> Option<Arc<dyn ExecutionPlan>> {
        self.input
            .with_preserve_order(preserve_order)
            .and_then(|new_input| {
                replace_children_if_necessary(Arc::new(self.clone()), vec![new_input])
                    .ok()
            })
    }

    #[cfg(feature = "proto")]
    fn try_to_proto(
        &self,
        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
        use datafusion_proto_models::protobuf;
        let input = ctx.encode_child(self.input())?;
        let expr = ctx.encode_expressions(self.expr().iter().map(|p| &p.expr))?;
        let expr_name = self.expr().iter().map(|p| p.alias.clone()).collect();
        Ok(Some(protobuf::PhysicalPlanNode {
            physical_plan_type: Some(
                protobuf::physical_plan_node::PhysicalPlanType::Projection(Box::new(
                    protobuf::ProjectionExecNode {
                        input: Some(Box::new(input)),
                        expr,
                        expr_name,
                    },
                )),
            ),
        }))
    }
}

#[cfg(feature = "proto")]
impl ProjectionExec {
    /// Reconstruct a [`ProjectionExec`] from its protobuf representation.
    ///
    /// The exact inverse of [`ExecutionPlan::try_to_proto`]: it takes the whole
    /// [`PhysicalPlanNode`] so every plan's `try_from_proto` shares one
    /// signature. Child plans and expressions are decoded recursively via the
    /// [`ExecutionPlanDecodeCtx`].
    ///
    /// [`PhysicalPlanNode`]: datafusion_proto_models::protobuf::PhysicalPlanNode
    /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
    /// [`ExecutionPlanDecodeCtx`]: crate::proto::ExecutionPlanDecodeCtx
    pub fn try_from_proto(
        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        use datafusion_proto_models::protobuf;
        let projection = crate::expect_plan_variant!(
            node,
            protobuf::physical_plan_node::PhysicalPlanType::Projection,
            "ProjectionExec",
        );
        let input = ctx.decode_required_child(
            projection.input.as_deref(),
            "ProjectionExec",
            "input",
        )?;
        let input_schema = input.schema();
        let exprs = projection
            .expr
            .iter()
            .zip(projection.expr_name.iter())
            .map(|(expr, name)| {
                Ok(ProjectionExpr {
                    expr: ctx.decode_expr(expr, input_schema.as_ref())?,
                    alias: name.to_string(),
                })
            })
            .collect::<Result<Vec<ProjectionExpr>>>()?;
        Ok(Arc::new(ProjectionExec::try_new(exprs, input)?))
    }
}

impl ProjectionStream {
    /// Create a new projection stream
    fn new(
        projector: Projector,
        input: SendableRecordBatchStream,
        baseline_metrics: BaselineMetrics,
    ) -> Result<Self> {
        Ok(Self {
            projector,
            input,
            baseline_metrics,
        })
    }

    fn batch_project(&self, batch: &RecordBatch) -> Result<RecordBatch> {
        // Records time on drop
        let _timer = self.baseline_metrics.elapsed_compute().timer();
        self.projector.project_batch(batch)
    }
}

/// Projection iterator
struct ProjectionStream {
    projector: Projector,
    input: SendableRecordBatchStream,
    baseline_metrics: BaselineMetrics,
}

impl Stream for ProjectionStream {
    type Item = Result<RecordBatch>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let poll = self.input.poll_next_unpin(cx).map(|x| match x {
            Some(Ok(batch)) => Some(self.batch_project(&batch)),
            other => other,
        });

        self.baseline_metrics.record_poll(poll)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        // Same number of record batches
        self.input.size_hint()
    }
}

impl RecordBatchStream for ProjectionStream {
    /// Get the schema
    fn schema(&self) -> SchemaRef {
        Arc::clone(self.projector.output_schema())
    }
}

/// Trait for execution plans that can embed a projection, avoiding a separate
/// [`ProjectionExec`] wrapper.
///
/// # Empty projections
///
/// `Some(vec![])` is a valid projection that produces zero output columns while
/// preserving the correct row count. Implementors must ensure that runtime batch
/// construction still returns batches with the right number of rows even when no
/// columns are selected (e.g. for `SELECT count(1) … JOIN …`).
pub trait EmbeddedProjection: ExecutionPlan + Sized {
    fn with_projection(&self, projection: Option<Vec<usize>>) -> Result<Self>;
}

/// Some projection can't be pushed down left input or right input of hash join because filter or on need may need some columns that won't be used in later.
/// By embed those projection to hash join, we can reduce the cost of build_batch_from_indices in hash join (build_batch_from_indices need to can compute::take() for each column) and avoid unnecessary output creation.
pub fn try_embed_projection<Exec: EmbeddedProjection + 'static>(
    projection: &ProjectionExec,
    execution_plan: &Exec,
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
    // If the projection has no expressions at all (e.g., ProjectionExec: expr=[]),
    // embed an empty projection into the execution plan so it outputs zero columns.
    // This avoids allocating throwaway null arrays for build-side columns
    // when no output columns are actually needed (e.g., count(1) over a right join).
    if projection.expr().is_empty() {
        let new_execution_plan = Arc::new(execution_plan.with_projection(Some(vec![]))?);
        return Ok(Some(new_execution_plan));
    }

    // Collect all column indices from the given projection expressions.
    let projection_index = collect_column_indices(projection.expr());

    if projection_index.is_empty() {
        return Ok(None);
    };

    let columns_reduced = projection_index.len() < execution_plan.schema().fields().len();

    let new_execution_plan =
        Arc::new(execution_plan.with_projection(Some(projection_index.to_vec()))?);

    // Build projection expressions for update_expr. Zip the projection_index with the new_execution_plan output schema fields.
    let embed_project_exprs = projection_index
        .iter()
        .zip(new_execution_plan.schema().fields())
        .map(|(index, field)| ProjectionExpr {
            expr: Arc::new(Column::new(field.name(), *index)) as Arc<dyn PhysicalExpr>,
            alias: field.name().to_owned(),
        })
        .collect::<Vec<_>>();

    let mut new_projection_exprs = Vec::with_capacity(projection.expr().len());

    for proj_expr in projection.expr() {
        // update column index for projection expression since the input schema has been changed.
        let Some(expr) =
            update_expr(&proj_expr.expr, embed_project_exprs.as_slice(), false)?
        else {
            return Ok(None);
        };
        new_projection_exprs.push(ProjectionExpr {
            expr,
            alias: proj_expr.alias.clone(),
        });
    }
    // Old projection may contain some alias or expression such as `a + 1` and `CAST('true' AS BOOLEAN)`, but our projection_exprs in hash join just contain column, so we need to create the new projection to keep the original projection.
    let new_projection = Arc::new(ProjectionExec::try_new(
        new_projection_exprs,
        Arc::clone(&new_execution_plan) as _,
    )?);
    if is_projection_removable(&new_projection) {
        // Residual is identity — embedding fully absorbed the projection.
        Ok(Some(new_execution_plan))
    } else if columns_reduced {
        // Embedding reduced columns even though a residual is still needed
        // for renames or expressions — worth keeping.
        Ok(Some(new_projection))
    } else {
        // No columns eliminated and residual still needed — embedding just
        // adds an unnecessary column reorder inside the operator.
        Ok(None)
    }
}

pub struct JoinData {
    pub projected_left_child: ProjectionExec,
    pub projected_right_child: ProjectionExec,
    pub join_filter: Option<JoinFilter>,
    pub join_on: JoinOn,
}

#[deprecated(
    since = "55.0.0",
    note = "Use try_pushdown_through_join_with_column_indices instead"
)]
pub fn try_pushdown_through_join(
    projection: &ProjectionExec,
    join_left: &Arc<dyn ExecutionPlan>,
    join_right: &Arc<dyn ExecutionPlan>,
    join_on: JoinOnRef,
    schema: &SchemaRef,
    filter: Option<&JoinFilter>,
) -> Result<Option<JoinData>> {
    let left_field_count = join_left.schema().fields().len();
    let column_indices = schema
        .fields()
        .iter()
        .enumerate()
        .map(|(index, _)| {
            if index < left_field_count {
                ColumnIndex {
                    index,
                    side: JoinSide::Left,
                }
            } else {
                ColumnIndex {
                    index: index - left_field_count,
                    side: JoinSide::Right,
                }
            }
        })
        .collect::<Vec<_>>();

    try_pushdown_through_join_with_column_indices(
        projection,
        join_left,
        join_right,
        join_on,
        schema,
        filter,
        &column_indices,
    )
}

/// Attempts to move a projection below a join by mapping each join output
/// column to the child column that produced it.
///
/// `schema` is the complete output schema of the join, not either child's
/// schema. `column_indices` must contain one entry for each field in `schema`.
/// Each [`JoinSide::Left`] or [`JoinSide::Right`] entry identifies the source
/// child and uses an index relative to that child's schema.
///
/// [`JoinSide::None`] identifies a column produced by the join itself, such as
/// a mark column. If `projection` references such a column, this function
/// returns `Ok(None)` because neither child can produce it.
///
/// Returns `Ok(None)` when the projection cannot be pushed down safely.
///
/// # Errors
///
/// Returns an error if `column_indices` does not match `schema` or contains an
/// index outside the corresponding child schema.
pub fn try_pushdown_through_join_with_column_indices(
    projection: &ProjectionExec,
    join_left: &Arc<dyn ExecutionPlan>,
    join_right: &Arc<dyn ExecutionPlan>,
    join_on: JoinOnRef,
    schema: &SchemaRef,
    filter: Option<&JoinFilter>,
    column_indices: &[ColumnIndex],
) -> Result<Option<JoinData>> {
    if column_indices.len() != schema.fields().len() {
        return plan_err!(
            "Column index mapping has {} entries but join schema has {} fields",
            column_indices.len(),
            schema.fields().len()
        );
    }
    // Validate each output-to-child mapping before using it to rewrite the
    // projection. Synthetic outputs have no child index to validate.
    for (output_index, column_index) in column_indices.iter().enumerate() {
        let (side, child_field_count) = match column_index.side {
            JoinSide::Left => ("left", join_left.schema().fields().len()),
            JoinSide::Right => ("right", join_right.schema().fields().len()),
            JoinSide::None => continue,
        };
        if column_index.index >= child_field_count {
            return plan_err!(
                "Join output column {output_index} maps to {side} child column {}, but the child has {child_field_count} fields",
                column_index.index
            );
        }
    }

    // Convert projected expressions to columns. We can not proceed if this is not possible.
    let Some(projection_as_columns) = physical_to_column_exprs(projection.expr()) else {
        return Ok(None);
    };

    if projection_as_columns.len() >= schema.fields().len() {
        return Ok(None);
    }
    let mut left_proj: Vec<(Column, String)> = Vec::new();
    let mut right_proj: Vec<(Column, String)> = Vec::new();
    let mut seen_right = false;
    for (col, alias) in &projection_as_columns {
        let Some(origin) = column_indices.get(col.index()) else {
            return plan_err!(
                "Projection column {} is outside the {}-entry column index mapping",
                col.index(),
                column_indices.len()
            );
        };
        match origin.side {
            // Keep the "left block before right block" contiguity the current
            // pushdown supports; a left column after a right one is "mixed".
            JoinSide::Left => {
                if seen_right {
                    return Ok(None);
                }
                left_proj.push((Column::new(col.name(), origin.index), alias.clone()));
            }
            JoinSide::Right => {
                seen_right = true;
                right_proj.push((Column::new(col.name(), origin.index), alias.clone()));
            }
            // Synthetic column (e.g. mark): belongs to neither child.
            // Phase 2 declines; Phase 3 keeps it at the join output instead.
            JoinSide::None => return Ok(None),
        }
    }

    // Parity: neither side fully dropped.
    if left_proj.is_empty() || right_proj.is_empty() {
        return Ok(None);
    }

    // `left_proj` / `right_proj` carry *child* indices (from `column_indices`),
    // so the shared `update_join_*` helpers must use a 0 column-index offset for
    // both sides (the offset bridges child -> join-output index, which is the
    // identity here).
    let new_filter = if let Some(filter) = filter {
        match update_join_filter(&left_proj, &right_proj, filter, 0) {
            Some(updated) => Some(updated),
            None => return Ok(None),
        }
    } else {
        None
    };

    let Some(new_on) = update_join_on(&left_proj, &right_proj, join_on, 0) else {
        return Ok(None);
    };

    let (new_left, new_right) =
        new_join_children_from_groups(&left_proj, &right_proj, join_left, join_right)?;

    Ok(Some(JoinData {
        projected_left_child: new_left,
        projected_right_child: new_right,
        join_filter: new_filter,
        join_on: new_on,
    }))
}

/// This function checks if `plan` is a [`ProjectionExec`], and inspects its
/// input(s) to test whether it can push `plan` under its input(s). This function
/// will operate on the entire tree and may ultimately remove `plan` entirely
/// by leveraging source providers with built-in projection capabilities.
pub fn remove_unnecessary_projections(
    plan: Arc<dyn ExecutionPlan>,
) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
    let maybe_modified = if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        // If the projection does not cause any change on the input, we can
        // safely remove it:
        if is_projection_removable(projection) {
            return Ok(Transformed::yes(Arc::clone(projection.input())));
        }
        // If it does, check if we can push it under its child(ren):
        projection
            .input()
            .try_swapping_with_projection(projection)?
    } else {
        return Ok(Transformed::no(plan));
    };
    Ok(maybe_modified.map_or_else(|| Transformed::no(plan), Transformed::yes))
}

/// Compare the inputs and outputs of the projection. All expressions must be
/// columns without alias, and projection does not change the order of fields.
/// For example, if the input schema is `a, b`, `SELECT a, b` is removable,
/// but `SELECT b, a` and `SELECT a+1, b` and `SELECT a AS c, b` are not.
fn is_projection_removable(projection: &ProjectionExec) -> bool {
    let exprs = projection.expr();
    exprs.iter().enumerate().all(|(idx, proj_expr)| {
        let Some(col) = proj_expr.expr.downcast_ref::<Column>() else {
            return false;
        };
        col.name() == proj_expr.alias && col.index() == idx
    }) && exprs.len() == projection.input().schema().fields().len()
}

/// Given the expression set of a projection, checks if the projection causes
/// any renaming or constructs a non-`Column` physical expression.
pub fn all_alias_free_columns(exprs: &[ProjectionExpr]) -> bool {
    exprs.iter().all(|proj_expr| {
        proj_expr
            .expr
            .downcast_ref::<Column>()
            .map(|column| column.name() == proj_expr.alias)
            .unwrap_or(false)
    })
}

/// Updates a source provider's projected columns according to the given
/// projection operator's expressions. To use this function safely, one must
/// ensure that all expressions are `Column` expressions without aliases.
pub fn new_projections_for_columns(
    projection: &[ProjectionExpr],
    source: &[usize],
) -> Vec<usize> {
    projection
        .iter()
        .filter_map(|proj_expr| {
            proj_expr
                .expr
                .downcast_ref::<Column>()
                .map(|expr| source[expr.index()])
        })
        .collect()
}

/// Creates a new [`ProjectionExec`] instance with the given child plan and
/// projected expressions.
pub fn make_with_child(
    projection: &ProjectionExec,
    child: &Arc<dyn ExecutionPlan>,
) -> Result<Arc<dyn ExecutionPlan>> {
    ProjectionExec::try_new(projection.expr().to_vec(), Arc::clone(child))
        .map(|e| Arc::new(e) as _)
}

/// Returns `true` if all the expressions in the argument are `Column`s.
pub fn all_columns(exprs: &[ProjectionExpr]) -> bool {
    exprs.iter().all(|proj_expr| proj_expr.expr.is::<Column>())
}

/// Updates the given lexicographic ordering according to given projected
/// expressions using the [`update_expr`] function.
pub fn update_ordering(
    ordering: LexOrdering,
    projected_exprs: &[ProjectionExpr],
) -> Result<Option<LexOrdering>> {
    let mut updated_exprs = vec![];
    for mut sort_expr in ordering.into_iter() {
        let Some(updated_expr) = update_expr(&sort_expr.expr, projected_exprs, false)?
        else {
            return Ok(None);
        };
        sort_expr.expr = updated_expr;
        updated_exprs.push(sort_expr);
    }
    Ok(LexOrdering::new(updated_exprs))
}

/// Updates the given lexicographic requirement according to given projected
/// expressions using the [`update_expr`] function.
pub fn update_ordering_requirement(
    reqs: LexRequirement,
    projected_exprs: &[ProjectionExpr],
) -> Result<Option<LexRequirement>> {
    let mut updated_exprs = vec![];
    for mut sort_expr in reqs.into_iter() {
        let Some(updated_expr) = update_expr(&sort_expr.expr, projected_exprs, false)?
        else {
            return Ok(None);
        };
        sort_expr.expr = updated_expr;
        updated_exprs.push(sort_expr);
    }
    Ok(LexRequirement::new(updated_exprs))
}

/// Downcasts all the expressions in `exprs` to `Column`s. If any of the given
/// expressions is not a `Column`, returns `None`.
pub fn physical_to_column_exprs(
    exprs: &[ProjectionExpr],
) -> Option<Vec<(Column, String)>> {
    exprs
        .iter()
        .map(|proj_expr| {
            proj_expr
                .expr
                .downcast_ref::<Column>()
                .map(|col| (col.clone(), proj_expr.alias.clone()))
        })
        .collect()
}

/// If pushing down the projection over this join's children seems possible,
/// this function constructs the new [`ProjectionExec`]s that will come on top
/// of the original children of the join.
pub fn new_join_children(
    projection_as_columns: &[(Column, String)],
    far_right_left_col_ind: i32,
    far_left_right_col_ind: i32,
    left_child: &Arc<dyn ExecutionPlan>,
    right_child: &Arc<dyn ExecutionPlan>,
) -> Result<(ProjectionExec, ProjectionExec)> {
    let new_left = ProjectionExec::try_new(
        projection_as_columns[0..=far_right_left_col_ind as _]
            .iter()
            .map(|(col, alias)| ProjectionExpr {
                expr: Arc::new(Column::new(col.name(), col.index())) as _,
                alias: alias.clone(),
            }),
        Arc::clone(left_child),
    )?;
    let left_size = left_child.schema().fields().len() as i32;
    let new_right = ProjectionExec::try_new(
        projection_as_columns[far_left_right_col_ind as _..]
            .iter()
            .map(|(col, alias)| {
                ProjectionExpr {
                    expr: Arc::new(Column::new(
                        col.name(),
                        // Align projected expressions coming from the right
                        // table with the new right child projection:
                        (col.index() as i32 - left_size) as _,
                    )) as _,
                    alias: alias.clone(),
                }
            }),
        Arc::clone(right_child),
    )?;

    Ok((new_left, new_right))
}

/// Build the projected left and right children from side-grouped projection
/// columns whose indices are already *child*-relative (e.g. derived from a
/// join's `ColumnIndex`). Unlike [`new_join_children`], this does not infer
/// child ownership from output position, so it is safe for join schemas whose
/// output is not a plain `left ++ right` (used by the schema-aware
/// `try_pushdown_through_join_with_column_indices`).
fn new_join_children_from_groups(
    left_proj: &[(Column, String)],
    right_proj: &[(Column, String)],
    left_child: &Arc<dyn ExecutionPlan>,
    right_child: &Arc<dyn ExecutionPlan>,
) -> Result<(ProjectionExec, ProjectionExec)> {
    let build = |cols: &[(Column, String)], child: &Arc<dyn ExecutionPlan>| {
        ProjectionExec::try_new(
            cols.iter().map(|(col, alias)| ProjectionExpr {
                expr: Arc::new(Column::new(col.name(), col.index())) as _,
                alias: alias.clone(),
            }),
            Arc::clone(child),
        )
    };

    Ok((
        build(left_proj, left_child)?,
        build(right_proj, right_child)?,
    ))
}

/// Checks three conditions for pushing a projection down through a join:
/// - Projection must narrow the join output schema.
/// - Columns coming from left/right tables must be collected at the left/right
///   sides of the output table.
/// - Left or right table is not lost after the projection.
pub fn join_allows_pushdown(
    projection_as_columns: &[(Column, String)],
    join_schema: &SchemaRef,
    far_right_left_col_ind: i32,
    far_left_right_col_ind: i32,
) -> bool {
    // Projection must narrow the join output:
    projection_as_columns.len() < join_schema.fields().len()
    // Are the columns from different tables mixed?
    && (far_right_left_col_ind + 1 == far_left_right_col_ind)
    // Left or right table is not lost after the projection.
    && far_right_left_col_ind >= 0
    && far_left_right_col_ind < projection_as_columns.len() as i32
}

/// Returns the last index before encountering a column coming from the right table when traveling
/// through the projection from left to right, and the last index before encountering a column
/// coming from the left table when traveling through the projection from right to left.
/// If there is no column in the projection coming from the left side, it returns (-1, ...),
/// if there is no column in the projection coming from the right side, it returns (..., projection length).
pub fn join_table_borders(
    left_table_column_count: usize,
    projection_as_columns: &[(Column, String)],
) -> (i32, i32) {
    let far_right_left_col_ind = projection_as_columns
        .iter()
        .enumerate()
        .take_while(|(_, (projection_column, _))| {
            projection_column.index() < left_table_column_count
        })
        .last()
        .map(|(index, _)| index as i32)
        .unwrap_or(-1);

    let far_left_right_col_ind = projection_as_columns
        .iter()
        .enumerate()
        .rev()
        .take_while(|(_, (projection_column, _))| {
            projection_column.index() >= left_table_column_count
        })
        .last()
        .map(|(index, _)| index as i32)
        .unwrap_or(projection_as_columns.len() as i32);

    (far_right_left_col_ind, far_left_right_col_ind)
}

/// Tries to update the equi-join `Column`'s of a join as if the input of
/// the join was replaced by a projection.
pub fn update_join_on(
    proj_left_exprs: &[(Column, String)],
    proj_right_exprs: &[(Column, String)],
    hash_join_on: &[(PhysicalExprRef, PhysicalExprRef)],
    left_field_size: usize,
) -> Option<Vec<(PhysicalExprRef, PhysicalExprRef)>> {
    let (left_idx, right_idx): (Vec<_>, Vec<_>) = hash_join_on
        .iter()
        .map(|(left, right)| (left, right))
        .unzip();

    let new_left = new_columns_for_join_on(&left_idx, proj_left_exprs, 0)?;
    let new_right =
        new_columns_for_join_on(&right_idx, proj_right_exprs, left_field_size)?;
    Some(new_left.into_iter().zip(new_right).collect())
}

/// Tries to update the column indices of a [`JoinFilter`] as if the input of
/// the join was replaced by a projection.
pub fn update_join_filter(
    projection_left_exprs: &[(Column, String)],
    projection_right_exprs: &[(Column, String)],
    join_filter: &JoinFilter,
    left_field_size: usize,
) -> Option<JoinFilter> {
    let mut new_left_indices = new_indices_for_join_filter(
        join_filter,
        JoinSide::Left,
        projection_left_exprs,
        0,
    )
    .into_iter();
    let mut new_right_indices = new_indices_for_join_filter(
        join_filter,
        JoinSide::Right,
        projection_right_exprs,
        left_field_size,
    )
    .into_iter();

    // Check if all columns match:
    (new_right_indices.len() + new_left_indices.len()
        == join_filter.column_indices().len())
    .then(|| {
        JoinFilter::new(
            Arc::clone(join_filter.expression()),
            join_filter
                .column_indices()
                .iter()
                .map(|col_idx| ColumnIndex {
                    index: if col_idx.side == JoinSide::Left {
                        new_left_indices.next().unwrap()
                    } else {
                        new_right_indices.next().unwrap()
                    },
                    side: col_idx.side,
                })
                .collect(),
            Arc::clone(join_filter.schema()),
        )
    })
}

/// Collapse a chain of consecutive [`ProjectionExec`]s into one. Returns
/// `None` if nothing could be merged.
fn try_collapse_projection_chain(
    outer: &ProjectionExec,
) -> Result<Option<Arc<dyn ExecutionPlan>>> {
    let mut current_exprs: Vec<ProjectionExpr> = outer.expr().to_vec();
    let mut current_input: Arc<dyn ExecutionPlan> = Arc::clone(outer.input());
    let mut column_ref_map: HashMap<Column, usize> = HashMap::new();
    let mut collapsed_any = false;

    'outer: while let Some(inner_proj) = current_input.downcast_ref::<ProjectionExec>() {
        // Collect the column references usage in the outer projection.
        column_ref_map.clear();
        for proj_expr in &current_exprs {
            proj_expr.expr.apply(|expr| {
                if let Some(column) = expr.downcast_ref::<Column>() {
                    *column_ref_map.entry(column.clone()).or_default() += 1;
                }
                Ok(TreeNodeRecursion::Continue)
            })?;
        }
        let inner_exprs = inner_proj.expr();
        // Merging these projections is not beneficial, e.g
        // If an expression is not trivial (KeepInPlace) and it is referred more than 1, unifies projections will be
        // beneficial as caching mechanism for non-trivial computations.
        // See discussion in: https://github.com/apache/datafusion/issues/8296
        let blocked = column_ref_map.iter().any(|(column, count)| {
            *count > 1
                && !inner_exprs[column.index()]
                    .expr
                    .placement()
                    .should_push_to_leaves()
        });
        if blocked {
            break;
        }

        let mut new_phys: Vec<Arc<dyn PhysicalExpr>> =
            Vec::with_capacity(current_exprs.len());
        for proj_expr in &current_exprs {
            // If there is no match in the input projection, we cannot unify these
            // projections. This case will arise if the projection expression contains
            // a `PhysicalExpr` variant `update_expr` doesn't support.
            let Some(expr) = update_expr(&proj_expr.expr, inner_exprs, true)? else {
                break 'outer;
            };
            new_phys.push(expr);
        }
        for (proj_expr, expr) in current_exprs.iter_mut().zip(new_phys) {
            proj_expr.expr = expr;
        }
        current_input = Arc::clone(inner_proj.input());
        collapsed_any = true;
    }

    if !collapsed_any {
        return Ok(None);
    }

    // To unify 3 or more sequential projections:
    let unified: Arc<dyn ExecutionPlan> =
        Arc::new(ProjectionExec::try_new(current_exprs, current_input)?);
    remove_unnecessary_projections(unified).data().map(Some)
}

/// Collect all column indices from the given projection expressions.
fn collect_column_indices(exprs: &[ProjectionExpr]) -> Vec<usize> {
    // Collect column indices in a deterministic order that preserves the
    // projection's column ordering. For simple Column expressions, we use
    // the column index directly. For complex expressions, we walk the
    // expression tree to collect column references in traversal order.
    // This allows the embedded projection to match the desired output
    // column order, avoiding a residual ProjectionExec.
    let mut seen = std::collections::HashSet::new();
    let mut indices = Vec::new();
    for proj_expr in exprs {
        if let Some(col) = proj_expr.expr.downcast_ref::<Column>() {
            // Simple column reference: preserve projection order.
            if seen.insert(col.index()) {
                indices.push(col.index());
            }
        } else {
            // Complex expression: collect all referenced columns in
            // expression tree traversal order (deterministic) to preserve
            // the natural ordering of column references.
            proj_expr
                .expr
                .apply(|expr| {
                    if let Some(col) = expr.downcast_ref::<Column>()
                        && seen.insert(col.index())
                    {
                        indices.push(col.index());
                    }
                    Ok(TreeNodeRecursion::Continue)
                })
                .expect("closure always returns OK");
        }
    }
    indices
}

/// This function determines and returns a vector of indices representing the
/// positions of columns in `projection_exprs` that are involved in `join_filter`,
/// and correspond to a particular side (`join_side`) of the join operation.
///
/// Notes: Column indices in the projection expressions are based on the join schema,
/// whereas the join filter is based on the join child schema. `column_index_offset`
/// represents the offset between them.
fn new_indices_for_join_filter(
    join_filter: &JoinFilter,
    join_side: JoinSide,
    projection_exprs: &[(Column, String)],
    column_index_offset: usize,
) -> Vec<usize> {
    join_filter
        .column_indices()
        .iter()
        .filter(|col_idx| col_idx.side == join_side)
        .filter_map(|col_idx| {
            projection_exprs
                .iter()
                .position(|(col, _)| col_idx.index + column_index_offset == col.index())
        })
        .collect()
}

/// This function generates a new set of columns to be used in a hash join
/// operation based on a set of equi-join conditions (`hash_join_on`) and a
/// list of projection expressions (`projection_exprs`).
///
/// Notes: Column indices in the projection expressions are based on the join schema,
/// whereas the join on expressions are based on the join child schema. `column_index_offset`
/// represents the offset between them.
fn new_columns_for_join_on(
    hash_join_on: &[&PhysicalExprRef],
    projection_exprs: &[(Column, String)],
    column_index_offset: usize,
) -> Option<Vec<PhysicalExprRef>> {
    let new_columns = hash_join_on
        .iter()
        .filter_map(|on| {
            // Rewrite all columns in `on`
            Arc::clone(*on)
                .transform(|expr| {
                    if let Some(column) = expr.downcast_ref::<Column>() {
                        // Find the column in the projection expressions
                        let new_column = projection_exprs
                            .iter()
                            .enumerate()
                            .find(|(_, (proj_column, _))| {
                                column.name() == proj_column.name()
                                    && column.index() + column_index_offset
                                        == proj_column.index()
                            })
                            .map(|(index, (_, alias))| Column::new(alias, index));
                        if let Some(new_column) = new_column {
                            Ok(Transformed::yes(Arc::new(new_column)))
                        } else {
                            // If the column is not found in the projection expressions,
                            // it means that the column is not projected. In this case,
                            // we cannot push the projection down.
                            internal_err!(
                                "Column {:?} not found in projection expressions",
                                column
                            )
                        }
                    } else {
                        Ok(Transformed::no(expr))
                    }
                })
                .data()
                .ok()
        })
        .collect::<Vec<_>>();
    (new_columns.len() == hash_join_on.len()).then_some(new_columns)
}

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

    use crate::common::collect;
    use crate::empty::EmptyExec;

    use crate::filter_pushdown::PushedDown;
    use crate::statistics::{StatisticsArgs, StatisticsContext};
    use crate::test;
    use crate::test::exec::StatisticsExec;

    use arrow::datatypes::{DataType, Field, Schema};
    use datafusion_common::ScalarValue;
    use datafusion_common::stats::{ColumnStatistics, Precision, Statistics};

    use datafusion_expr::Operator;
    use datafusion_physical_expr::expressions::{
        BinaryExpr, Column, DynamicFilterPhysicalExpr, Literal, binary, col, lit,
    };

    #[test]
    fn test_try_new_with_schema_metadata_only_replaces_metadata() -> Result<()> {
        let input_schema = Arc::new(Schema::new(vec![Field::new(
            "input",
            DataType::Int32,
            false,
        )]));
        let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(input_schema));
        let field_metadata =
            HashMap::from([("field-key".to_string(), "field-value".to_string())]);
        let schema_metadata =
            HashMap::from([("schema-key".to_string(), "schema-value".to_string())]);
        let metadata_schema = Schema::new_with_metadata(
            vec![
                Field::new("ignored", DataType::Utf8, true)
                    .with_metadata(field_metadata.clone()),
            ],
            schema_metadata.clone(),
        );

        let projection = ProjectionExec::try_new_with_schema_metadata(
            [ProjectionExpr {
                expr: Arc::new(Column::new("input", 0)),
                alias: "output".to_string(),
            }],
            input,
            &metadata_schema,
        )?;

        let expected_schema = Arc::new(Schema::new_with_metadata(
            vec![
                Field::new("output", DataType::Int32, false)
                    .with_metadata(field_metadata),
            ],
            schema_metadata,
        ));
        assert_eq!(projection.schema(), expected_schema);
        Ok(())
    }

    #[test]
    fn test_collect_column_indices() -> Result<()> {
        let expr = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("b", 7)),
            Operator::Minus,
            Arc::new(BinaryExpr::new(
                Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
                Operator::Plus,
                Arc::new(Column::new("a", 1)),
            )),
        ));
        let column_indices = collect_column_indices(&[ProjectionExpr {
            expr,
            alias: "b-(1+a)".to_string(),
        }]);
        // Tree traversal order: b@7 is visited before a@1
        assert_eq!(column_indices, vec![7, 1]);
        Ok(())
    }

    #[test]
    fn test_try_pushdown_through_join_validates_column_indices() -> Result<()> {
        let child_schema =
            Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, false)]));
        let left: Arc<dyn ExecutionPlan> =
            Arc::new(EmptyExec::new(Arc::clone(&child_schema)));
        let right: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(child_schema));
        let join_schema = Arc::new(Schema::new(vec![
            Field::new("left_i", DataType::Int32, false),
            Field::new("right_i", DataType::Int32, false),
        ]));
        let join: Arc<dyn ExecutionPlan> =
            Arc::new(EmptyExec::new(Arc::clone(&join_schema)));
        let projection = ProjectionExec::try_new(
            vec![ProjectionExpr {
                expr: Arc::new(Column::new("left_i", 0)),
                alias: "left_i".to_string(),
            }],
            join,
        )?;

        let Err(error) = try_pushdown_through_join_with_column_indices(
            &projection,
            &left,
            &right,
            &[],
            &join_schema,
            None,
            &[],
        ) else {
            panic!("expected a mismatched mapping length to return an error");
        };
        assert!(
            error.to_string().contains(
                "Column index mapping has 0 entries but join schema has 2 fields"
            )
        );

        let invalid_child_index = [
            ColumnIndex {
                index: 1,
                side: JoinSide::Left,
            },
            ColumnIndex {
                index: 0,
                side: JoinSide::Right,
            },
        ];
        let Err(error) = try_pushdown_through_join_with_column_indices(
            &projection,
            &left,
            &right,
            &[],
            &join_schema,
            None,
            &invalid_child_index,
        ) else {
            panic!("expected an invalid child index to return an error");
        };
        assert!(error.to_string().contains(
            "Join output column 0 maps to left child column 1, but the child has 1 fields"
        ));

        let wider_join_schema = Arc::new(Schema::new(vec![
            Field::new("left_i", DataType::Int32, false),
            Field::new("right_i", DataType::Int32, false),
            Field::new("extra", DataType::Int32, false),
        ]));
        let wider_join: Arc<dyn ExecutionPlan> =
            Arc::new(EmptyExec::new(wider_join_schema));
        let out_of_mapping_projection = ProjectionExec::try_new(
            vec![ProjectionExpr {
                expr: Arc::new(Column::new("extra", 2)),
                alias: "extra".to_string(),
            }],
            wider_join,
        )?;
        let valid_child_indices = [
            ColumnIndex {
                index: 0,
                side: JoinSide::Left,
            },
            ColumnIndex {
                index: 0,
                side: JoinSide::Right,
            },
        ];
        let Err(error) = try_pushdown_through_join_with_column_indices(
            &out_of_mapping_projection,
            &left,
            &right,
            &[],
            &join_schema,
            None,
            &valid_child_indices,
        ) else {
            panic!("expected an out-of-mapping projection to return an error");
        };
        assert!(
            error.to_string().contains(
                "Projection column 2 is outside the 2-entry column index mapping"
            )
        );

        Ok(())
    }

    #[test]
    fn test_join_table_borders() -> Result<()> {
        let projections = vec![
            (Column::new("b", 1), "b".to_owned()),
            (Column::new("c", 2), "c".to_owned()),
            (Column::new("e", 4), "e".to_owned()),
            (Column::new("d", 3), "d".to_owned()),
            (Column::new("c", 2), "c".to_owned()),
            (Column::new("f", 5), "f".to_owned()),
            (Column::new("h", 7), "h".to_owned()),
            (Column::new("g", 6), "g".to_owned()),
        ];
        let left_table_column_count = 5;
        assert_eq!(
            join_table_borders(left_table_column_count, &projections),
            (4, 5)
        );

        let left_table_column_count = 8;
        assert_eq!(
            join_table_borders(left_table_column_count, &projections),
            (7, 8)
        );

        let left_table_column_count = 1;
        assert_eq!(
            join_table_borders(left_table_column_count, &projections),
            (-1, 0)
        );

        let projections = vec![
            (Column::new("a", 0), "a".to_owned()),
            (Column::new("b", 1), "b".to_owned()),
            (Column::new("d", 3), "d".to_owned()),
            (Column::new("g", 6), "g".to_owned()),
            (Column::new("e", 4), "e".to_owned()),
            (Column::new("f", 5), "f".to_owned()),
            (Column::new("e", 4), "e".to_owned()),
            (Column::new("h", 7), "h".to_owned()),
        ];
        let left_table_column_count = 5;
        assert_eq!(
            join_table_borders(left_table_column_count, &projections),
            (2, 7)
        );

        let left_table_column_count = 7;
        assert_eq!(
            join_table_borders(left_table_column_count, &projections),
            (6, 7)
        );

        Ok(())
    }

    #[tokio::test]
    async fn project_no_column() -> Result<()> {
        let task_ctx = Arc::new(TaskContext::default());

        let exec = test::scan_partitioned(1);
        let expected = collect(exec.execute(0, Arc::clone(&task_ctx))?).await?;

        let projection = ProjectionExec::try_new(vec![] as Vec<ProjectionExpr>, exec)?;
        let stream = projection.execute(0, Arc::clone(&task_ctx))?;
        let output = collect(stream).await?;
        assert_eq!(output.len(), expected.len());

        Ok(())
    }

    #[tokio::test]
    async fn project_old_syntax() {
        let exec = test::scan_partitioned(1);
        let schema = exec.schema();
        let expr = col("i", &schema).unwrap();
        ProjectionExec::try_new(
            vec![
                // use From impl of ProjectionExpr to create ProjectionExpr
                // to test old syntax
                (expr, "c".to_string()),
            ],
            exec,
        )
        // expect this to succeed
        .unwrap();
    }

    #[test]
    fn test_projection_statistics_uses_input_schema() {
        let input_schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
            Field::new("c", DataType::Int32, false),
            Field::new("d", DataType::Int32, false),
            Field::new("e", DataType::Int32, false),
            Field::new("f", DataType::Int32, false),
        ]);

        let input_statistics = Statistics {
            num_rows: Precision::Exact(10),
            column_statistics: vec![
                ColumnStatistics {
                    min_value: Precision::Exact(ScalarValue::Int32(Some(1))),
                    max_value: Precision::Exact(ScalarValue::Int32(Some(100))),
                    ..Default::default()
                },
                ColumnStatistics {
                    min_value: Precision::Exact(ScalarValue::Int32(Some(5))),
                    max_value: Precision::Exact(ScalarValue::Int32(Some(50))),
                    ..Default::default()
                },
                ColumnStatistics {
                    min_value: Precision::Exact(ScalarValue::Int32(Some(10))),
                    max_value: Precision::Exact(ScalarValue::Int32(Some(40))),
                    ..Default::default()
                },
                ColumnStatistics {
                    min_value: Precision::Exact(ScalarValue::Int32(Some(20))),
                    max_value: Precision::Exact(ScalarValue::Int32(Some(30))),
                    ..Default::default()
                },
                ColumnStatistics {
                    min_value: Precision::Exact(ScalarValue::Int32(Some(21))),
                    max_value: Precision::Exact(ScalarValue::Int32(Some(29))),
                    ..Default::default()
                },
                ColumnStatistics {
                    min_value: Precision::Exact(ScalarValue::Int32(Some(24))),
                    max_value: Precision::Exact(ScalarValue::Int32(Some(26))),
                    ..Default::default()
                },
            ],
            ..Default::default()
        };

        let input = Arc::new(StatisticsExec::new(input_statistics, input_schema));

        // Create projection expressions that reference columns from the input schema and the length
        // of output schema columns < input schema columns and hence if we use the last few columns
        // from the input schema in the expressions here, bounds_check would fail on them if output
        // schema is supplied to the partitions_statistics method.
        let exprs: Vec<ProjectionExpr> = vec![
            ProjectionExpr {
                expr: Arc::new(Column::new("c", 2)) as Arc<dyn PhysicalExpr>,
                alias: "c_renamed".to_string(),
            },
            ProjectionExpr {
                expr: Arc::new(BinaryExpr::new(
                    Arc::new(Column::new("e", 4)),
                    Operator::Plus,
                    Arc::new(Column::new("f", 5)),
                )) as Arc<dyn PhysicalExpr>,
                alias: "e_plus_f".to_string(),
            },
        ];

        let projection = ProjectionExec::try_new(exprs, input).unwrap();

        let stats = StatisticsContext::new()
            .compute(&projection, &StatisticsArgs::new())
            .unwrap();

        assert_eq!(stats.num_rows, Precision::Exact(10));
        assert_eq!(
            stats.column_statistics.len(),
            2,
            "Expected 2 columns in projection statistics"
        );
        assert!(stats.total_byte_size.is_exact().unwrap_or(false));
    }

    #[test]
    fn test_filter_pushdown_with_alias() -> Result<()> {
        let input_schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let input = Arc::new(StatisticsExec::new(
            Statistics::new_unknown(&input_schema),
            input_schema.clone(),
        ));

        // project "a" as "b"
        let projection = ProjectionExec::try_new(
            vec![ProjectionExpr {
                expr: Arc::new(Column::new("a", 0)),
                alias: "b".to_string(),
            }],
            input,
        )?;

        // filter "b > 5"
        let filter = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("b", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![filter],
            &ConfigOptions::default(),
        )?;

        // Should be converted to "a > 5"
        // "a" is index 0 in input
        let expected_filter = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("a", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        assert_eq!(description.self_filters(), vec![vec![]]);
        let pushed_filters = &description.parent_filters()[0];
        assert_eq!(
            format!("{}", pushed_filters[0].predicate),
            format!("{}", expected_filter)
        );
        // Verify the predicate was actually pushed down
        assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes));

        Ok(())
    }

    #[test]
    fn test_filter_pushdown_with_multiple_aliases() -> Result<()> {
        let input_schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let input = Arc::new(StatisticsExec::new(
            Statistics {
                column_statistics: vec![Default::default(); input_schema.fields().len()],
                ..Default::default()
            },
            input_schema.clone(),
        ));

        // project "a" as "x", "b" as "y"
        let projection = ProjectionExec::try_new(
            vec![
                ProjectionExpr {
                    expr: Arc::new(Column::new("a", 0)),
                    alias: "x".to_string(),
                },
                ProjectionExpr {
                    expr: Arc::new(Column::new("b", 1)),
                    alias: "y".to_string(),
                },
            ],
            input,
        )?;

        // filter "x > 5"
        let filter1 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        // filter "y < 10"
        let filter2 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("y", 1)),
            Operator::Lt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![filter1, filter2],
            &ConfigOptions::default(),
        )?;

        // Should be converted to "a > 5" and "b < 10"
        let expected_filter1 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("a", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        let expected_filter2 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("b", 1)),
            Operator::Lt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;

        let pushed_filters = &description.parent_filters()[0];
        assert_eq!(pushed_filters.len(), 2);
        // Note: The order of filters is preserved
        assert_eq!(
            format!("{}", pushed_filters[0].predicate),
            format!("{}", expected_filter1)
        );
        assert_eq!(
            format!("{}", pushed_filters[1].predicate),
            format!("{}", expected_filter2)
        );
        // Verify the predicates were actually pushed down
        assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes));
        assert!(matches!(pushed_filters[1].discriminant, PushedDown::Yes));

        Ok(())
    }

    #[test]
    fn test_filter_pushdown_with_swapped_aliases() -> Result<()> {
        let input_schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let input = Arc::new(StatisticsExec::new(
            Statistics {
                column_statistics: vec![Default::default(); input_schema.fields().len()],
                ..Default::default()
            },
            input_schema.clone(),
        ));

        // project "a" as "b", "b" as "a"
        let projection = ProjectionExec::try_new(
            vec![
                ProjectionExpr {
                    expr: Arc::new(Column::new("a", 0)),
                    alias: "b".to_string(),
                },
                ProjectionExpr {
                    expr: Arc::new(Column::new("b", 1)),
                    alias: "a".to_string(),
                },
            ],
            input,
        )?;

        // filter "b > 5" (output column 0, which is "a" in input)
        let filter1 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("b", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        // filter "a < 10" (output column 1, which is "b" in input)
        let filter2 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("a", 1)),
            Operator::Lt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![filter1, filter2],
            &ConfigOptions::default(),
        )?;

        let pushed_filters = &description.parent_filters()[0];
        assert_eq!(pushed_filters.len(), 2);

        // "b" (output index 0) -> "a" (input index 0)
        let expected_filter1 = "a@0 > 5";
        // "a" (output index 1) -> "b" (input index 1)
        let expected_filter2 = "b@1 < 10";

        assert_eq!(format!("{}", pushed_filters[0].predicate), expected_filter1);
        assert_eq!(format!("{}", pushed_filters[1].predicate), expected_filter2);
        // Verify the predicates were actually pushed down
        assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes));
        assert!(matches!(pushed_filters[1].discriminant, PushedDown::Yes));

        Ok(())
    }

    #[test]
    fn test_filter_pushdown_with_mixed_columns() -> Result<()> {
        let input_schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let input = Arc::new(StatisticsExec::new(
            Statistics {
                column_statistics: vec![Default::default(); input_schema.fields().len()],
                ..Default::default()
            },
            input_schema.clone(),
        ));

        // project "a" as "x", "b" as "b" (pass through)
        let projection = ProjectionExec::try_new(
            vec![
                ProjectionExpr {
                    expr: Arc::new(Column::new("a", 0)),
                    alias: "x".to_string(),
                },
                ProjectionExpr {
                    expr: Arc::new(Column::new("b", 1)),
                    alias: "b".to_string(),
                },
            ],
            input,
        )?;

        // filter "x > 5"
        let filter1 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("x", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        // filter "b < 10" (using output index 1 which corresponds to 'b')
        let filter2 = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("b", 1)),
            Operator::Lt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![filter1, filter2],
            &ConfigOptions::default(),
        )?;

        let pushed_filters = &description.parent_filters()[0];
        assert_eq!(pushed_filters.len(), 2);
        // "x" -> "a" (index 0)
        let expected_filter1 = "a@0 > 5";
        // "b" -> "b" (index 1)
        let expected_filter2 = "b@1 < 10";

        assert_eq!(format!("{}", pushed_filters[0].predicate), expected_filter1);
        assert_eq!(format!("{}", pushed_filters[1].predicate), expected_filter2);
        // Verify the predicates were actually pushed down
        assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes));
        assert!(matches!(pushed_filters[1].discriminant, PushedDown::Yes));

        Ok(())
    }

    #[test]
    fn test_filter_pushdown_with_complex_expression() -> Result<()> {
        let input_schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let input = Arc::new(StatisticsExec::new(
            Statistics {
                column_statistics: vec![Default::default(); input_schema.fields().len()],
                ..Default::default()
            },
            input_schema.clone(),
        ));

        // project "a + 1" as "z"
        let projection = ProjectionExec::try_new(
            vec![ProjectionExpr {
                expr: Arc::new(BinaryExpr::new(
                    Arc::new(Column::new("a", 0)),
                    Operator::Plus,
                    Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
                )),
                alias: "z".to_string(),
            }],
            input,
        )?;

        // filter "z > 10"
        let filter = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("z", 0)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
        )) as Arc<dyn PhysicalExpr>;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![filter],
            &ConfigOptions::default(),
        )?;

        // expand to `a + 1 > 10`
        let pushed_filters = &description.parent_filters()[0];
        assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes));
        assert_eq!(format!("{}", pushed_filters[0].predicate), "a@0 + 1 > 10");

        Ok(())
    }

    #[test]
    fn test_filter_pushdown_with_unknown_column() -> Result<()> {
        let input_schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let input = Arc::new(StatisticsExec::new(
            Statistics {
                column_statistics: vec![Default::default(); input_schema.fields().len()],
                ..Default::default()
            },
            input_schema.clone(),
        ));

        // project "a" as "a"
        let projection = ProjectionExec::try_new(
            vec![ProjectionExpr {
                expr: Arc::new(Column::new("a", 0)),
                alias: "a".to_string(),
            }],
            input,
        )?;

        // filter "unknown_col > 5" - using a column name that doesn't exist in projection output
        // Column constructor: name, index. Index 1 doesn't exist.
        let filter = Arc::new(BinaryExpr::new(
            Arc::new(Column::new("unknown_col", 1)),
            Operator::Gt,
            Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
        )) as Arc<dyn PhysicalExpr>;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![filter],
            &ConfigOptions::default(),
        )?;

        let pushed_filters = &description.parent_filters()[0];
        assert!(matches!(pushed_filters[0].discriminant, PushedDown::No));
        // The column shouldn't be found in the alias map, so it remains unchanged with its index
        assert_eq!(
            format!("{}", pushed_filters[0].predicate),
            "unknown_col@1 > 5"
        );

        Ok(())
    }

    /// Basic test for `DynamicFilterPhysicalExpr` can correctly update its child expression
    /// i.e. starting with lit(true) and after update it becomes `a > 5`
    /// with projection [b - 1 as a], the pushed down filter should be `b - 1 > 5`
    #[test]
    fn test_basic_dyn_filter_projection_pushdown_update_child() -> Result<()> {
        let input_schema =
            Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, false)]));

        let input = Arc::new(StatisticsExec::new(
            Statistics {
                column_statistics: vec![Default::default(); input_schema.fields().len()],
                ..Default::default()
            },
            input_schema.as_ref().clone(),
        ));

        // project "b" - 1 as "a"
        let projection = ProjectionExec::try_new(
            vec![ProjectionExpr {
                expr: binary(
                    Arc::new(Column::new("b", 0)),
                    Operator::Minus,
                    lit(1),
                    &input_schema,
                )
                .unwrap(),
                alias: "a".to_string(),
            }],
            input,
        )?;

        // simulate projection's parent create a dynamic filter on "a"
        let projected_schema = projection.schema();
        let col_a = col("a", &projected_schema)?;
        let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
            vec![Arc::clone(&col_a)],
            lit(true),
        ));
        // Initial state should be lit(true)
        let current = dynamic_filter.current()?;
        assert_eq!(format!("{current}"), "true");

        let dyn_phy_expr: Arc<dyn PhysicalExpr> = Arc::clone(&dynamic_filter) as _;

        let description = projection.gather_filters_for_pushdown(
            FilterPushdownPhase::Post,
            vec![dyn_phy_expr],
            &ConfigOptions::default(),
        )?;

        let pushed_filters = &description.parent_filters()[0][0];

        // Check currently pushed_filters is lit(true)
        assert_eq!(
            format!("{}", pushed_filters.predicate),
            "DynamicFilter [ empty ]"
        );

        // Update to a > 5 (after projection, b is now called a)
        let new_expr =
            Arc::new(BinaryExpr::new(Arc::clone(&col_a), Operator::Gt, lit(5i32)));
        dynamic_filter.update(new_expr)?;

        // Now it should be a > 5
        let current = dynamic_filter.current()?;
        assert_eq!(format!("{current}"), "a@0 > 5");

        // Check currently pushed_filters is b - 1 > 5 (because b - 1 is projected as a)
        assert_eq!(
            format!("{}", pushed_filters.predicate),
            "DynamicFilter [ b@0 - 1 > 5 ]"
        );

        Ok(())
    }
}