polyglot_sql/

resolver.rs

//! Column Resolver Module
//!
//! This module provides functionality for resolving column references to their
//! source tables. It handles:
//! - Finding which table a column belongs to
//! - Resolving ambiguous column references
//! - Handling join context for disambiguation
//! - Supporting set operations (UNION, INTERSECT, EXCEPT)
//!
//! Based on the Python implementation in `sqlglot/optimizer/resolver.py`.

use crate::dialects::DialectType;
use crate::expressions::{Expression, Identifier, TableRef};
#[cfg(feature = "generate")]
use crate::generator::Generator;
use crate::schema::{normalize_name, Schema};
use crate::scope::{Scope, SourceInfo};
use crate::traversal::ExpressionWalk;
use std::collections::{HashMap, HashSet};
use thiserror::Error;

/// Errors that can occur during column resolution
#[derive(Debug, Error, Clone)]
pub enum ResolverError {
    #[error("Unknown table: {0}")]
    UnknownTable(String),

    #[error("Ambiguous column: {column} appears in multiple sources: {sources}")]
    AmbiguousColumn { column: String, sources: String },

    #[error("Column not found: {0}")]
    ColumnNotFound(String),

    #[error("Unknown set operation: {0}")]
    UnknownSetOperation(String),

    #[error("Invalid set operation: {0}")]
    InvalidSetOperation(String),
}

/// Result type for resolver operations
pub type ResolverResult<T> = Result<T, ResolverError>;

/// Helper for resolving columns to their source tables.
///
/// This is a struct so we can lazily load some things and easily share
/// them across functions.
pub struct Resolver<'a> {
    /// The scope being analyzed
    pub scope: &'a Scope,
    /// The schema for table/column information
    schema: &'a dyn Schema,
    /// The dialect being used
    pub dialect: Option<DialectType>,
    /// Whether to infer schema from context
    infer_schema: bool,
    /// Cached source columns: source_name -> column names
    source_columns_cache: HashMap<String, Vec<String>>,
    /// Cached unambiguous columns: column_name -> source_name
    unambiguous_columns_cache: Option<HashMap<String, String>>,
    /// Cached set of all available columns
    all_columns_cache: Option<HashSet<String>>,
}

impl<'a> Resolver<'a> {
    /// Create a new resolver for a scope
    pub fn new(scope: &'a Scope, schema: &'a dyn Schema, infer_schema: bool) -> Self {
        Self {
            scope,
            schema,
            dialect: schema.dialect(),
            infer_schema,
            source_columns_cache: HashMap::new(),
            unambiguous_columns_cache: None,
            all_columns_cache: None,
        }
    }

    /// Get the table for a column name.
    ///
    /// Returns the table name if it can be found/inferred.
    pub fn get_table(&mut self, column_name: &str) -> Option<String> {
        // Try to find table from all sources (unambiguous lookup)
        let table_name = self.get_table_name_from_sources(column_name, None);

        // If we found a table, return it
        if table_name.is_some() {
            return table_name;
        }

        // If schema inference is enabled and exactly one source has no schema,
        // assume the column belongs to that source
        if self.infer_schema {
            let sources_without_schema: Vec<_> = self
                .get_all_source_columns()
                .iter()
                .filter(|(_, columns)| columns.is_empty() || columns.contains(&"*".to_string()))
                .map(|(name, _)| name.clone())
                .collect();

            if sources_without_schema.len() == 1 {
                return Some(sources_without_schema[0].clone());
            }
        }

        None
    }

    /// Get the table for a column, returning an Identifier
    pub fn get_table_identifier(&mut self, column_name: &str) -> Option<Identifier> {
        self.get_table(column_name).map(Identifier::new)
    }

    /// Check if a table exists in the schema (not necessarily in the current scope).
    /// Used to detect correlated references to outer scope tables.
    pub fn table_exists_in_schema(&self, table_name: &str) -> bool {
        self.schema.column_names(table_name).is_ok()
    }

    /// Find the table for a column by searching all schema tables not in the current scope.
    /// Used for correlated subquery resolution: if an unqualified column can't be resolved
    /// in the current scope, check if it uniquely belongs to an outer-scope table.
    /// Returns Some(table_name) if the column is found in exactly one non-local table.
    pub fn find_column_in_outer_schema_tables(&self, column_name: &str) -> Option<String> {
        let tables = self.schema.find_tables_for_column(column_name);
        // Filter to tables NOT in the current scope
        let outer_tables: Vec<String> = tables
            .into_iter()
            .filter(|t| !self.scope.sources.contains_key(t))
            .collect();
        // Only return if unambiguous (exactly one outer table has this column)
        if outer_tables.len() == 1 {
            Some(outer_tables.into_iter().next().unwrap())
        } else {
            None
        }
    }

    /// Get all available columns across all sources in this scope
    pub fn all_columns(&mut self) -> &HashSet<String> {
        if self.all_columns_cache.is_none() {
            let mut all = HashSet::new();
            for columns in self.get_all_source_columns().values() {
                all.extend(columns.iter().cloned());
            }
            self.all_columns_cache = Some(all);
        }
        self.all_columns_cache
            .as_ref()
            .expect("cache populated above")
    }

    /// Get column names for a source.
    ///
    /// Returns the list of column names available from the given source.
    pub fn get_source_columns(&mut self, source_name: &str) -> ResolverResult<Vec<String>> {
        // Check cache first
        if let Some(columns) = self.source_columns_cache.get(source_name) {
            return Ok(columns.clone());
        }

        // Get the source info
        let source_info = self
            .scope
            .sources
            .get(source_name)
            .ok_or_else(|| ResolverError::UnknownTable(source_name.to_string()))?;

        let columns = self.extract_columns_from_source(source_info)?;

        // Cache the result
        self.source_columns_cache
            .insert(source_name.to_string(), columns.clone());

        Ok(columns)
    }

    /// Extract column names from a source expression
    fn extract_columns_from_source(&self, source_info: &SourceInfo) -> ResolverResult<Vec<String>> {
        self.get_source_columns_for_expression(&source_info.expression)
    }

    fn get_source_columns_for_expression(
        &self,
        expression: &Expression,
    ) -> ResolverResult<Vec<String>> {
        let columns = match expression {
            Expression::Table(table) => {
                // For tables, try to get columns from schema.
                // Build the fully qualified name (catalog.schema.table) to
                // match how MappingSchema stores hierarchical keys.
                let table_name = qualified_table_name(table);
                match self.schema.column_names(&table_name) {
                    Ok(cols) => cols,
                    Err(_) => Vec::new(), // Schema might not have this table
                }
            }
            Expression::Subquery(subquery) => {
                // For subqueries, get named_selects from the inner query
                self.get_named_selects(&subquery.this)
            }
            Expression::Select(select) => {
                // For derived tables that are SELECT expressions
                self.get_select_column_names(select)
            }
            Expression::Union(union) => {
                // For UNION, columns come from the set operation
                self.get_source_columns_from_set_op(&Expression::Union(union.clone()))?
            }
            Expression::Intersect(intersect) => {
                self.get_source_columns_from_set_op(&Expression::Intersect(intersect.clone()))?
            }
            Expression::Except(except) => {
                self.get_source_columns_from_set_op(&Expression::Except(except.clone()))?
            }
            Expression::Cte(cte) => {
                if !cte.columns.is_empty() {
                    cte.columns.iter().map(|c| c.name.clone()).collect()
                } else {
                    self.get_named_selects(&cte.this)
                }
            }
            Expression::Pivot(pivot) => self.get_pivot_output_columns(pivot),
            Expression::Unpivot(unpivot) => self.get_unpivot_output_columns(unpivot),
            Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) => {
                alias_output_columns(alias)
            }
            Expression::Alias(alias) => {
                let columns = self.get_source_columns_for_expression(&alias.this)?;
                apply_alias_columns(columns, &alias.column_aliases)
            }
            Expression::Unnest(unnest) => unnest_output_columns(unnest),
            Expression::Lateral(lateral) => lateral_output_columns(lateral),
            Expression::LateralView(lateral_view) => lateral_view_output_columns(lateral_view),
            Expression::Paren(paren) => self.get_source_columns_for_expression(&paren.this)?,
            _ => Vec::new(),
        };

        Ok(columns)
    }

    /// Get named selects (column names) from an expression
    fn get_named_selects(&self, expr: &Expression) -> Vec<String> {
        if let Ok(identifiers) = crate::set_operation::query_output_identifiers(expr, self.dialect)
        {
            return identifiers
                .into_iter()
                .map(|identifier| identifier.name)
                .collect();
        }

        match expr {
            Expression::Select(select) => self.get_select_column_names(select),
            Expression::Union(union) => {
                // For unions, use the left side's columns
                self.get_named_selects(&union.left)
            }
            Expression::Intersect(intersect) => self.get_named_selects(&intersect.left),
            Expression::Except(except) => self.get_named_selects(&except.left),
            Expression::Subquery(subquery) => self.get_named_selects(&subquery.this),
            Expression::Alias(alias) => {
                let columns = self.get_named_selects(&alias.this);
                apply_alias_columns(columns, &alias.column_aliases)
            }
            Expression::Paren(paren) => self.get_named_selects(&paren.this),
            _ => Vec::new(),
        }
    }

    /// Get column names from a SELECT expression
    fn get_select_column_names(&self, select: &crate::expressions::Select) -> Vec<String> {
        select
            .expressions
            .iter()
            .filter_map(|expr| self.get_expression_alias(expr))
            .collect()
    }

    /// Get the alias or name for a select expression
    fn get_expression_alias(&self, expr: &Expression) -> Option<String> {
        match expr {
            Expression::Alias(alias) => Some(alias.alias.name.clone()),
            Expression::Column(col) => Some(col.name.name.clone()),
            Expression::Star(_) => Some("*".to_string()),
            Expression::Identifier(id) => Some(id.name.clone()),
            _ => None,
        }
    }

    fn get_pivot_output_columns(&self, pivot: &crate::expressions::Pivot) -> Vec<String> {
        if pivot.unpivot {
            return self.get_pivot_unpivot_output_columns(pivot);
        }

        let pre_columns = self.get_source_output_columns(&pivot.this);
        if pre_columns.is_empty() || pre_columns.iter().any(|column| column == "*") {
            return Vec::new();
        }

        let excluded = pivot_excluded_source_columns(pivot, self.dialect);
        let generated = pivot_generated_output_columns(pivot, self.dialect);
        if excluded.is_empty() || generated.is_empty() {
            return Vec::new();
        }

        let mut columns: Vec<String> = pre_columns
            .into_iter()
            .filter(|column| !excluded.contains(&normalize_column_name(column, self.dialect)))
            .collect();
        columns.extend(generated);
        apply_alias_columns(columns, &pivot.alias_columns)
    }

    fn get_pivot_unpivot_output_columns(&self, pivot: &crate::expressions::Pivot) -> Vec<String> {
        let pre_columns = self.get_source_output_columns(&pivot.this);
        if pre_columns.is_empty() || pre_columns.iter().any(|column| column == "*") {
            return Vec::new();
        }

        let input_columns: HashSet<String> = pivot
            .expressions
            .iter()
            .flat_map(expression_column_names)
            .map(|column| normalize_column_name(&column, self.dialect))
            .collect();
        let mut columns: Vec<String> = pre_columns
            .into_iter()
            .filter(|column| !input_columns.contains(&normalize_column_name(column, self.dialect)))
            .collect();

        if let Some(Expression::UnpivotColumns(unpivot_columns)) = pivot.into.as_deref() {
            if let Some(name) = expression_name(&unpivot_columns.this) {
                columns.push(name);
            }
            for value_column in &unpivot_columns.expressions {
                if let Some(name) = expression_name(value_column) {
                    columns.push(name);
                }
            }
        }

        apply_alias_columns(columns, &pivot.alias_columns)
    }

    fn get_unpivot_output_columns(&self, unpivot: &crate::expressions::Unpivot) -> Vec<String> {
        let pre_columns = self.get_source_output_columns(&unpivot.this);
        if pre_columns.is_empty() || pre_columns.iter().any(|column| column == "*") {
            return Vec::new();
        }

        let input_columns: HashSet<String> = unpivot
            .columns
            .iter()
            .flat_map(expression_column_names)
            .map(|column| normalize_column_name(&column, self.dialect))
            .collect();
        let mut columns: Vec<String> = pre_columns
            .into_iter()
            .filter(|column| !input_columns.contains(&normalize_column_name(column, self.dialect)))
            .collect();
        columns.push(unpivot.name_column.name.clone());
        columns.push(unpivot.value_column.name.clone());
        columns.extend(
            unpivot
                .extra_value_columns
                .iter()
                .map(|column| column.name.clone()),
        );
        apply_alias_columns(columns, &unpivot.alias_columns)
    }

    fn get_source_output_columns(&self, source: &Expression) -> Vec<String> {
        match source {
            Expression::Table(table) => {
                if table.schema.is_none() && table.catalog.is_none() {
                    if let Some(source) = self.scope.cte_sources.get(&table.name.name) {
                        return self.extract_columns_from_source(source).unwrap_or_default();
                    }
                }

                let table_name = qualified_table_name(table);
                self.schema.column_names(&table_name).unwrap_or_default()
            }
            Expression::Subquery(subquery) => self.get_named_selects(&subquery.this),
            Expression::Select(select) => self.get_select_column_names(select),
            Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_) => self
                .get_source_columns_from_set_op(source)
                .unwrap_or_default(),
            Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) => {
                alias_output_columns(alias)
            }
            Expression::Alias(alias) => {
                let columns = self.get_source_output_columns(&alias.this);
                apply_alias_columns(columns, &alias.column_aliases)
            }
            Expression::Unnest(unnest) => unnest_output_columns(unnest),
            Expression::Lateral(lateral) => lateral_output_columns(lateral),
            Expression::LateralView(lateral_view) => lateral_view_output_columns(lateral_view),
            Expression::Cte(cte) => {
                if cte.columns.is_empty() {
                    self.get_named_selects(&cte.this)
                } else {
                    cte.columns
                        .iter()
                        .map(|column| column.name.clone())
                        .collect()
                }
            }
            Expression::Paren(paren) => self.get_source_output_columns(&paren.this),
            _ => Vec::new(),
        }
    }

    /// Get columns from a set operation (UNION, INTERSECT, EXCEPT)
    pub fn get_source_columns_from_set_op(
        &self,
        expression: &Expression,
    ) -> ResolverResult<Vec<String>> {
        if matches!(
            expression,
            Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_)
        ) {
            match crate::set_operation::set_operation_layout(expression, self.dialect) {
                Ok(Some(layout)) => {
                    return Ok(layout
                        .outputs
                        .into_iter()
                        .map(|output| output.identifier.name)
                        .collect())
                }
                Ok(None) => {}
                Err(error) if error.is_indeterminate() => {}
                Err(error) => return Err(ResolverError::InvalidSetOperation(error.to_string())),
            }
        }

        match expression {
            Expression::Select(select) => Ok(self.get_select_column_names(select)),
            Expression::Subquery(subquery) => {
                if matches!(
                    &subquery.this,
                    Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_)
                ) {
                    self.get_source_columns_from_set_op(&subquery.this)
                } else {
                    Ok(self.get_named_selects(&subquery.this))
                }
            }
            Expression::Alias(alias) => {
                let columns = self.get_source_columns_from_set_op(&alias.this)?;
                Ok(apply_alias_columns(columns, &alias.column_aliases))
            }
            Expression::Paren(paren) => self.get_source_columns_from_set_op(&paren.this),
            Expression::Union(union) => {
                // Standard UNION: columns come from the left side
                self.get_source_columns_from_set_op(&union.left)
            }
            Expression::Intersect(intersect) => {
                self.get_source_columns_from_set_op(&intersect.left)
            }
            Expression::Except(except) => self.get_source_columns_from_set_op(&except.left),
            _ => Err(ResolverError::UnknownSetOperation(format!(
                "{:?}",
                expression
            ))),
        }
    }

    /// Get all source columns for all sources in the scope
    fn get_all_source_columns(&mut self) -> HashMap<String, Vec<String>> {
        let source_names: Vec<_> = self.scope.sources.keys().cloned().collect();

        let mut result = HashMap::new();
        for source_name in source_names {
            if let Ok(columns) = self.get_source_columns(&source_name) {
                result.insert(source_name, columns);
            }
        }
        result
    }

    /// Get the table name for a column from the sources
    fn get_table_name_from_sources(
        &mut self,
        column_name: &str,
        source_columns: Option<&HashMap<String, Vec<String>>>,
    ) -> Option<String> {
        let normalized_column_name = normalize_column_name(column_name, self.dialect);
        let unambiguous = match source_columns {
            Some(cols) => self.compute_unambiguous_columns(cols),
            None => {
                if self.unambiguous_columns_cache.is_none() {
                    let all_source_columns = self.get_all_source_columns();
                    self.unambiguous_columns_cache =
                        Some(self.compute_unambiguous_columns(&all_source_columns));
                }
                self.unambiguous_columns_cache
                    .clone()
                    .expect("cache populated above")
            }
        };

        unambiguous.get(&normalized_column_name).cloned()
    }

    /// Compute unambiguous columns mapping
    ///
    /// A column is unambiguous if it appears in exactly one source.
    fn compute_unambiguous_columns(
        &self,
        source_columns: &HashMap<String, Vec<String>>,
    ) -> HashMap<String, String> {
        if source_columns.is_empty() {
            return HashMap::new();
        }

        let mut column_to_sources: HashMap<String, Vec<String>> = HashMap::new();

        for (source_name, columns) in source_columns {
            for column in columns {
                column_to_sources
                    .entry(normalize_column_name(column, self.dialect))
                    .or_default()
                    .push(source_name.clone());
            }
        }

        // Keep only columns that appear in exactly one source
        column_to_sources
            .into_iter()
            .filter(|(_, sources)| sources.len() == 1)
            .map(|(column, sources)| (column, sources.into_iter().next().unwrap()))
            .collect()
    }

    /// Check if a column is ambiguous (appears in multiple sources)
    pub fn is_ambiguous(&mut self, column_name: &str) -> bool {
        let normalized_column_name = normalize_column_name(column_name, self.dialect);
        let all_source_columns = self.get_all_source_columns();
        let sources_with_column: Vec<_> = all_source_columns
            .iter()
            .filter(|(_, columns)| {
                columns.iter().any(|column| {
                    normalize_column_name(column, self.dialect) == normalized_column_name
                })
            })
            .map(|(name, _)| name.clone())
            .collect();

        sources_with_column.len() > 1
    }

    /// Get all sources that contain a given column
    pub fn sources_for_column(&mut self, column_name: &str) -> Vec<String> {
        let normalized_column_name = normalize_column_name(column_name, self.dialect);
        let all_source_columns = self.get_all_source_columns();
        all_source_columns
            .iter()
            .filter(|(_, columns)| {
                columns.iter().any(|column| {
                    normalize_column_name(column, self.dialect) == normalized_column_name
                })
            })
            .map(|(name, _)| name.clone())
            .collect()
    }

    /// Try to disambiguate a column based on join context
    ///
    /// In join conditions, a column can sometimes be disambiguated based on
    /// which tables have been joined up to that point.
    pub fn disambiguate_in_join_context(
        &mut self,
        column_name: &str,
        available_sources: &[String],
    ) -> Option<String> {
        let normalized_column_name = normalize_column_name(column_name, self.dialect);
        let mut matching_sources = Vec::new();

        for source_name in available_sources {
            if let Ok(columns) = self.get_source_columns(source_name) {
                if columns.iter().any(|column| {
                    normalize_column_name(column, self.dialect) == normalized_column_name
                }) {
                    matching_sources.push(source_name.clone());
                }
            }
        }

        if matching_sources.len() == 1 {
            Some(matching_sources.remove(0))
        } else {
            None
        }
    }
}

fn normalize_column_name(name: &str, dialect: Option<DialectType>) -> String {
    normalize_name(name, dialect, false, true)
}

fn apply_alias_columns(mut columns: Vec<String>, alias_columns: &[Identifier]) -> Vec<String> {
    for (idx, alias) in alias_columns.iter().enumerate() {
        if let Some(column) = columns.get_mut(idx) {
            *column = alias.name.clone();
        }
    }
    columns
}

fn unnest_output_columns(unnest: &crate::expressions::UnnestFunc) -> Vec<String> {
    unnest
        .alias
        .iter()
        .map(|alias| alias.name.clone())
        .chain(unnest.offset_alias.iter().map(|alias| alias.name.clone()))
        .collect()
}

fn alias_output_columns(alias: &crate::expressions::Alias) -> Vec<String> {
    if alias.column_aliases.is_empty() {
        vec![alias.alias.name.clone()]
    } else {
        alias
            .column_aliases
            .iter()
            .map(|column| column.name.clone())
            .collect()
    }
}

fn lateral_output_columns(lateral: &crate::expressions::Lateral) -> Vec<String> {
    if lateral.column_aliases.is_empty() {
        default_virtual_output_columns(&lateral.this)
    } else {
        lateral.column_aliases.clone()
    }
}

fn lateral_view_output_columns(lateral_view: &crate::expressions::LateralView) -> Vec<String> {
    lateral_view
        .column_aliases
        .iter()
        .map(|column| column.name.clone())
        .collect()
}

fn default_virtual_output_columns(expression: &Expression) -> Vec<String> {
    match expression {
        Expression::Unnest(unnest) => unnest_output_columns(unnest),
        Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) => {
            alias_output_columns(alias)
        }
        Expression::Function(function) if function.name.eq_ignore_ascii_case("FLATTEN") => {
            ["seq", "key", "path", "index", "value", "this"]
                .into_iter()
                .map(String::from)
                .collect()
        }
        _ => Vec::new(),
    }
}

fn pivot_excluded_source_columns(
    pivot: &crate::expressions::Pivot,
    dialect: Option<DialectType>,
) -> HashSet<String> {
    pivot
        .fields
        .iter()
        .chain(pivot.expressions.iter())
        .chain(pivot.using.iter())
        .flat_map(expression_column_names)
        .map(|column| normalize_column_name(&column, dialect))
        .collect()
}

fn pivot_generated_output_columns(
    pivot: &crate::expressions::Pivot,
    _dialect: Option<DialectType>,
) -> Vec<String> {
    let fields = pivot_field_output_names(pivot);
    let aggregations = if pivot.using.is_empty() {
        &pivot.expressions
    } else {
        &pivot.using
    };

    if fields.is_empty() || aggregations.is_empty() {
        return Vec::new();
    }

    let needs_suffix = aggregations.len() > 1;
    let mut outputs = Vec::new();
    for field in fields {
        for aggregation in aggregations {
            if let Some(suffix) = pivot_aggregation_output_suffix(aggregation, needs_suffix) {
                outputs.push(format!("{field}_{suffix}"));
            } else {
                outputs.push(field.clone());
            }
        }
    }
    outputs
}

fn pivot_field_output_names(pivot: &crate::expressions::Pivot) -> Vec<String> {
    pivot
        .fields
        .iter()
        .filter_map(|field| match field {
            Expression::In(in_expr) => Some(
                in_expr
                    .expressions
                    .iter()
                    .filter_map(expression_name)
                    .collect::<Vec<_>>(),
            ),
            _ => None,
        })
        .flatten()
        .collect()
}

fn pivot_aggregation_output_suffix(expr: &Expression, needs_suffix: bool) -> Option<String> {
    match expr {
        Expression::Alias(alias) => Some(alias.alias.name.clone()),
        _ if needs_suffix => pivot_generated_aggregation_suffix(expr),
        _ => None,
    }
}

#[cfg(feature = "generate")]
fn pivot_generated_aggregation_suffix(expr: &Expression) -> Option<String> {
    Generator::sql(expr).ok().map(|sql| sql.to_lowercase())
}

#[cfg(not(feature = "generate"))]
fn pivot_generated_aggregation_suffix(expr: &Expression) -> Option<String> {
    expression_name(expr).or_else(|| Some(expr.variant_name().to_string()))
}

fn expression_name(expr: &Expression) -> Option<String> {
    match expr {
        Expression::PivotAlias(alias) => expression_name(&alias.alias),
        Expression::Alias(alias) => Some(alias.alias.name.clone()),
        Expression::Identifier(identifier) => Some(identifier.name.clone()),
        Expression::Column(column) => Some(column.name.name.clone()),
        Expression::Literal(literal) => Some(literal.value_str().to_string()),
        Expression::Var(var) => Some(var.this.clone()),
        Expression::Tuple(tuple) => tuple.expressions.first().and_then(expression_name),
        _ => None,
    }
}

fn expression_column_names(expr: &Expression) -> Vec<String> {
    expr.find_all(|node| matches!(node, Expression::Column(_)))
        .into_iter()
        .filter_map(|node| match node {
            Expression::Column(column) => Some(column.name.name.clone()),
            _ => None,
        })
        .collect()
}

/// Resolve a column to its source table.
///
/// This is a convenience function that creates a Resolver and calls get_table.
pub fn resolve_column(
    scope: &Scope,
    schema: &dyn Schema,
    column_name: &str,
    infer_schema: bool,
) -> Option<String> {
    let mut resolver = Resolver::new(scope, schema, infer_schema);
    resolver.get_table(column_name)
}

/// Check if a column is ambiguous in the given scope.
pub fn is_column_ambiguous(scope: &Scope, schema: &dyn Schema, column_name: &str) -> bool {
    let mut resolver = Resolver::new(scope, schema, true);
    resolver.is_ambiguous(column_name)
}

/// Build the fully qualified table name (catalog.schema.table) from a TableRef.
fn qualified_table_name(table: &TableRef) -> String {
    let mut parts = Vec::new();
    if let Some(catalog) = &table.catalog {
        parts.push(catalog.name.clone());
    }
    if let Some(schema) = &table.schema {
        parts.push(schema.name.clone());
    }
    parts.push(table.name.name.clone());
    parts.join(".")
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dialects::Dialect;
    use crate::expressions::DataType;
    use crate::parser::Parser;
    use crate::schema::MappingSchema;
    use crate::scope::build_scope;

    fn create_test_schema() -> MappingSchema {
        let mut schema = MappingSchema::new();
        // Add tables with columns
        schema
            .add_table(
                "users",
                &[
                    (
                        "id".to_string(),
                        DataType::Int {
                            length: None,
                            integer_spelling: false,
                        },
                    ),
                    ("name".to_string(), DataType::Text),
                    ("email".to_string(), DataType::Text),
                ],
                None,
            )
            .unwrap();
        schema
            .add_table(
                "orders",
                &[
                    (
                        "id".to_string(),
                        DataType::Int {
                            length: None,
                            integer_spelling: false,
                        },
                    ),
                    (
                        "user_id".to_string(),
                        DataType::Int {
                            length: None,
                            integer_spelling: false,
                        },
                    ),
                    (
                        "amount".to_string(),
                        DataType::Double {
                            precision: None,
                            scale: None,
                        },
                    ),
                ],
                None,
            )
            .unwrap();
        schema
    }

    #[test]
    fn test_resolver_basic() {
        let ast = Parser::parse_sql("SELECT id, name FROM users").expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let mut resolver = Resolver::new(&scope, &schema, true);

        // 'name' should resolve to 'users' since it's the only source
        let table = resolver.get_table("name");
        assert_eq!(table, Some("users".to_string()));
    }

    #[test]
    fn test_resolver_ambiguous_column() {
        let ast =
            Parser::parse_sql("SELECT id FROM users JOIN orders ON users.id = orders.user_id")
                .expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let mut resolver = Resolver::new(&scope, &schema, true);

        // 'id' appears in both tables, so it's ambiguous
        assert!(resolver.is_ambiguous("id"));

        // 'name' only appears in users
        assert!(!resolver.is_ambiguous("name"));

        // 'amount' only appears in orders
        assert!(!resolver.is_ambiguous("amount"));
    }

    #[test]
    fn test_resolver_unambiguous_column() {
        let ast = Parser::parse_sql(
            "SELECT name, amount FROM users JOIN orders ON users.id = orders.user_id",
        )
        .expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let mut resolver = Resolver::new(&scope, &schema, true);

        // 'name' should resolve to 'users'
        let table = resolver.get_table("name");
        assert_eq!(table, Some("users".to_string()));

        // 'amount' should resolve to 'orders'
        let table = resolver.get_table("amount");
        assert_eq!(table, Some("orders".to_string()));
    }

    #[test]
    fn test_resolver_with_alias() {
        let ast = Parser::parse_sql("SELECT u.id FROM users AS u").expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let _resolver = Resolver::new(&scope, &schema, true);

        // Source should be indexed by alias 'u'
        assert!(scope.sources.contains_key("u"));
    }

    #[test]
    fn test_sources_for_column() {
        let ast = Parser::parse_sql("SELECT * FROM users JOIN orders ON users.id = orders.user_id")
            .expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let mut resolver = Resolver::new(&scope, &schema, true);

        // 'id' should be in both users and orders
        let sources = resolver.sources_for_column("id");
        assert!(sources.contains(&"users".to_string()));
        assert!(sources.contains(&"orders".to_string()));

        // 'email' should only be in users
        let sources = resolver.sources_for_column("email");
        assert_eq!(sources, vec!["users".to_string()]);
    }

    #[test]
    fn test_all_columns() {
        let ast = Parser::parse_sql("SELECT * FROM users").expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let mut resolver = Resolver::new(&scope, &schema, true);

        let all = resolver.all_columns();
        assert!(all.contains("id"));
        assert!(all.contains("name"));
        assert!(all.contains("email"));
    }

    #[test]
    fn test_resolver_cte_projected_alias_column() {
        let ast = Parser::parse_sql(
            "WITH my_cte AS (SELECT id AS emp_id FROM users) SELECT emp_id FROM my_cte",
        )
        .expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();
        let mut resolver = Resolver::new(&scope, &schema, true);

        let table = resolver.get_table("emp_id");
        assert_eq!(table, Some("my_cte".to_string()));
    }

    #[test]
    fn test_resolver_uses_name_aligned_set_operation_outputs() {
        for dialect_type in [DialectType::DuckDB, DialectType::Snowflake] {
            let expr = Dialect::get(dialect_type)
                .parse(
                    "SELECT combined.right_value FROM (\
                     SELECT 1 AS left_value UNION ALL BY NAME \
                     SELECT 2 AS right_value\
                     ) AS combined",
                )
                .expect("parse")
                .remove(0);
            let scope = build_scope(&expr);
            let schema = MappingSchema::with_dialect(dialect_type);
            let mut resolver = Resolver::new(&scope, &schema, true);

            assert_eq!(
                resolver.get_table("right_value"),
                Some("combined".to_string()),
                "right-only output was not exposed for {dialect_type:?}"
            );
        }

        let dialect_type = DialectType::BigQuery;
        let expr = Dialect::get(dialect_type)
            .parse(
                "SELECT combined.c FROM (\
                 SELECT 1 AS a, 2 AS b FULL OUTER UNION ALL BY NAME \
                 SELECT 3 AS b, 4 AS c\
                 ) AS combined",
            )
            .expect("parse")
            .remove(0);
        let scope = build_scope(&expr);
        let schema = MappingSchema::with_dialect(dialect_type);
        let mut resolver = Resolver::new(&scope, &schema, true);
        assert_eq!(resolver.get_table("c"), Some("combined".to_string()));
    }

    #[test]
    fn test_resolve_column_helper() {
        let ast = Parser::parse_sql("SELECT name FROM users").expect("Failed to parse");
        let scope = build_scope(&ast[0]);
        let schema = create_test_schema();

        let table = resolve_column(&scope, &schema, "name", true);
        assert_eq!(table, Some("users".to_string()));
    }

    #[test]
    fn test_resolver_bigquery_mixed_case_column_names() {
        let dialect = Dialect::get(DialectType::BigQuery);
        let expr = dialect
            .parse("SELECT Name AS name FROM teams")
            .unwrap()
            .into_iter()
            .next()
            .expect("expected one expression");
        let scope = build_scope(&expr);

        let mut schema = MappingSchema::with_dialect(DialectType::BigQuery);
        schema
            .add_table(
                "teams",
                &[("Name".into(), DataType::String { length: None })],
                None,
            )
            .expect("schema setup");

        let mut resolver = Resolver::new(&scope, &schema, true);
        let table = resolver.get_table("Name");
        assert_eq!(table, Some("teams".to_string()));

        let table = resolver.get_table("name");
        assert_eq!(table, Some("teams".to_string()));
    }
}