datafusion_sql/unparser/

plan.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.

use super::{
    Unparser,
    ast::{
        BuilderError, DerivedRelationBuilder, QueryBuilder, RelationBuilder,
        SelectBuilder, TableRelationBuilder, TableWithJoinsBuilder,
    },
    rewrite::{
        TableAliasRewriter, inject_column_aliases_into_subquery, normalize_union_schema,
        rewrite_plan_for_sort_on_non_projected_fields,
        subquery_alias_inner_query_and_columns,
    },
    utils::{
        find_agg_node_within_select, find_unnest_node_within_select,
        find_window_nodes_within_select, try_transform_to_simple_table_scan_with_filters,
        unproject_sort_expr, unproject_unnest_expr,
        unproject_unnest_expr_as_flatten_value, unproject_window_exprs,
    },
};
use crate::unparser::extension_unparser::{
    UnparseToStatementResult, UnparseWithinStatementResult,
};
use crate::unparser::utils::{find_unnest_node_until_relation, unproject_agg_exprs};
use crate::unparser::{
    ast::FlattenRelationBuilder, ast::UnnestRelationBuilder, rewrite::rewrite_qualify,
};
use crate::utils::UNNEST_PLACEHOLDER;
use datafusion_common::{
    Column, DFSchema, DataFusionError, Result, ScalarValue, TableReference,
    assert_or_internal_err, internal_datafusion_err, internal_err, not_impl_err,
    tree_node::{Transformed, TransformedResult, TreeNode, TreeNodeRecursion},
    utils::combine_limit,
};
use datafusion_expr::expr::{OUTER_REFERENCE_COLUMN_PREFIX, UNNEST_COLUMN_PREFIX};
use datafusion_expr::{
    Aggregate, BinaryExpr, Distinct, Expr, FetchType, JoinConstraint, JoinType,
    LogicalPlan, LogicalPlanBuilder, Operator, Projection, SkipType, Sort, SortExpr,
    TableScan, Unnest, UserDefinedLogicalNode, Window, expr::Alias,
};
use sqlparser::ast::{self, Ident, OrderByKind, SetExpr, TableAliasColumnDef};
use std::{sync::Arc, vec};

/// Convert a DataFusion [`LogicalPlan`] to [`ast::Statement`]
///
/// This function is the opposite of [`SqlToRel::sql_statement_to_plan`] and can
/// be used to, among other things, to convert `LogicalPlan`s to SQL strings.
///
/// # Errors
///
/// This function returns an error if the plan cannot be converted to SQL.
///
/// # See Also
///
/// * [`expr_to_sql`] for converting [`Expr`], a single expression to SQL
///
/// # Example
/// ```
/// use arrow::datatypes::{DataType, Field, Schema};
/// use datafusion_expr::{col, logical_plan::table_scan};
/// use datafusion_sql::unparser::plan_to_sql;
/// let schema = Schema::new(vec![
///     Field::new("id", DataType::Utf8, false),
///     Field::new("value", DataType::Utf8, false),
/// ]);
/// // Scan 'table' and select columns 'id' and 'value'
/// let plan = table_scan(Some("table"), &schema, None)
///     .unwrap()
///     .project(vec![col("id"), col("value")])
///     .unwrap()
///     .build()
///     .unwrap();
/// // convert to AST
/// let sql = plan_to_sql(&plan).unwrap();
/// // use the Display impl to convert to SQL text
/// assert_eq!(
///     sql.to_string(),
///     "SELECT \"table\".id, \"table\".\"value\" FROM \"table\""
/// )
/// ```
///
/// [`SqlToRel::sql_statement_to_plan`]: crate::planner::SqlToRel::sql_statement_to_plan
/// [`expr_to_sql`]: crate::unparser::expr_to_sql
pub fn plan_to_sql(plan: &LogicalPlan) -> Result<ast::Statement> {
    let unparser = Unparser::default();
    unparser.plan_to_sql(plan)
}

/// Aggregate-expression scope for one rendered SELECT block.
///
/// When an aggregate's input is itself emitted as a derived subquery (a
/// projection sits between the aggregate and its relation), the input columns
/// are only reachable by that derived table's output names. Base-table
/// qualifiers like `t.col` name a relation that is out of scope above the
/// boundary, so emitting them produces SQL a strict engine rejects.
///
/// Every clause that renders an aggregate expression (SELECT / GROUP BY /
/// HAVING / QUALIFY / ORDER BY) has to apply the same rule. Detect the
/// boundary once here and reuse it, so the clauses can't drift apart (which is
/// how earlier fixes left some clauses correct and others not).
struct UnparserAggScope<'a> {
    agg: &'a Aggregate,
    /// `agg.input` renders as a derived projection, so out-of-scope qualifiers
    /// must be stripped from expressions in this scope.
    input_is_derived_projection: bool,
}

impl<'a> UnparserAggScope<'a> {
    fn new(agg: &'a Aggregate) -> Self {
        Self {
            agg,
            input_is_derived_projection: Unparser::contains_projection_before_relation(
                agg.input.as_ref(),
            ),
        }
    }

    /// Prepare a projected column or predicate that still references the
    /// aggregate by its output columns: unproject it back onto the aggregate
    /// (and `windows`) expressions, then normalize it for this scope.
    fn prepare(&self, expr: Expr, windows: Option<&[&Window]>) -> Result<Expr> {
        self.normalize(unproject_agg_exprs(expr, self.agg, windows)?)
    }

    /// Normalize an expression that is already in aggregate form (group / aggr
    /// exprs, or an unprojected sort expr): strip the qualifiers that fall out
    /// of scope once the input is a derived projection. No-op otherwise.
    fn normalize(&self, expr: Expr) -> Result<Expr> {
        if self.input_is_derived_projection {
            Unparser::strip_column_qualifiers_for_schema(
                expr,
                self.agg.input.schema().as_ref(),
            )
        } else {
            Ok(expr)
        }
    }

    /// Unproject a sort expression onto this aggregate, then normalize it so
    /// ORDER BY uses the same scope as the other clauses.
    fn prepare_sort_expr(
        &self,
        sort_expr: SortExpr,
        input: &LogicalPlan,
    ) -> Result<SortExpr> {
        let mut sort_expr = unproject_sort_expr(sort_expr, Some(self.agg), input)?;
        sort_expr.expr = self.normalize(sort_expr.expr)?;
        Ok(sort_expr)
    }
}

impl Unparser<'_> {
    pub fn plan_to_sql(&self, plan: &LogicalPlan) -> Result<ast::Statement> {
        let mut plan = normalize_union_schema(plan)?;
        if !self.dialect.supports_qualify() {
            plan = rewrite_qualify(plan)?;
        }

        match plan {
            LogicalPlan::Projection(_)
            | LogicalPlan::Filter(_)
            | LogicalPlan::Window(_)
            | LogicalPlan::Aggregate(_)
            | LogicalPlan::Sort(_)
            | LogicalPlan::Join(_)
            | LogicalPlan::Repartition(_)
            | LogicalPlan::Union(_)
            | LogicalPlan::TableScan(_)
            | LogicalPlan::EmptyRelation(_)
            | LogicalPlan::Subquery(_)
            | LogicalPlan::SubqueryAlias(_)
            | LogicalPlan::Limit(_)
            | LogicalPlan::Statement(_)
            | LogicalPlan::Values(_)
            | LogicalPlan::Distinct(_) => self.select_to_sql_statement(&plan),
            LogicalPlan::Dml(_) => self.dml_to_sql(&plan),
            LogicalPlan::Extension(extension) => {
                self.extension_to_statement(extension.node.as_ref())
            }
            LogicalPlan::Explain(_)
            | LogicalPlan::Analyze(_)
            | LogicalPlan::Ddl(_)
            | LogicalPlan::Copy(_)
            | LogicalPlan::DescribeTable(_)
            | LogicalPlan::RecursiveQuery(_)
            | LogicalPlan::Unnest(_) => not_impl_err!("Unsupported plan: {plan:?}"),
        }
    }

    /// Try to unparse a [UserDefinedLogicalNode] to a SQL statement.
    /// If multiple unparsers are registered for the same [UserDefinedLogicalNode],
    /// the first unparsing result will be returned.
    fn extension_to_statement(
        &self,
        node: &dyn UserDefinedLogicalNode,
    ) -> Result<ast::Statement> {
        let mut statement = None;
        for unparser in &self.extension_unparsers {
            match unparser.unparse_to_statement(node, self)? {
                UnparseToStatementResult::Modified(stmt) => {
                    statement = Some(stmt);
                    break;
                }
                UnparseToStatementResult::Unmodified => {}
            }
        }
        if let Some(statement) = statement {
            Ok(statement)
        } else {
            not_impl_err!("Unsupported extension node: {node:?}")
        }
    }

    /// Try to unparse a [UserDefinedLogicalNode] to a SQL statement.
    /// If multiple unparsers are registered for the same [UserDefinedLogicalNode],
    /// the first unparser supporting the node will be used.
    fn extension_to_sql(
        &self,
        node: &dyn UserDefinedLogicalNode,
        query: &mut Option<&mut QueryBuilder>,
        select: &mut Option<&mut SelectBuilder>,
        relation: &mut Option<&mut RelationBuilder>,
    ) -> Result<()> {
        for unparser in &self.extension_unparsers {
            match unparser.unparse(node, self, query, select, relation)? {
                UnparseWithinStatementResult::Modified => return Ok(()),
                UnparseWithinStatementResult::Unmodified => {}
            }
        }
        not_impl_err!("Unsupported extension node: {node:?}")
    }

    fn select_to_sql_statement(&self, plan: &LogicalPlan) -> Result<ast::Statement> {
        let mut query_builder = Some(QueryBuilder::default());

        let body = self.select_to_sql_expr(plan, &mut query_builder)?;

        let query = query_builder.unwrap().body(Box::new(body)).build()?;

        Ok(ast::Statement::Query(Box::new(query)))
    }

    fn select_to_sql_expr(
        &self,
        plan: &LogicalPlan,
        query: &mut Option<QueryBuilder>,
    ) -> Result<SetExpr> {
        let mut select_builder = SelectBuilder::default();
        select_builder.push_from(TableWithJoinsBuilder::default());
        let mut relation_builder = RelationBuilder::default();
        self.select_to_sql_recursively(
            plan,
            query,
            &mut select_builder,
            &mut relation_builder,
        )?;

        // If we were able to construct a full body (i.e. UNION ALL), return it
        if let Some(body) = query.as_mut().and_then(|q| q.take_body()) {
            return Ok(*body);
        }

        // If no projection is set, add a wildcard projection to the select
        // which will be translated to `SELECT *` in the SQL statement
        if !select_builder.already_projected() {
            select_builder.projection(vec![ast::SelectItem::Wildcard(
                ast::WildcardAdditionalOptions::default(),
            )]);
        }

        let mut twj = select_builder.pop_from().unwrap();
        twj.relation(relation_builder);
        select_builder.push_from(twj);

        Ok(SetExpr::Select(Box::new(select_builder.build()?)))
    }

    /// Reconstructs a SELECT SQL statement from a logical plan by
    /// unprojecting column expressions found in a [Projection] node. This
    /// requires scanning the plan tree for relevant Aggregate and Window
    /// nodes and matching column expressions to the appropriate agg or
    /// window expressions.
    ///
    /// `fully_absorbed` reports whether the Projection arm was able to
    /// absorb every `Sort`/`Limit` node between this Projection and the
    /// Aggregate/Window into the current SELECT. When `false`, the
    /// Aggregate/Window will end up in a derived subquery, so we fall
    /// back to passthrough column references that resolve against that
    /// subquery's output instead of unprojecting onto the original
    /// aggregate expressions.
    ///
    /// Returns `true` if an Aggregate node was found and claimed for this
    /// SELECT.
    fn reconstruct_select_statement(
        &self,
        plan: &LogicalPlan,
        p: &Projection,
        select: &mut SelectBuilder,
        fully_absorbed: bool,
    ) -> Result<bool> {
        let mut exprs = p.expr.clone();

        // If an Unnest node is found within the select, find and unproject the unnest column
        let flatten_alias = select.current_flatten_alias();
        if let Some(unnest) = find_unnest_node_within_select(plan) {
            if let Some(ref alias) = flatten_alias {
                exprs = exprs
                    .into_iter()
                    .map(|e| unproject_unnest_expr_as_flatten_value(e, unnest, alias))
                    .collect::<Result<Vec<_>>>()?;
            } else {
                exprs = exprs
                    .into_iter()
                    .map(|e| unproject_unnest_expr(e, unnest))
                    .collect::<Result<Vec<_>>>()?;
            }
        };

        // Rewrite column references that point to FLATTEN table aliases:
        // in Snowflake, FLATTEN output is accessed via .VALUE, not the
        // original column name.
        if !select.flatten_table_aliases_empty() {
            exprs = exprs
                .into_iter()
                .map(|e| {
                    e.transform(|expr| {
                        if let Expr::Column(ref col) = expr
                            && let Some(ref relation) = col.relation
                            && select.is_flatten_table_alias(relation.table())
                        {
                            return Ok(Transformed::yes(Expr::Column(Column::new(
                                Some(relation.clone()),
                                "VALUE",
                            ))));
                        }
                        Ok(Transformed::no(expr))
                    })
                    .map(|t| t.data)
                })
                .collect::<Result<Vec<_>>>()?;
        }

        // When some Sort/Limit nodes between this Projection and the
        // Aggregate/Window couldn't be absorbed into the current SELECT,
        // the Aggregate/Window will live inside a derived subquery. In
        // that case we use the passthrough projection path — column refs
        // resolve against the derived subquery's output columns instead
        // of being unprojected onto the original aggregate/window
        // expressions.
        let agg = if fully_absorbed {
            find_agg_node_within_select(plan, true)
        } else {
            None
        };
        let window = if fully_absorbed {
            find_window_nodes_within_select(plan, None, true)
        } else {
            None
        };
        match (agg, window) {
            (Some(agg), window) => {
                let window_option = window.as_deref();
                let unparser_agg_scope = UnparserAggScope::new(agg);
                let items = exprs
                    .into_iter()
                    .map(|proj_expr| {
                        let unproj =
                            unparser_agg_scope.prepare(proj_expr, window_option)?;
                        self.select_item_to_sql(&unproj)
                    })
                    .collect::<Result<Vec<_>>>()?;

                select.projection(items);
                select.group_by(ast::GroupByExpr::Expressions(
                    agg.group_expr
                        .iter()
                        .cloned()
                        .map(|expr| {
                            self.expr_to_sql(&unparser_agg_scope.normalize(expr)?)
                        })
                        .collect::<Result<Vec<_>>>()?,
                    vec![],
                ));
                Ok(true)
            }
            (None, Some(window)) => {
                let items = exprs
                    .into_iter()
                    .map(|proj_expr| {
                        let unproj = unproject_window_exprs(proj_expr, &window)?;
                        self.select_item_to_sql(&unproj)
                    })
                    .collect::<Result<Vec<_>>>()?;

                select.projection(items);
                Ok(false)
            }
            _ => {
                let items = exprs
                    .iter()
                    .map(|e| {
                        // After unproject_unnest_expr_as_flatten_value, an
                        // internal UNNEST display-name alias may still wrap
                        // the rewritten _unnest.VALUE column. Replace it
                        // with the bare FLATTEN VALUE select item.
                        if let Some(ref alias) = flatten_alias
                            && Self::has_internal_unnest_alias(e)
                        {
                            return Ok(self.build_flatten_value_select_item(alias, None));
                        }
                        self.select_item_to_sql(e)
                    })
                    .collect::<Result<Vec<_>>>()?;
                select.projection(items);
                Ok(false)
            }
        }
    }

    fn contains_projection_before_relation(plan: &LogicalPlan) -> bool {
        match plan {
            LogicalPlan::Projection(_) => true,
            LogicalPlan::TableScan(_)
            | LogicalPlan::Subquery(_)
            | LogicalPlan::SubqueryAlias(_)
            | LogicalPlan::Join(_)
            | LogicalPlan::EmptyRelation(_)
            | LogicalPlan::Values(_) => false,
            _ => {
                let inputs = plan.inputs();
                matches!(
                    inputs.as_slice(),
                    [input] if Self::contains_projection_before_relation(input)
                )
            }
        }
    }

    fn contains_aggregate_before_relation(plan: &LogicalPlan) -> bool {
        match plan {
            LogicalPlan::Aggregate(_) => true,
            LogicalPlan::TableScan(_)
            | LogicalPlan::Subquery(_)
            | LogicalPlan::SubqueryAlias(_)
            | LogicalPlan::Join(_)
            | LogicalPlan::EmptyRelation(_)
            | LogicalPlan::Values(_) => false,
            _ => {
                let inputs = plan.inputs();
                matches!(
                    inputs.as_slice(),
                    [input] if Self::contains_aggregate_before_relation(input)
                )
            }
        }
    }

    /// Unproject a sort expression; normalize it when the sort is above an
    /// aggregate, otherwise just unproject (no scope to normalize against).
    fn unproject_sort_expr_in_scope(
        sort_expr: SortExpr,
        agg: Option<&Aggregate>,
        input: &LogicalPlan,
    ) -> Result<SortExpr> {
        match agg {
            Some(agg) => UnparserAggScope::new(agg).prepare_sort_expr(sort_expr, input),
            None => unproject_sort_expr(sort_expr, None, input),
        }
    }

    fn derive(
        &self,
        plan: &LogicalPlan,
        relation: &mut RelationBuilder,
        alias: Option<ast::TableAlias>,
        lateral: bool,
    ) -> Result<()> {
        let mut derived_builder = DerivedRelationBuilder::default();
        derived_builder.lateral(lateral).alias(alias).subquery({
            let inner_statement = self.plan_to_sql(plan)?;
            if let ast::Statement::Query(inner_query) = inner_statement {
                inner_query
            } else {
                return internal_err!(
                    "Subquery must be a Query, but found {inner_statement:?}"
                );
            }
        });
        relation.derived(derived_builder);

        Ok(())
    }

    fn derive_with_dialect_alias(
        &self,
        alias: &str,
        plan: &LogicalPlan,
        relation: &mut RelationBuilder,
        lateral: bool,
        columns: Vec<Ident>,
    ) -> Result<()> {
        if self.dialect.requires_derived_table_alias() || !columns.is_empty() {
            self.derive(
                plan,
                relation,
                Some(self.new_table_alias(alias.to_string(), columns)),
                lateral,
            )
        } else {
            self.derive(plan, relation, None, lateral)
        }
    }

    /// Projection unparsing when [`super::dialect::Dialect::unnest_as_lateral_flatten`] is enabled:
    /// Snowflake-style `LATERAL FLATTEN` for unnest (not other dialect spellings).
    ///
    /// [`Self::peel_to_unnest_with_modifiers`] walks through any intermediate
    /// Limit/Sort nodes (the optimizer can insert these between the Projection
    /// and the Unnest), applies their modifiers to the query, and returns the
    /// Unnest plus the [`LogicalPlan`] ref to recurse into. This bypasses the
    /// normal Limit/Sort handlers which would wrap the subtree in a derived
    /// subquery.
    ///
    /// SELECT rendering is delegated to [`Self::reconstruct_select_statement`],
    /// which rewrites placeholder columns to `alias."VALUE"` via
    /// [`unproject_unnest_expr_as_flatten_value`].
    ///
    /// Returns `Ok(true)` when this path fully handled the projection.
    fn try_projection_unnest_as_lateral_flatten(
        &self,
        plan: &LogicalPlan,
        p: &Projection,
        query: &mut Option<QueryBuilder>,
        select: &mut SelectBuilder,
        relation: &mut RelationBuilder,
        unnest_input_type: Option<&UnnestInputType>,
    ) -> Result<bool> {
        // unnest_as_lateral_flatten: Snowflake LATERAL FLATTEN
        //
        // Generate the alias up front so that peel_to_unnest_with_modifiers
        // can rewrite ORDER BY placeholder columns to alias.VALUE.
        if self.dialect.unnest_as_lateral_flatten() && unnest_input_type.is_some() {
            let flatten_alias_name = if !select.already_projected() {
                select.next_flatten_alias()
            } else {
                select
                    .current_flatten_alias()
                    .unwrap_or_else(|| select.next_flatten_alias())
            };

            if let Some((unnest, unnest_plan)) = self.peel_to_unnest_with_modifiers(
                p.input.as_ref(),
                query,
                Some(&flatten_alias_name),
            )? && let Some(mut flatten) =
                self.try_unnest_to_lateral_flatten_sql(unnest)?
            {
                let inner_projection = Self::peel_to_inner_projection(
                    unnest.input.as_ref(),
                )
                .ok_or_else(|| {
                    internal_datafusion_err!(
                        "Unnest input is not a Projection: {:?}",
                        unnest.input
                    )
                })?;

                flatten.alias(Some(ast::TableAlias {
                    name: Ident::with_quote('"', &flatten_alias_name),
                    columns: vec![],
                    explicit: true,
                    at: None,
                }));

                if !select.already_projected() {
                    self.reconstruct_select_statement(plan, p, select, true)?;
                }

                if matches!(
                    inner_projection.input.as_ref(),
                    LogicalPlan::EmptyRelation(_)
                ) {
                    relation.flatten(flatten);
                    self.select_to_sql_recursively(unnest_plan, query, select, relation)?;
                    return Ok(true);
                }

                self.select_to_sql_recursively(unnest_plan, query, select, relation)?;

                let flatten_factor = flatten.build().map_err(|e| {
                    internal_datafusion_err!("Failed to build FLATTEN: {e}")
                })?;
                let cross_join = ast::Join {
                    relation: flatten_factor,
                    global: false,
                    join_operator: ast::JoinOperator::CrossJoin(
                        ast::JoinConstraint::None,
                    ),
                };
                if let Some(mut from) = select.pop_from() {
                    from.push_join(cross_join);
                    select.push_from(from);
                } else {
                    let mut twj = TableWithJoinsBuilder::default();
                    twj.push_join(cross_join);
                    select.push_from(twj);
                }

                return Ok(true);
            }
        }

        Ok(false)
    }

    fn project_window_output(
        &self,
        window_expr: &[Expr],
        select: &mut SelectBuilder,
        agg: Option<&Aggregate>,
    ) -> Result<()> {
        let mut items = if select.already_projected() {
            select.pop_projections()
        } else {
            vec![ast::SelectItem::Wildcard(
                ast::WildcardAdditionalOptions::default(),
            )]
        };

        items.extend(
            window_expr
                .iter()
                .map(|expr| {
                    // No normalization: this agg branch is only reachable from a
                    // hand-built plan. SQL wraps windows in a projection, which
                    // reconstruct_select_statement handles (and normalizes).
                    let expr = if let Some(agg) = agg {
                        unproject_agg_exprs(expr.clone(), agg, None)?
                    } else {
                        expr.clone()
                    };
                    self.select_item_to_sql(&expr)
                })
                .collect::<Result<Vec<_>>>()?,
        );
        select.projection(items);

        Ok(())
    }

    fn window_input_requires_derived_subquery(plan: &LogicalPlan) -> bool {
        // These operators either produce a SELECT list or apply SQL clauses
        // that are evaluated after window functions in a single SELECT block.
        // Keep them below the Window node by emitting a derived table.
        matches!(
            plan,
            LogicalPlan::Projection(_)
                | LogicalPlan::Distinct(_)
                | LogicalPlan::Limit(_)
                | LogicalPlan::Sort(_)
                | LogicalPlan::Union(_)
        )
    }

    fn window_to_sql_with_derived_input(
        &self,
        window: &Window,
        select: &mut SelectBuilder,
        relation: &mut RelationBuilder,
    ) -> Result<()> {
        let input_alias = "derived_window_input";
        self.derive(
            window.input.as_ref(),
            relation,
            Some(self.new_table_alias(input_alias.to_string(), vec![])),
            false,
        )?;

        let input_schema = window.input.schema();
        let mut alias_rewriter = TableAliasRewriter {
            table_schema: input_schema.as_ref(),
            alias_name: TableReference::bare(input_alias),
            rewrite_unqualified: true,
        };
        let window_expr = window
            .window_expr
            .iter()
            .map(|expr| expr.clone().rewrite(&mut alias_rewriter).data())
            .collect::<Result<Vec<_>>>()?;

        self.project_window_output(&window_expr, select, None)
    }

    fn extract_join_input_table_scan_filters(
        plan: &Arc<LogicalPlan>,
        table_scan_filters: &mut Vec<Expr>,
    ) -> Result<Arc<LogicalPlan>> {
        match try_transform_to_simple_table_scan_with_filters(plan)? {
            Some((plan, filters)) => {
                table_scan_filters.extend(filters);
                Ok(Arc::new(plan))
            }
            None => Ok(Arc::clone(plan)),
        }
    }

    #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    fn select_to_sql_recursively(
        &self,
        plan: &LogicalPlan,
        query: &mut Option<QueryBuilder>,
        select: &mut SelectBuilder,
        relation: &mut RelationBuilder,
    ) -> Result<()> {
        match plan {
            LogicalPlan::TableScan(scan) => {
                if let Some(unparsed_table_scan) = self.unparse_table_scan_pushdown(
                    plan,
                    None,
                    select.already_projected(),
                )? {
                    return self.select_to_sql_recursively(
                        &unparsed_table_scan,
                        query,
                        select,
                        relation,
                    );
                }
                let mut builder = TableRelationBuilder::default();
                let mut table_parts = vec![];
                if let Some(catalog_name) = scan.table_name.catalog() {
                    table_parts
                        .push(self.new_ident_quoted_if_needs(catalog_name.to_string()));
                }
                if let Some(schema_name) = scan.table_name.schema() {
                    table_parts
                        .push(self.new_ident_quoted_if_needs(schema_name.to_string()));
                }
                table_parts.push(
                    self.new_ident_quoted_if_needs(scan.table_name.table().to_string()),
                );
                builder.name(ast::ObjectName::from(table_parts));
                relation.table(builder);

                Ok(())
            }
            LogicalPlan::Projection(p) => {
                if let Some(new_plan) = rewrite_plan_for_sort_on_non_projected_fields(p) {
                    return self
                        .select_to_sql_recursively(&new_plan, query, select, relation);
                }

                // Projection can be top-level plan for unnest relation.
                // The projection generated by the `RecursiveUnnestRewriter`
                // will have at least one expression referencing an unnest
                // placeholder column.
                let unnest_input_type: Option<UnnestInputType> =
                    p.expr.iter().find_map(Self::find_unnest_placeholder);

                // --- UNNEST table factor path (BigQuery, etc.) ---
                // Only fires for a single bare-placeholder projection.
                // Uses peel_to_unnest_with_modifiers (rather than matching
                // p.input directly) to handle Limit/Sort between Projection
                // and Unnest.
                if self.dialect.unnest_as_table_factor()
                    && p.expr.len() == 1
                    && Self::is_bare_unnest_placeholder(&p.expr[0])
                    && let Some((unnest, unnest_plan)) =
                        self.peel_to_unnest_with_modifiers(p.input.as_ref(), query, None)?
                    && let Some(unnest_relation) =
                        self.try_unnest_to_table_factor_sql(unnest)?
                {
                    relation.unnest(unnest_relation);
                    return self.select_to_sql_recursively(
                        unnest_plan,
                        query,
                        select,
                        relation,
                    );
                }

                if self.try_projection_unnest_as_lateral_flatten(
                    plan,
                    p,
                    query,
                    select,
                    relation,
                    unnest_input_type.as_ref(),
                )? {
                    return Ok(());
                }

                // If it's a unnest projection, we should provide the table column alias
                // to provide a column name for the unnest relation.
                let columns = if unnest_input_type.is_some() {
                    p.expr
                        .iter()
                        .map(|e| {
                            self.new_ident_quoted_if_needs(e.schema_name().to_string())
                        })
                        .collect()
                } else {
                    vec![]
                };
                // Projection can be top-level plan for derived table
                if select.already_projected() {
                    return self.derive_with_dialect_alias(
                        "derived_projection",
                        plan,
                        relation,
                        unnest_input_type
                            .filter(|t| matches!(t, UnnestInputType::OuterReference))
                            .is_some(),
                        columns,
                    );
                }
                // For Snowflake FLATTEN: when the outer Projection has
                // UNNEST(...) display-name columns (from SELECT * / SELECT
                // UNNEST(...)), generate a flatten alias now so that
                // reconstruct_select_statement and the downstream Unnest
                // handler both use the same alias.
                if self.dialect.unnest_as_lateral_flatten()
                    && p.expr.iter().any(Self::has_internal_unnest_alias)
                {
                    select.next_flatten_alias();
                }
                // Pre-register FLATTEN table aliases from SubqueryAlias
                // nodes in the plan tree so that
                // reconstruct_select_statement can rewrite column
                // references (e.g. a.col → a.VALUE) before the
                // SubqueryAlias handler runs.
                if self.dialect.unnest_as_lateral_flatten() {
                    Self::collect_flatten_aliases(p.input.as_ref(), select);
                }
                // Walk down through consecutive Sort/Limit nodes, greedily
                // absorbing what can be folded into the SELECT we're
                // building around the Aggregate. A single SQL SELECT can
                // carry at most one `ORDER BY` (applied before `LIMIT`),
                // so the safe shape between us and the Aggregate is
                // `Limit* Sort?` (outer→inner). We stop at the first node
                // that would violate this; that node becomes the
                // subquery boundary, and recursion (seeing
                // `already_projected = true`) wraps it in a derived
                // relation. If we walk all the way to a non-Sort/non-Limit
                // terminator, the entire chain folds into one SELECT.
                //
                // Stacked Sorts with nothing between them collapse to the
                // outermost — the same simplification `EnforceSorting`
                // applies on the physical side — but only when no Limit
                // has been absorbed since the previous Sort, since the
                // inner Sort would otherwise be determining which rows
                // the Limit keeps.
                //
                // The fold is collected here without touching `query`
                // (apart from non-literal direct Limits, which don't
                // depend on projection form). Once we know whether every
                // Sort/Limit was absorbed we can pick the right
                // projection form and emit `ORDER BY` with or without
                // unprojection.
                let mut cur = p.input.as_ref();
                let mut absorbed_sort: Option<&Sort> = None;
                let mut combined_skip: usize = 0;
                let mut combined_fetch: Option<usize> = None;
                let mut have_combined_limit = false;
                let mut have_direct_limit = false;
                let mut have_order_by = false;
                loop {
                    match cur {
                        LogicalPlan::Limit(limit) => {
                            if have_order_by {
                                // Limit-below-Sort: `ORDER BY … LIMIT N`
                                // would apply the sort first, but the
                                // logical plan applies the Limit first.
                                break;
                            }
                            let skip_lit = limit.get_skip_type()?;
                            let fetch_lit = limit.get_fetch_type()?;
                            match (skip_lit, fetch_lit) {
                                (SkipType::Literal(s), FetchType::Literal(f)) => {
                                    if have_direct_limit {
                                        break;
                                    }
                                    if have_combined_limit {
                                        // outer = already-accumulated;
                                        // inner = this Limit. Same merge
                                        // rule as the optimizer.
                                        let (cs, cf) = combine_limit(
                                            combined_skip,
                                            combined_fetch,
                                            s,
                                            f,
                                        );
                                        combined_skip = cs;
                                        combined_fetch = cf;
                                    } else {
                                        combined_skip = s;
                                        combined_fetch = f;
                                        have_combined_limit = true;
                                    }
                                }
                                _ => {
                                    if have_combined_limit || have_direct_limit {
                                        // Cannot safely merge a
                                        // non-literal Limit with a prior
                                        // one; let recursion handle it.
                                        break;
                                    }
                                    let Some(query_ref) = query.as_mut() else {
                                        return internal_err!(
                                            "Limit operator only valid in a statement context."
                                        );
                                    };
                                    if let Some(fetch) = &limit.fetch {
                                        query_ref.limit(Some(self.expr_to_sql(fetch)?));
                                    }
                                    if let Some(skip) = &limit.skip {
                                        query_ref.offset(Some(ast::Offset {
                                            rows: ast::OffsetRows::None,
                                            value: self.expr_to_sql(skip)?,
                                        }));
                                    }
                                    have_direct_limit = true;
                                }
                            }
                            cur = limit.input.as_ref();
                        }
                        LogicalPlan::Sort(sort) if sort.fetch.is_some() => {
                            // `Sort { fetch }` is logically
                            // `Limit(fetch) -> Sort`. Try to absorb the
                            // virtual Limit first; only if that succeeds
                            // do we absorb the Sort. Otherwise we'd
                            // silently drop the fetch.
                            let fetch = sort.fetch.expect("guarded above");
                            if have_order_by {
                                // The virtual Limit would sit below an
                                // already-absorbed outer Sort.
                                break;
                            }
                            if have_direct_limit {
                                // Cannot combine a literal fetch with a
                                // non-literal direct Limit; let the
                                // derived subquery preserve both.
                                break;
                            }
                            if have_combined_limit {
                                let (cs, cf) = combine_limit(
                                    combined_skip,
                                    combined_fetch,
                                    0,
                                    Some(fetch),
                                );
                                combined_skip = cs;
                                combined_fetch = cf;
                            } else {
                                combined_skip = 0;
                                combined_fetch = Some(fetch);
                                have_combined_limit = true;
                            }
                            // Now the Sort itself. We know
                            // `!have_order_by` from the check above.
                            absorbed_sort = Some(sort);
                            have_order_by = true;
                            cur = sort.input.as_ref();
                        }
                        LogicalPlan::Sort(sort) => {
                            // Sort without `fetch`.
                            if have_order_by {
                                // Outer Sort already absorbed; the inner
                                // Sort is reordered by it and is
                                // conventionally dropped, matching
                                // `EnforceSorting` on the physical side.
                                cur = sort.input.as_ref();
                                continue;
                            }
                            absorbed_sort = Some(sort);
                            have_order_by = true;
                            cur = sort.input.as_ref();
                        }
                        _ => break,
                    }
                }

                // `fully_absorbed` is the bottom-up algorithm's "walked
                // all the way to the terminator without stopping": the
                // Aggregate/Window will live in the same SELECT as this
                // Projection, so we can unproject sort exprs and let
                // `reconstruct_select_statement` claim it.
                let fully_absorbed =
                    !matches!(cur, LogicalPlan::Limit(_) | LogicalPlan::Sort(_));
                let found_agg =
                    self.reconstruct_select_statement(plan, p, select, fully_absorbed)?;

                // Whether to bother emitting the absorbed clauses: only
                // if there's an Aggregate either claimed in this SELECT
                // or about to live in a derived subquery below us. If
                // there's nothing aggregate-like to fold over, fall
                // through and let the normal recursion handle the
                // Projection's input.
                let agg_below =
                    !fully_absorbed && find_agg_node_within_select(plan, true).is_some();
                if !(found_agg || agg_below) {
                    return self.select_to_sql_recursively(
                        p.input.as_ref(),
                        query,
                        select,
                        relation,
                    );
                }

                if let Some(sort) = absorbed_sort {
                    let Some(query_ref) = query.as_mut() else {
                        return internal_err!(
                            "Sort operator only valid in a statement context."
                        );
                    };
                    let sort_exprs: Vec<SortExpr> = if fully_absorbed {
                        let agg =
                            find_agg_node_within_select(plan, select.already_projected());
                        sort.expr
                            .iter()
                            .map(|sort_expr| {
                                Self::unproject_sort_expr_in_scope(
                                    sort_expr.clone(),
                                    agg,
                                    sort.input.as_ref(),
                                )
                            })
                            .collect::<Result<Vec<_>>>()?
                    } else {
                        sort.expr.clone()
                    };
                    query_ref.order_by(self.sorts_to_sql(&sort_exprs)?);
                }
                if have_combined_limit {
                    let Some(query_ref) = query.as_mut() else {
                        return internal_err!(
                            "Limit operator only valid in a statement context."
                        );
                    };
                    if let Some(fetch) = combined_fetch {
                        query_ref.limit(Some(ast::Expr::value(ast::Value::Number(
                            fetch.to_string(),
                            false,
                        ))));
                    }
                    if combined_skip > 0 {
                        query_ref.offset(Some(ast::Offset {
                            rows: ast::OffsetRows::None,
                            value: ast::Expr::value(ast::Value::Number(
                                combined_skip.to_string(),
                                false,
                            )),
                        }));
                    }
                }

                self.select_to_sql_recursively(cur, query, select, relation)
            }
            LogicalPlan::Filter(filter) => {
                let window = find_window_nodes_within_select(
                    plan,
                    None,
                    select.already_projected(),
                );
                let agg = find_agg_node_within_select(plan, select.already_projected());

                if let (Some(window), true) =
                    (window.as_deref(), self.dialect.supports_qualify())
                {
                    let mut unprojected =
                        unproject_window_exprs(filter.predicate.clone(), window)?;
                    if let Some(agg) = agg {
                        unprojected =
                            UnparserAggScope::new(agg).prepare(unprojected, None)?;
                    }
                    let filter_expr = self.expr_to_sql(&unprojected)?;
                    select.qualify(Some(filter_expr));
                } else if let Some(agg) = agg {
                    let unprojected = UnparserAggScope::new(agg)
                        .prepare(filter.predicate.clone(), None)?;
                    let filter_expr = self.expr_to_sql(&unprojected)?;
                    select.having(Some(filter_expr));
                } else {
                    let filter_expr = self.expr_to_sql(&filter.predicate)?;
                    select.selection(Some(filter_expr));
                }

                self.select_to_sql_recursively(
                    filter.input.as_ref(),
                    query,
                    select,
                    relation,
                )
            }
            LogicalPlan::Limit(limit) => {
                // Limit can be top-level plan for derived table
                if select.already_projected() {
                    return self.derive_with_dialect_alias(
                        "derived_limit",
                        plan,
                        relation,
                        false,
                        vec![],
                    );
                }
                if let Some(fetch) = &limit.fetch {
                    let Some(query) = query.as_mut() else {
                        return internal_err!(
                            "Limit operator only valid in a statement context."
                        );
                    };
                    query.limit(Some(self.expr_to_sql(fetch)?));
                }

                if let Some(skip) = &limit.skip {
                    let Some(query) = query.as_mut() else {
                        return internal_err!(
                            "Offset operator only valid in a statement context."
                        );
                    };

                    query.offset(Some(ast::Offset {
                        rows: ast::OffsetRows::None,
                        value: self.expr_to_sql(skip)?,
                    }));
                }

                self.select_to_sql_recursively(
                    limit.input.as_ref(),
                    query,
                    select,
                    relation,
                )
            }
            LogicalPlan::Sort(sort) => {
                // Sort can be top-level plan for derived table
                if select.already_projected() {
                    return self.derive_with_dialect_alias(
                        "derived_sort",
                        plan,
                        relation,
                        false,
                        vec![],
                    );
                }

                let Some(query_ref) = query else {
                    return internal_err!(
                        "Sort operator only valid in a statement context."
                    );
                };

                if let Some(fetch) = sort.fetch {
                    query_ref.limit(Some(ast::Expr::value(ast::Value::Number(
                        fetch.to_string(),
                        false,
                    ))));
                };

                let agg = find_agg_node_within_select(plan, select.already_projected());
                // unproject sort expressions
                let sort_exprs: Vec<SortExpr> = sort
                    .expr
                    .iter()
                    .map(|sort_expr| {
                        Self::unproject_sort_expr_in_scope(
                            sort_expr.clone(),
                            agg,
                            sort.input.as_ref(),
                        )
                    })
                    .collect::<Result<Vec<_>>>()?;

                query_ref.order_by(self.sorts_to_sql(&sort_exprs)?);

                self.select_to_sql_recursively(
                    sort.input.as_ref(),
                    query,
                    select,
                    relation,
                )
            }
            LogicalPlan::Aggregate(agg) => {
                // Aggregation can be already handled in the projection case
                if !select.already_projected() {
                    let unparser_agg_scope = UnparserAggScope::new(agg);
                    // The query returns aggregate and group expressions. If that weren't the case,
                    // the aggregate would have been placed inside a projection, making the check above^ false
                    let exprs: Vec<_> = agg
                        .aggr_expr
                        .iter()
                        .chain(agg.group_expr.iter())
                        .cloned()
                        .map(|expr| {
                            self.select_item_to_sql(&unparser_agg_scope.normalize(expr)?)
                        })
                        .collect::<Result<Vec<_>>>()?;
                    select.projection(exprs);

                    select.group_by(ast::GroupByExpr::Expressions(
                        agg.group_expr
                            .iter()
                            .cloned()
                            .map(|expr| {
                                self.expr_to_sql(&unparser_agg_scope.normalize(expr)?)
                            })
                            .collect::<Result<Vec<_>>>()?,
                        vec![],
                    ));
                } else if Self::contains_aggregate_before_relation(agg.input.as_ref()) {
                    return self.derive_with_dialect_alias(
                        "derived_aggregate",
                        agg.input.as_ref(),
                        relation,
                        false,
                        vec![],
                    );
                }

                self.select_to_sql_recursively(
                    agg.input.as_ref(),
                    query,
                    select,
                    relation,
                )
            }
            LogicalPlan::Distinct(distinct) => {
                // Distinct can be top-level plan for derived table
                if select.already_projected() {
                    return self.derive_with_dialect_alias(
                        "derived_distinct",
                        plan,
                        relation,
                        false,
                        vec![],
                    );
                }

                // If this distinct is the parent of a Union and we're in a query context,
                // then we need to unparse as a `UNION` rather than a `UNION ALL`.
                if let Distinct::All(input) = distinct
                    && matches!(input.as_ref(), LogicalPlan::Union(_))
                    && let Some(query_mut) = query.as_mut()
                {
                    query_mut.distinct_union();
                    return self.select_to_sql_recursively(
                        input.as_ref(),
                        query,
                        select,
                        relation,
                    );
                }

                let (select_distinct, input) = match distinct {
                    Distinct::All(input) => (ast::Distinct::Distinct, input.as_ref()),
                    Distinct::On(on) => {
                        let exprs = on
                            .on_expr
                            .iter()
                            .map(|e| self.expr_to_sql(e))
                            .collect::<Result<Vec<_>>>()?;
                        let items = on
                            .select_expr
                            .iter()
                            .map(|e| self.select_item_to_sql(e))
                            .collect::<Result<Vec<_>>>()?;
                        if let Some(sort_expr) = &on.sort_expr {
                            if let Some(query_ref) = query {
                                query_ref.order_by(self.sorts_to_sql(sort_expr)?);
                            } else {
                                return internal_err!(
                                    "Sort operator only valid in a statement context."
                                );
                            }
                        }
                        select.projection(items);
                        (ast::Distinct::On(exprs), on.input.as_ref())
                    }
                };
                select.distinct(Some(select_distinct));
                self.select_to_sql_recursively(input, query, select, relation)
            }
            LogicalPlan::Join(join) => {
                let mut table_scan_filters = vec![];
                let (left_plan, right_plan) = match join.join_type {
                    JoinType::RightSemi | JoinType::RightAnti => {
                        (&join.right, &join.left)
                    }
                    _ => (&join.left, &join.right),
                };
                // If there's an outer projection plan, it will already set up the projection.
                // In that case, we don't need to worry about setting up the projection here.
                // The outer projection plan will handle projecting the correct columns.
                let already_projected = select.already_projected();

                let left_plan = Self::extract_join_input_table_scan_filters(
                    left_plan,
                    &mut table_scan_filters,
                )?;
                let left_plan = if already_projected {
                    Self::unwrap_qualified_passthrough_join_projection(left_plan)
                } else {
                    left_plan
                };

                self.select_to_sql_recursively(
                    left_plan.as_ref(),
                    query,
                    select,
                    relation,
                )?;

                let left_projection: Option<Vec<ast::SelectItem>> = if !already_projected
                {
                    Some(select.pop_projections())
                } else {
                    None
                };

                let right_plan = Self::extract_join_input_table_scan_filters(
                    right_plan,
                    &mut table_scan_filters,
                )?;

                let mut right_relation = RelationBuilder::default();
                if already_projected
                    && let Some(nested_relation) = self
                        .qualified_passthrough_join_projection_to_nested_relation(
                            right_plan.as_ref(),
                            query,
                        )?
                {
                    right_relation = nested_relation;
                } else {
                    self.select_to_sql_recursively(
                        right_plan.as_ref(),
                        query,
                        select,
                        &mut right_relation,
                    )?;
                }

                let (join_filters, where_filters) = Self::split_join_on_and_where_filters(
                    join.join_type,
                    &join.filter,
                    table_scan_filters,
                );
                for filter in where_filters {
                    let filter_expr = self.expr_to_sql(&filter)?;
                    select.selection(Some(filter_expr));
                }

                let join_constraint = self.join_constraint_to_sql(
                    join.join_constraint,
                    &join.on,
                    join_filters.as_ref(),
                )?;

                let right_projection: Option<Vec<ast::SelectItem>> = if !already_projected
                {
                    Some(select.pop_projections())
                } else {
                    None
                };

                match join.join_type {
                    JoinType::LeftSemi
                    | JoinType::LeftAnti
                    | JoinType::LeftMark
                    | JoinType::RightSemi
                    | JoinType::RightAnti
                    | JoinType::RightMark => {
                        let mut query_builder = QueryBuilder::default();
                        let mut from = TableWithJoinsBuilder::default();
                        let mut exists_select: SelectBuilder = SelectBuilder::default();
                        from.relation(right_relation);
                        exists_select.push_from(from);
                        if let Some(filter) = &join.filter {
                            exists_select.selection(Some(self.expr_to_sql(filter)?));
                        }
                        for (left, right) in &join.on {
                            exists_select.selection(Some(
                                self.expr_to_sql(&left.clone().eq(right.clone()))?,
                            ));
                        }
                        exists_select.projection(vec![ast::SelectItem::UnnamedExpr(
                            ast::Expr::value(ast::Value::Number("1".to_string(), false)),
                        )]);
                        query_builder.body(Box::new(SetExpr::Select(Box::new(
                            exists_select.build()?,
                        ))));

                        let negated = match join.join_type {
                            JoinType::LeftSemi
                            | JoinType::RightSemi
                            | JoinType::LeftMark
                            | JoinType::RightMark => false,
                            JoinType::LeftAnti | JoinType::RightAnti => true,
                            _ => unreachable!(),
                        };
                        let exists_expr = ast::Expr::Exists {
                            subquery: Box::new(query_builder.build()?),
                            negated,
                        };

                        match join.join_type {
                            JoinType::LeftMark | JoinType::RightMark => {
                                let source_schema =
                                    if join.join_type == JoinType::LeftMark {
                                        right_plan.schema()
                                    } else {
                                        left_plan.schema()
                                    };
                                let (table_ref, _) = source_schema.qualified_field(0);
                                let column = self.col_to_sql(&Column::new(
                                    table_ref.cloned(),
                                    "mark",
                                ))?;
                                select.replace_mark(&column, &exists_expr);
                            }
                            _ => {
                                select.selection(Some(exists_expr));
                            }
                        }
                        if let Some(projection) = left_projection {
                            select.projection(projection);
                        }
                    }
                    JoinType::Inner
                    | JoinType::Left
                    | JoinType::Right
                    | JoinType::Full => {
                        let Ok(Some(relation)) = right_relation.build() else {
                            return internal_err!("Failed to build right relation");
                        };
                        let ast_join = ast::Join {
                            relation,
                            global: false,
                            join_operator: self
                                .join_operator_to_sql(join.join_type, join_constraint)?,
                        };
                        let mut from = select.pop_from().unwrap();
                        from.push_join(ast_join);
                        select.push_from(from);
                        if !already_projected {
                            let Some(left_projection) = left_projection else {
                                return internal_err!("Left projection is missing");
                            };

                            let Some(right_projection) = right_projection else {
                                return internal_err!("Right projection is missing");
                            };

                            let projection = left_projection
                                .into_iter()
                                .chain(right_projection)
                                .collect();
                            select.projection(projection);
                        }
                    }
                };

                Ok(())
            }
            LogicalPlan::SubqueryAlias(plan_alias) => {
                let (plan, mut columns) =
                    subquery_alias_inner_query_and_columns(plan_alias);
                let unparsed_table_scan = self.unparse_table_scan_pushdown(
                    plan,
                    Some(plan_alias.alias.clone()),
                    select.already_projected(),
                )?;

                // If the (possibly rewritten) inner plan builds its own
                // SELECT clauses (e.g. Aggregate adds GROUP BY, Window adds
                // OVER, etc.) and unparse_table_scan_pushdown couldn't reduce it,
                // we must emit a derived subquery: (SELECT ...) AS alias.
                // Without this, the recursive handler would merge those clauses
                // into the outer SELECT, losing the subquery structure entirely.
                if unparsed_table_scan.is_none() && Self::requires_derived_subquery(plan)
                {
                    // When the dialect does not support column aliases in
                    // table aliases (e.g. SQLite), inject the aliases into
                    // the inner projection before wrapping as a derived
                    // subquery.
                    if !columns.is_empty()
                        && !self.dialect.supports_column_alias_in_table_alias()
                    {
                        let Ok(rewritten_plan) =
                            inject_column_aliases_into_subquery(plan.clone(), columns)
                        else {
                            return internal_err!(
                                "Failed to transform SubqueryAlias plan"
                            );
                        };
                        return self.derive(
                            &rewritten_plan,
                            relation,
                            Some(self.new_table_alias(
                                plan_alias.alias.table().to_string(),
                                vec![],
                            )),
                            false,
                        );
                    }
                    return self.derive(
                        plan,
                        relation,
                        Some(self.new_table_alias(
                            plan_alias.alias.table().to_string(),
                            columns,
                        )),
                        false,
                    );
                }

                // if the child plan is a TableScan with pushdown operations, we don't need to
                // create an additional subquery for it
                if !select.already_projected() && unparsed_table_scan.is_none() {
                    select.projection(vec![ast::SelectItem::Wildcard(
                        ast::WildcardAdditionalOptions::default(),
                    )]);
                }
                let plan = unparsed_table_scan.unwrap_or_else(|| plan.clone());
                if !columns.is_empty()
                    && !self.dialect.supports_column_alias_in_table_alias()
                {
                    // Instead of specifying column aliases as part of the outer table, inject them directly into the inner projection
                    let rewritten_plan =
                        match inject_column_aliases_into_subquery(plan, columns) {
                            Ok(p) => p,
                            Err(e) => {
                                return internal_err!(
                                    "Failed to transform SubqueryAlias plan: {e}"
                                );
                            }
                        };

                    columns = vec![];

                    self.select_to_sql_recursively(
                        &rewritten_plan,
                        query,
                        select,
                        relation,
                    )?;
                } else {
                    self.select_to_sql_recursively(&plan, query, select, relation)?;
                }

                relation.alias(Some(
                    self.new_table_alias(plan_alias.alias.table().to_string(), columns),
                ));

                // If this SubqueryAlias wraps a FLATTEN (Snowflake unnest),
                // register the alias so the outer Projection can rewrite
                // column references to use VALUE.
                if self.dialect.unnest_as_lateral_flatten()
                    && find_unnest_node_until_relation(plan_alias.input.as_ref())
                        .is_some()
                {
                    select.add_flatten_table_alias(plan_alias.alias.table().to_string());
                }

                Ok(())
            }
            LogicalPlan::Union(union) => {
                // Covers cases where the UNION is a subquery and the projection is at the top level
                if select.already_projected() {
                    return self.derive_with_dialect_alias(
                        "derived_union",
                        plan,
                        relation,
                        false,
                        vec![],
                    );
                }

                let input_exprs: Vec<SetExpr> = union
                    .inputs
                    .iter()
                    .map(|input| self.select_to_sql_expr(input, query))
                    .collect::<Result<Vec<_>>>()?;

                assert_or_internal_err!(
                    input_exprs.len() >= 2,
                    "UNION operator requires at least 2 inputs"
                );

                let set_quantifier =
                    if query.as_ref().is_some_and(|q| q.is_distinct_union()) {
                        // Setting the SetQuantifier to None will unparse as a `UNION`
                        // rather than a `UNION ALL`.
                        ast::SetQuantifier::None
                    } else {
                        ast::SetQuantifier::All
                    };

                // Build the union expression tree bottom-up by reversing the order
                // note that we are also swapping left and right inputs because of the rev
                let union_expr = input_exprs
                    .into_iter()
                    .rev()
                    .reduce(|a, b| SetExpr::SetOperation {
                        op: ast::SetOperator::Union,
                        set_quantifier,
                        left: Box::new(b),
                        right: Box::new(a),
                    })
                    .unwrap();

                let Some(query) = query.as_mut() else {
                    return internal_err!(
                        "UNION ALL operator only valid in a statement context"
                    );
                };
                query.body(Box::new(union_expr));

                Ok(())
            }
            LogicalPlan::Window(window) => {
                // Window nodes are usually handled simultaneously with Projection
                // nodes, where projected columns are unprojected back into their
                // corresponding window expressions. Manually built plans can have
                // Window nodes without an enclosing Projection, so in that case
                // the Window node itself must contribute its output expressions.
                let project_window_output = !select.already_projected();
                if project_window_output
                    && Self::window_input_requires_derived_subquery(window.input.as_ref())
                {
                    return self
                        .window_to_sql_with_derived_input(window, select, relation);
                }

                let agg = if project_window_output {
                    find_agg_node_within_select(plan, false)
                } else {
                    None
                };

                self.select_to_sql_recursively(
                    window.input.as_ref(),
                    query,
                    select,
                    relation,
                )?;

                if project_window_output {
                    self.project_window_output(&window.window_expr, select, agg)?;
                }

                Ok(())
            }
            LogicalPlan::EmptyRelation(_) => {
                // An EmptyRelation could be behind an UNNEST node. If the dialect supports UNNEST as a table factor,
                // a TableRelationBuilder will be created for the UNNEST node first.
                if !relation.has_relation() {
                    relation.empty();
                }
                Ok(())
            }
            LogicalPlan::Extension(extension) => {
                if let Some(query) = query.as_mut() {
                    self.extension_to_sql(
                        extension.node.as_ref(),
                        &mut Some(query),
                        &mut Some(select),
                        &mut Some(relation),
                    )
                } else {
                    self.extension_to_sql(
                        extension.node.as_ref(),
                        &mut None,
                        &mut Some(select),
                        &mut Some(relation),
                    )
                }
            }
            LogicalPlan::Unnest(unnest) => {
                if !unnest.struct_type_columns.is_empty() {
                    if self.dialect.unnest_as_lateral_flatten() {
                        return not_impl_err!(
                            "Snowflake FLATTEN cannot unparse struct unnest: \
                             DataFusion expands struct fields into columns (horizontal), \
                             but Snowflake FLATTEN expands them into rows (vertical). \
                             Columns: {:?}",
                            unnest.struct_type_columns
                        );
                    }
                    return internal_err!(
                        "Struct type columns are not currently supported in UNNEST: {:?}",
                        unnest.struct_type_columns
                    );
                }

                // For Snowflake FLATTEN: if the relation hasn't been set yet
                // (UNNEST was in SELECT clause, not FROM clause), set the FLATTEN
                // relation here so the FROM clause is emitted.
                if self.dialect.unnest_as_lateral_flatten()
                    && !relation.has_relation()
                    && let Some(mut flatten_relation) =
                        self.try_unnest_to_lateral_flatten_sql(unnest)?
                {
                    // Use the alias already generated by the Projection
                    // handler so SELECT items and the FLATTEN relation
                    // reference the same name.
                    if let Some(alias) = select.current_flatten_alias() {
                        flatten_relation.alias(Some(ast::TableAlias {
                            name: Ident::with_quote('"', &alias),
                            columns: vec![],
                            explicit: true,
                            at: None,
                        }));
                    }
                    relation.flatten(flatten_relation);
                }

                // In the case of UNNEST, the Unnest node is followed by a duplicate Projection node that we should skip.
                // Otherwise, there will be a duplicate SELECT clause.
                // | Projection: table.col1, UNNEST(table.col2)
                // |   Unnest: UNNEST(table.col2)
                // |     Projection: table.col1, table.col2 AS UNNEST(table.col2)
                // |       Filter: table.col3 = Int64(3)
                // |         TableScan: table projection=None
                if let Some(p) = Self::peel_to_inner_projection(unnest.input.as_ref()) {
                    // Skip the inner Projection (synthetic rewriter node)
                    // and continue with its input.
                    self.select_to_sql_recursively(&p.input, query, select, relation)
                } else {
                    internal_err!("Unnest input is not a Projection: {unnest:?}")
                }
            }
            LogicalPlan::Subquery(subquery)
                if find_unnest_node_until_relation(subquery.subquery.as_ref())
                    .is_some() =>
            {
                if self.dialect.unnest_as_table_factor()
                    || self.dialect.unnest_as_lateral_flatten()
                {
                    self.select_to_sql_recursively(
                        subquery.subquery.as_ref(),
                        query,
                        select,
                        relation,
                    )
                } else {
                    self.derive_with_dialect_alias(
                        "derived_unnest",
                        subquery.subquery.as_ref(),
                        relation,
                        true,
                        vec![],
                    )
                }
            }
            _ => {
                not_impl_err!("Unsupported operator: {plan:?}")
            }
        }
    }

    /// Walk through transparent nodes (SubqueryAlias) to find the inner
    /// Projection that feeds an Unnest node.
    ///
    /// The inner Projection is created atomically by the
    /// `RecursiveUnnestRewriter` and contains the array expression that the
    /// Unnest operates on. A `SubqueryAlias` (e.g. from a virtual/passthrough
    /// table) may wrap the Projection.
    fn peel_to_inner_projection(plan: &LogicalPlan) -> Option<&Projection> {
        match plan {
            LogicalPlan::Projection(p) => Some(p),
            LogicalPlan::SubqueryAlias(alias) => {
                Self::peel_to_inner_projection(alias.input.as_ref())
            }
            _ => None,
        }
    }

    /// Walk through transparent nodes (Limit, Sort) between the outer
    /// Projection and the Unnest, applying their SQL modifiers (LIMIT,
    /// OFFSET, ORDER BY) to the query builder. Returns the `Unnest` node
    /// and a reference to the enclosing `LogicalPlan` for recursion, or
    /// `Ok(None)` if no Unnest is found.
    ///
    /// By processing Limit/Sort inline and then recursing into the Unnest
    /// plan directly, we bypass the normal Limit/Sort handlers which would
    /// create unwanted derived subqueries (since `already_projected` is
    /// set at the point this is called).
    fn peel_to_unnest_with_modifiers<'a>(
        &self,
        plan: &'a LogicalPlan,
        query: &mut Option<QueryBuilder>,
        flatten_alias: Option<&str>,
    ) -> Result<Option<(&'a Unnest, &'a LogicalPlan)>> {
        match plan {
            LogicalPlan::Unnest(unnest) => Ok(Some((unnest, plan))),
            LogicalPlan::Limit(limit) => {
                if let Some(fetch) = &limit.fetch
                    && let Some(q) = query.as_mut()
                {
                    q.limit(Some(self.expr_to_sql(fetch)?));
                }
                if let Some(skip) = &limit.skip
                    && let Some(q) = query.as_mut()
                {
                    q.offset(Some(ast::Offset {
                        rows: ast::OffsetRows::None,
                        value: self.expr_to_sql(skip)?,
                    }));
                }
                self.peel_to_unnest_with_modifiers(
                    limit.input.as_ref(),
                    query,
                    flatten_alias,
                )
            }
            LogicalPlan::Sort(sort) => {
                let Some(query_ref) = query.as_mut() else {
                    return internal_err!(
                        "Sort between Projection and Unnest requires a statement context."
                    );
                };
                if let Some(fetch) = sort.fetch {
                    query_ref.limit(Some(ast::Expr::value(ast::Value::Number(
                        fetch.to_string(),
                        false,
                    ))));
                }
                // When a flatten_alias is provided, rewrite
                // __unnest_placeholder(...) columns in sort expressions to
                // alias.VALUE so ORDER BY references the FLATTEN output.
                let unnest_node = match sort.input.as_ref() {
                    LogicalPlan::Unnest(u) => Some(u),
                    _ => find_unnest_node_within_select(sort.input.as_ref()),
                };
                let sort_exprs = if let Some(alias) = flatten_alias
                    && let Some(unnest) = unnest_node
                {
                    sort.expr
                        .iter()
                        .map(|s| {
                            let rewritten = unproject_unnest_expr_as_flatten_value(
                                s.expr.clone(),
                                unnest,
                                alias,
                            )?;
                            Ok(SortExpr {
                                expr: rewritten,
                                ..s.clone()
                            })
                        })
                        .collect::<Result<Vec<_>>>()?
                } else {
                    sort.expr.clone()
                };
                query_ref.order_by(self.sorts_to_sql(&sort_exprs)?);
                self.peel_to_unnest_with_modifiers(
                    sort.input.as_ref(),
                    query,
                    flatten_alias,
                )
            }
            _ => Ok(None),
        }
    }

    /// Search an expression tree for an unnest placeholder column reference.
    ///
    /// Returns the [`UnnestInputType`] if any sub-expression is a column
    /// whose name starts with `__unnest_placeholder`. The placeholder may
    /// be at the top level (bare), inside a function call, or one of several
    /// expressions — this function finds it regardless.
    fn find_unnest_placeholder(expr: &Expr) -> Option<UnnestInputType> {
        let mut result = None;
        let _ = expr.apply(|e| {
            if let Some(t) = Self::classify_placeholder_column(e) {
                result = Some(t);
                return Ok(TreeNodeRecursion::Stop);
            }
            Ok(TreeNodeRecursion::Continue)
        });
        result
    }

    /// Returns true if `expr` is a placeholder column, optionally wrapped
    /// in a single alias (the rewriter's internal `UNNEST(...)` name).
    /// Does NOT match when a user alias wraps the internal alias
    /// (e.g. `Alias("c1", Alias("UNNEST(...)", Column(placeholder)))`),
    /// so the table-factor path correctly falls through to
    /// `reconstruct_select_statement` which preserves user aliases.
    fn is_bare_unnest_placeholder(expr: &Expr) -> bool {
        // Peel at most one alias layer (the rewriter's internal name).
        let inner = match expr {
            Expr::Alias(Alias { expr, .. }) => expr.as_ref(),
            other => other,
        };
        Self::classify_placeholder_column(inner).is_some()
    }

    /// If `expr` is a `Column` whose name starts with `__unnest_placeholder`,
    /// classify it as [`UnnestInputType::OuterReference`] or
    /// [`UnnestInputType::Scalar`].
    fn classify_placeholder_column(expr: &Expr) -> Option<UnnestInputType> {
        if let Expr::Column(Column { name, .. }) = expr
            && let Some(prefix) = name.strip_prefix(UNNEST_PLACEHOLDER)
        {
            if prefix.starts_with(&format!("({OUTER_REFERENCE_COLUMN_PREFIX}(")) {
                return Some(UnnestInputType::OuterReference);
            }
            return Some(UnnestInputType::Scalar);
        }
        None
    }

    /// Check whether an expression carries an internal `UNNEST(...)` display
    /// name as its column name or outermost alias. After
    /// [`unproject_unnest_expr_as_flatten_value`] rewrites the placeholder
    /// column to `_unnest.VALUE`, the internal alias may still linger
    /// (e.g. `Alias("UNNEST(make_array(...))", Column("_unnest.VALUE"))`).
    /// Callers use this to replace the expression with a clean
    /// `_unnest."VALUE"` select item.
    fn has_internal_unnest_alias(expr: &Expr) -> bool {
        match expr {
            Expr::Column(col) => {
                col.name.starts_with(&format!("{UNNEST_COLUMN_PREFIX}("))
            }
            Expr::Alias(Alias { name, .. }) => {
                name.starts_with(&format!("{UNNEST_COLUMN_PREFIX}("))
            }
            _ => false,
        }
    }

    /// Walk the plan tree and register any SubqueryAlias that wraps an
    /// unnest as a FLATTEN table alias on the SelectBuilder. This allows
    /// `reconstruct_select_statement` to rewrite column references (e.g.
    /// `a.col` → `a.VALUE`) before the SubqueryAlias handler runs.
    /// Returns true if a plan tree contains an Unnest node, searching
    /// through Projection, Subquery, and SubqueryAlias wrappers.
    fn contains_unnest(plan: &LogicalPlan) -> bool {
        match plan {
            LogicalPlan::Unnest(_) => true,
            LogicalPlan::Projection(p) => Self::contains_unnest(&p.input),
            LogicalPlan::Subquery(s) => Self::contains_unnest(&s.subquery),
            LogicalPlan::SubqueryAlias(a) => Self::contains_unnest(&a.input),
            _ => false,
        }
    }

    fn collect_flatten_aliases(plan: &LogicalPlan, select: &mut SelectBuilder) {
        match plan {
            LogicalPlan::SubqueryAlias(alias)
                if Self::contains_unnest(alias.input.as_ref()) =>
            {
                select.add_flatten_table_alias(alias.alias.table().to_string());
            }
            LogicalPlan::Join(join) => {
                Self::collect_flatten_aliases(&join.left, select);
                Self::collect_flatten_aliases(&join.right, select);
            }
            _ => {}
        }
    }

    fn try_unnest_to_table_factor_sql(
        &self,
        unnest: &Unnest,
    ) -> Result<Option<UnnestRelationBuilder>> {
        let mut unnest_relation = UnnestRelationBuilder::default();
        let LogicalPlan::Projection(projection) = unnest.input.as_ref() else {
            return Ok(None);
        };

        if !matches!(projection.input.as_ref(), LogicalPlan::EmptyRelation(_)) {
            // It may be possible that UNNEST is used as a source for the query.
            // However, at this point, we don't yet know if it is just a single expression
            // from another source or if it's from UNNEST.
            //
            // Unnest(Projection(EmptyRelation)) denotes a case with `UNNEST([...])`,
            // which is normally safe to unnest as a table factor.
            // However, in the future, more comprehensive checks can be added here.
            return Ok(None);
        };

        let exprs = projection
            .expr
            .iter()
            .map(|e| self.expr_to_sql(e))
            .collect::<Result<Vec<_>>>()?;
        unnest_relation.array_exprs(exprs);

        Ok(Some(unnest_relation))
    }

    /// Build a `SELECT alias."VALUE"` item for Snowflake FLATTEN output.
    fn build_flatten_value_select_item(
        &self,
        flatten_alias: &str,
        user_alias: Option<&str>,
    ) -> ast::SelectItem {
        let compound = ast::Expr::CompoundIdentifier(vec![
            self.new_ident_quoted_if_needs(flatten_alias.to_string()),
            Ident::with_quote('"', "VALUE"),
        ]);
        match user_alias {
            Some(alias) => ast::SelectItem::ExprWithAlias {
                expr: compound,
                alias: self.new_ident_quoted_if_needs(alias.to_string()),
            },
            None => ast::SelectItem::UnnamedExpr(compound),
        }
    }

    /// Convert an `Unnest` logical plan node to a `LATERAL FLATTEN(INPUT => expr, ...)`
    /// table factor for Snowflake-style SQL output.
    fn try_unnest_to_lateral_flatten_sql(
        &self,
        unnest: &Unnest,
    ) -> Result<Option<FlattenRelationBuilder>> {
        let Some(projection) = Self::peel_to_inner_projection(unnest.input.as_ref())
        else {
            return Ok(None);
        };

        // For now, handle the simple case of a single expression to flatten.
        // Multi-expression would require multiple LATERAL FLATTEN calls chained together.
        let Some(first_expr) = projection.expr.first() else {
            return Ok(None);
        };

        let input_expr = self.expr_to_sql(first_expr)?;

        let mut flatten = FlattenRelationBuilder::default();
        flatten.input_expr(input_expr);
        flatten.outer(unnest.options.preserve_nulls());

        Ok(Some(flatten))
    }

    fn is_scan_with_pushdown(scan: &TableScan) -> bool {
        scan.projection.is_some() || !scan.filters.is_empty() || scan.fetch.is_some()
    }

    /// Returns true if a plan, when used as the direct child of a SubqueryAlias,
    /// must be emitted as a derived subquery `(SELECT ...) AS alias`.
    ///
    /// Plans like Aggregate or Window build their own SELECT clauses (GROUP BY,
    /// window functions).
    fn requires_derived_subquery(plan: &LogicalPlan) -> bool {
        matches!(
            plan,
            LogicalPlan::Aggregate(_)
                | LogicalPlan::Window(_)
                | LogicalPlan::Sort(_)
                | LogicalPlan::Limit(_)
                | LogicalPlan::Union(_)
        )
    }

    fn is_qualified_passthrough_projection(projection: &Projection) -> bool {
        projection
            .expr
            .iter()
            .all(|expr| matches!(expr, Expr::Column(column) if column.relation.is_some()))
    }

    fn unwrap_qualified_passthrough_join_projection(
        plan: Arc<LogicalPlan>,
    ) -> Arc<LogicalPlan> {
        if let LogicalPlan::Projection(projection) = plan.as_ref()
            && matches!(projection.input.as_ref(), LogicalPlan::Join(_))
            && Self::is_qualified_passthrough_projection(projection)
        {
            Arc::clone(&projection.input)
        } else {
            plan
        }
    }

    fn qualified_passthrough_join_projection_to_nested_relation(
        &self,
        plan: &LogicalPlan,
        query: &mut Option<QueryBuilder>,
    ) -> Result<Option<RelationBuilder>> {
        let LogicalPlan::Projection(projection) = plan else {
            return Ok(None);
        };
        if !matches!(projection.input.as_ref(), LogicalPlan::Join(_))
            || !Self::is_qualified_passthrough_projection(projection)
        {
            return Ok(None);
        }

        let original_query = query.clone();
        let mut nested_select = SelectBuilder::default();
        nested_select.push_from(TableWithJoinsBuilder::default());
        let mut nested_relation = RelationBuilder::default();
        self.select_to_sql_recursively(
            projection.input.as_ref(),
            query,
            &mut nested_select,
            &mut nested_relation,
        )?;
        if nested_select.has_selection() {
            *query = original_query;
            return Ok(None);
        }

        let Some(mut nested_from) = nested_select.pop_from() else {
            return internal_err!("Failed to build nested join relation");
        };
        nested_from.relation(nested_relation);
        let Some(table_with_joins) = nested_from.build()? else {
            return internal_err!("Failed to build nested join relation");
        };

        let mut relation = RelationBuilder::default();
        relation.nested_join(table_with_joins, None);
        Ok(Some(relation))
    }

    /// Strip the table qualifier from every column in an expression that must
    /// resolve against an unnamed derived table's output columns rather than a
    /// deeper table alias that is out of scope at this nesting level.
    fn strip_column_qualifiers(expr: Expr) -> Result<Expr> {
        expr.transform(|e| match e {
            Expr::Column(mut column) => {
                column.relation = None;
                Ok(Transformed::yes(Expr::Column(column)))
            }
            other => Ok(Transformed::no(other)),
        })
        .data()
    }

    fn strip_column_qualifiers_for_schema(expr: Expr, schema: &DFSchema) -> Result<Expr> {
        expr.transform(|e| match e {
            Expr::Column(mut column)
                if column.relation.is_some()
                    && schema.index_of_column(&column).is_ok() =>
            {
                column.relation = None;
                Ok(Transformed::yes(Expr::Column(column)))
            }
            other => Ok(Transformed::no(other)),
        })
        .data()
    }

    /// Try to unparse a table scan with pushdown operations into a new subquery plan.
    /// If the table scan is without any pushdown operations, return None.
    fn unparse_table_scan_pushdown(
        &self,
        plan: &LogicalPlan,
        alias: Option<TableReference>,
        already_projected: bool,
    ) -> Result<Option<LogicalPlan>> {
        match plan {
            LogicalPlan::TableScan(table_scan) => {
                if !Self::is_scan_with_pushdown(table_scan) {
                    return Ok(None);
                }
                let table_schema = table_scan.source.schema();
                let filter_schema = DFSchema::try_from_qualified_schema(
                    table_scan.table_name.clone(),
                    table_schema.as_ref(),
                )?;
                let mut filter_alias_rewriter =
                    alias.as_ref().map(|alias_name| TableAliasRewriter {
                        table_schema: &filter_schema,
                        alias_name: alias_name.clone(),
                        rewrite_unqualified: true,
                    });

                let mut builder = LogicalPlanBuilder::scan(
                    table_scan.table_name.clone(),
                    Arc::clone(&table_scan.source),
                    None,
                )?;
                // We will rebase the column references to the new alias if it exists.
                // If the projection or filters are empty, we will append alias to the table scan.
                //
                // Example:
                //   select t1.c1 from t1 where t1.c1 > 1 -> select a.c1 from t1 as a where a.c1 > 1
                if let Some(ref alias) = alias
                    && (table_scan.projection.is_some() || !table_scan.filters.is_empty())
                {
                    builder = builder.alias(alias.clone())?;
                }

                // Avoid creating a duplicate Projection node, which would result in an additional subquery if a projection already exists.
                // For example, if the `optimize_projection` rule is applied, there will be a Projection node, and duplicate projection
                // information included in the TableScan node.
                if !already_projected && let Some(project_vec) = &table_scan.projection {
                    if project_vec.is_empty() {
                        builder = builder.project(self.empty_projection_fallback())?;
                    } else {
                        let project_columns = project_vec
                            .iter()
                            .cloned()
                            .map(|i| {
                                let schema = table_scan.source.schema();
                                let field = schema.field(i);
                                if alias.is_some() {
                                    Column::new(alias.clone(), field.name().clone())
                                } else {
                                    Column::new(
                                        Some(table_scan.table_name.clone()),
                                        field.name().clone(),
                                    )
                                }
                            })
                            .collect::<Vec<_>>();
                        builder = builder.project(project_columns)?;
                    };
                }

                let filter_expr: Result<Option<Expr>> = table_scan
                    .filters
                    .iter()
                    .cloned()
                    .map(|expr| {
                        if let Some(ref mut rewriter) = filter_alias_rewriter {
                            expr.rewrite(rewriter).data()
                        } else {
                            Ok(expr)
                        }
                    })
                    .reduce(|acc, expr_result| {
                        acc.and_then(|acc_expr| {
                            expr_result.map(|expr| acc_expr.and(expr))
                        })
                    })
                    .transpose();

                if let Some(filter) = filter_expr? {
                    builder = builder.filter(filter)?;
                }

                if let Some(fetch) = table_scan.fetch {
                    builder = builder.limit(0, Some(fetch))?;
                }

                // If the table scan has an alias but no projection or filters, it means no column references are rebased.
                // So we will append the alias to this subquery.
                // Example:
                //   select * from t1 limit 10 -> (select * from t1 limit 10) as a
                if let Some(alias) = alias
                    && table_scan.projection.is_none()
                    && table_scan.filters.is_empty()
                {
                    builder = builder.alias(alias)?;
                }

                Ok(Some(builder.build()?))
            }
            LogicalPlan::SubqueryAlias(subquery_alias) => {
                let ret = self.unparse_table_scan_pushdown(
                    &subquery_alias.input,
                    Some(subquery_alias.alias.clone()),
                    already_projected,
                )?;
                if let Some(alias) = alias
                    && let Some(plan) = ret
                {
                    let plan = LogicalPlanBuilder::new(plan).alias(alias)?.build()?;
                    return Ok(Some(plan));
                }
                Ok(ret)
            }
            // SubqueryAlias could be rewritten to a plan with a projection as the top node by [rewrite::subquery_alias_inner_query_and_columns].
            // The inner table scan could be a scan with pushdown operations.
            LogicalPlan::Projection(projection) => {
                if let Some(plan) = self.unparse_table_scan_pushdown(
                    &projection.input,
                    alias.clone(),
                    already_projected,
                )? {
                    // The pushed-down scan alias is only in scope for the
                    // projection directly above the aliased table scan. `plan`
                    // is the result of pushing the alias further down: if it is
                    // itself a `Projection`, the input was another projection
                    // (e.g. common subexpression elimination stacked one), so
                    // this projection sits over a derived table rather than
                    // directly over the aliased scan, and the alias is out of
                    // scope here. Its qualified pass-through columns must then
                    // reference the derived table's output unqualified instead
                    // of being rebased to the alias. Build it directly so the
                    // unqualified columns are not re-normalized back to the
                    // alias. (Otherwise `plan` is the scan-derived plan and we
                    // fall through to rebase to the alias, correct one level
                    // above the scan.)
                    if alias.is_some() && matches!(plan, LogicalPlan::Projection(_)) {
                        let exprs = projection
                            .expr
                            .iter()
                            .cloned()
                            .map(Self::strip_column_qualifiers)
                            .collect::<Result<Vec<_>>>()?;
                        return Ok(Some(LogicalPlan::Projection(Projection::try_new(
                            exprs,
                            Arc::new(plan),
                        )?)));
                    }

                    let exprs = if alias.is_some() {
                        let mut alias_rewriter =
                            alias.as_ref().map(|alias_name| TableAliasRewriter {
                                table_schema: plan.schema().as_ref(),
                                alias_name: alias_name.clone(),
                                rewrite_unqualified: false,
                            });
                        projection
                            .expr
                            .iter()
                            .cloned()
                            .map(|expr| {
                                if let Some(ref mut rewriter) = alias_rewriter {
                                    expr.rewrite(rewriter).data()
                                } else {
                                    Ok(expr)
                                }
                            })
                            .collect::<Result<Vec<_>>>()?
                    } else {
                        projection.expr.clone()
                    };
                    Ok(Some(
                        LogicalPlanBuilder::from(plan).project(exprs)?.build()?,
                    ))
                } else {
                    Ok(None)
                }
            }
            _ => Ok(None),
        }
    }

    fn select_item_to_sql(&self, expr: &Expr) -> Result<ast::SelectItem> {
        match expr {
            Expr::Alias(Alias { expr, name, .. }) => {
                let inner = self.expr_to_sql(expr)?;

                // Determine the alias name to use
                let col_name = if let Some(rewritten_name) =
                    self.dialect.col_alias_overrides(name)?
                {
                    rewritten_name.to_string()
                } else {
                    name.to_string()
                };

                Ok(ast::SelectItem::ExprWithAlias {
                    expr: inner,
                    alias: self.new_ident_quoted_if_needs(col_name),
                })
            }
            _ => {
                let inner = self.expr_to_sql(expr)?;

                Ok(ast::SelectItem::UnnamedExpr(inner))
            }
        }
    }

    fn sorts_to_sql(&self, sort_exprs: &[SortExpr]) -> Result<OrderByKind> {
        Ok(OrderByKind::Expressions(
            sort_exprs
                .iter()
                .map(|sort_expr| self.sort_to_sql(sort_expr))
                .collect::<Result<Vec<_>>>()?,
        ))
    }

    fn join_operator_to_sql(
        &self,
        join_type: JoinType,
        constraint: ast::JoinConstraint,
    ) -> Result<ast::JoinOperator> {
        Ok(match join_type {
            JoinType::Inner => match &constraint {
                ast::JoinConstraint::On(_)
                | ast::JoinConstraint::Using(_)
                | ast::JoinConstraint::Natural => ast::JoinOperator::Inner(constraint),
                ast::JoinConstraint::None => {
                    // Inner joins with no conditions or filters are not valid SQL in most systems,
                    // return a CROSS JOIN instead
                    ast::JoinOperator::CrossJoin(constraint)
                }
            },
            JoinType::Left => ast::JoinOperator::LeftOuter(constraint),
            JoinType::Right => ast::JoinOperator::RightOuter(constraint),
            JoinType::Full => ast::JoinOperator::FullOuter(constraint),
            JoinType::LeftAnti => ast::JoinOperator::LeftAnti(constraint),
            JoinType::LeftSemi => ast::JoinOperator::LeftSemi(constraint),
            JoinType::RightAnti => ast::JoinOperator::RightAnti(constraint),
            JoinType::RightSemi => ast::JoinOperator::RightSemi(constraint),
            JoinType::LeftMark | JoinType::RightMark => {
                unimplemented!("Unparsing of Mark join type")
            }
        })
    }

    /// Convert the components of a USING clause to the USING AST. Returns
    /// 'None' if the conditions are not compatible with a USING expression,
    /// e.g. non-column expressions or non-matching names.
    fn join_using_to_sql(
        &self,
        join_conditions: &[(Expr, Expr)],
    ) -> Option<ast::JoinConstraint> {
        let mut object_names = Vec::with_capacity(join_conditions.len());
        for (left, right) in join_conditions {
            match (left, right) {
                (
                    Expr::Column(Column {
                        relation: _,
                        name: left_name,
                        spans: _,
                    }),
                    Expr::Column(Column {
                        relation: _,
                        name: right_name,
                        spans: _,
                    }),
                ) if left_name == right_name => {
                    // For example, if the join condition `t1.id = t2.id`
                    // this is represented as two columns like `[t1.id, t2.id]`
                    // This code forms `id` (without relation name)
                    let ident = self.new_ident_quoted_if_needs(left_name.to_string());
                    object_names.push(ast::ObjectName::from(vec![ident]));
                }
                // USING is only valid with matching column names; arbitrary expressions
                // are not allowed
                _ => return None,
            }
        }
        Some(ast::JoinConstraint::Using(object_names))
    }

    /// Convert a join constraint and associated conditions and filter to a SQL AST node
    fn join_constraint_to_sql(
        &self,
        constraint: JoinConstraint,
        conditions: &[(Expr, Expr)],
        filter: Option<&Expr>,
    ) -> Result<ast::JoinConstraint> {
        match (constraint, conditions, filter) {
            // No constraints
            (JoinConstraint::On | JoinConstraint::Using, [], None) => {
                Ok(ast::JoinConstraint::None)
            }

            (JoinConstraint::Using, conditions, None) => {
                match self.join_using_to_sql(conditions) {
                    Some(using) => Ok(using),
                    // As above, this should not be reachable from parsed SQL,
                    // but a user could create this; we "downgrade" to ON.
                    None => self.join_conditions_to_sql_on(conditions, None),
                }
            }

            // Two cases here:
            // 1. Straightforward ON case, with possible equi-join conditions
            //    and additional filters
            // 2. USING with additional filters; we "downgrade" to ON, because
            //    you can't use USING with arbitrary filters. (This should not
            //    be accessible from parsed SQL, but may have been a
            //    custom-built JOIN by a user.)
            (JoinConstraint::On | JoinConstraint::Using, conditions, filter) => {
                self.join_conditions_to_sql_on(conditions, filter)
            }
        }
    }

    // Convert a list of equi0join conditions and an optional filter to a SQL ON
    // AST node, with the equi-join conditions and the filter merged into a
    // single conditional expression
    fn join_conditions_to_sql_on(
        &self,
        join_conditions: &[(Expr, Expr)],
        filter: Option<&Expr>,
    ) -> Result<ast::JoinConstraint> {
        let mut condition = None;
        // AND the join conditions together to create the overall condition
        for (left, right) in join_conditions {
            // Parse left and right
            let l = self.expr_to_sql(left)?;
            let r = self.expr_to_sql(right)?;
            let e = self.binary_op_to_sql(l, r, ast::BinaryOperator::Eq);
            condition = match condition {
                Some(expr) => Some(self.and_op_to_sql(expr, e)),
                None => Some(e),
            };
        }

        // Then AND the non-equijoin filter condition as well
        condition = match (condition, filter) {
            (Some(expr), Some(filter)) => {
                Some(self.and_op_to_sql(expr, self.expr_to_sql(filter)?))
            }
            (Some(expr), None) => Some(expr),
            (None, Some(filter)) => Some(self.expr_to_sql(filter)?),
            (None, None) => None,
        };

        let constraint = match condition {
            Some(filter) => ast::JoinConstraint::On(filter),
            None => ast::JoinConstraint::None,
        };

        Ok(constraint)
    }

    fn and_op_to_sql(&self, lhs: ast::Expr, rhs: ast::Expr) -> ast::Expr {
        self.binary_op_to_sql(lhs, rhs, ast::BinaryOperator::And)
    }

    fn new_table_alias(&self, alias: String, columns: Vec<Ident>) -> ast::TableAlias {
        let columns = columns
            .into_iter()
            .map(|ident| TableAliasColumnDef {
                name: ident,
                data_type: None,
            })
            .collect();
        ast::TableAlias {
            name: self.new_ident_quoted_if_needs(alias),
            columns,
            explicit: true,
            at: None,
        }
    }

    fn dml_to_sql(&self, plan: &LogicalPlan) -> Result<ast::Statement> {
        not_impl_err!("Unsupported plan: {plan:?}")
    }

    /// Generates appropriate projection expression for empty projection lists.
    /// Returns an empty vec for dialects supporting empty select lists,
    /// or a dummy literal `1` for other dialects.
    fn empty_projection_fallback(&self) -> Vec<Expr> {
        if self.dialect.supports_empty_select_list() {
            Vec::new()
        } else {
            vec![Expr::Literal(ScalarValue::Int64(Some(1)), None)]
        }
    }

    /// Decides where extracted table-scan filters belong in the unparsed SQL:
    /// in the `JOIN ON` clause or in `WHERE`.
    ///
    /// For inner joins the two are semantically equivalent, so filters go to
    /// `WHERE` (some dialects reject subqueries inside `JOIN ON`).
    /// For outer joins the filters are AND-folded into `ON` to preserve correctness.
    ///
    /// Returns `(on_filter, where_filters)`.
    fn split_join_on_and_where_filters(
        join_type: JoinType,
        join_filter: &Option<Expr>,
        table_scan_filters: Vec<Expr>,
    ) -> (Option<Expr>, Vec<Expr>) {
        if table_scan_filters.is_empty() {
            return (join_filter.clone(), vec![]);
        }

        if join_type == JoinType::Inner {
            // ON and WHERE are equivalent for inner joins; prefer WHERE
            // because some dialects reject subqueries inside JOIN ON.
            return (join_filter.clone(), table_scan_filters);
        }

        // Outer joins: fold table-scan filters into ON to preserve semantics.
        let combined = table_scan_filters.into_iter().reduce(|acc, filter| {
            Expr::BinaryExpr(BinaryExpr {
                left: Box::new(acc),
                op: Operator::And,
                right: Box::new(filter),
            })
        });

        let on_filter = match (join_filter, combined) {
            (Some(jf), Some(c)) => Some(Expr::BinaryExpr(BinaryExpr {
                left: Box::new(jf.clone()),
                op: Operator::And,
                right: Box::new(c),
            })),
            (Some(jf), None) => Some(jf.clone()),
            (None, Some(c)) => Some(c),
            (None, None) => None,
        };

        (on_filter, vec![])
    }
}

impl From<BuilderError> for DataFusionError {
    fn from(e: BuilderError) -> Self {
        DataFusionError::External(Box::new(e))
    }
}

/// The type of the input to the UNNEST table factor.
#[derive(Debug)]
enum UnnestInputType {
    /// The input is a column reference. It will be presented like `outer_ref(column_name)`.
    OuterReference,
    /// The input is a scalar value. It will be presented like a scalar array or struct.
    Scalar,
}