Skip to main content

polyglot_sql/
lineage.rs

1//! Column Lineage Tracking
2//!
3//! This module provides functionality to track column lineage through SQL queries,
4//! building a graph of how columns flow from source tables to the result set.
5//! Supports UNION/INTERSECT/EXCEPT, CTEs, derived tables, subqueries, and star expansion.
6//!
7
8use crate::dialects::DialectType;
9use crate::error::{ColumnResolutionReason, ColumnResolutionTarget};
10use crate::expressions::{
11    DataType, Expression, Identifier, JoinKind, NamedWindow, Select, Star, With,
12};
13#[cfg(feature = "generate")]
14use crate::generator::Generator;
15use crate::optimizer::annotate_types::annotate_types;
16use crate::optimizer::qualify_schema_aware_expression;
17use crate::schema::{normalize_name, Schema};
18use crate::scope::{
19    build_scope, find_all_in_scope, Scope, ScopeType, SourceInfo as ScopeSourceInfo, SourceKind,
20};
21use crate::{Error, Result};
22use serde::{Deserialize, Serialize};
23use std::collections::{HashMap, HashSet};
24
25/// The ordered output description of a query.
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(rename_all = "camelCase")]
28pub struct QueryOutput {
29    /// Output entries in projection order.
30    pub columns: Vec<OutputColumn>,
31    /// Whether every entry has a stable zero-based ordinal.
32    pub ordinal_complete: bool,
33}
34
35/// One entry in a query's ordered output description.
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37#[serde(tag = "kind", rename_all = "snake_case")]
38pub enum OutputColumn {
39    /// A single output column with a known name.
40    Named {
41        name: String,
42        /// The zero-based output ordinal, when knowable.
43        ordinal: Option<usize>,
44    },
45    /// A single output column whose database-provided name is not reliable.
46    Unnamed {
47        /// The zero-based output ordinal, when knowable.
48        ordinal: Option<usize>,
49    },
50    /// An unresolved wildcard that contributes an unknown number of columns.
51    Wildcard {
52        /// Optional table or source qualifier from `table.*`.
53        qualifier: Option<String>,
54        /// The first possible output ordinal, when no earlier wildcard exists.
55        #[serde(rename = "startOrdinal")]
56        start_ordinal: Option<usize>,
57    },
58}
59
60/// The set operator that owns an immediate lineage branch.
61#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(rename_all = "snake_case")]
63pub enum SetOperator {
64    Union,
65    Intersect,
66    Except,
67}
68
69/// Metadata describing a lineage node's position in an immediate set operation.
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
71#[serde(rename_all = "camelCase")]
72pub struct SetBranch {
73    /// The set operator that owns this branch.
74    pub operator: SetOperator,
75    /// The zero-based branch ordinal within the immediate binary operation.
76    pub ordinal: usize,
77    /// Whether the immediate set operation uses `ALL` semantics.
78    pub all: bool,
79}
80
81/// A node in the column lineage graph
82#[derive(Debug, Clone, Serialize, Deserialize)]
83pub struct LineageNode {
84    /// Name of this lineage step (e.g., "table.column")
85    pub name: String,
86    /// The expression at this node
87    pub expression: Expression,
88    /// The source expression (the full query context)
89    pub source: Expression,
90    /// Downstream nodes that depend on this one
91    pub downstream: Vec<LineageNode>,
92    /// Optional source name (e.g., for derived tables)
93    pub source_name: String,
94    /// Semantic source kind for downstream consumers.
95    pub source_kind: SourceKind,
96    /// User-written source alias when different from canonical source name.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub source_alias: Option<String>,
99    /// Immediate set-operation branch metadata, when this node is a branch root.
100    #[serde(default, skip_serializing_if = "Option::is_none")]
101    pub set_branch: Option<SetBranch>,
102    /// Optional reference node name (e.g., for CTEs)
103    pub reference_node_name: String,
104}
105
106impl LineageNode {
107    /// Create a new lineage node
108    pub fn new(name: impl Into<String>, expression: Expression, source: Expression) -> Self {
109        Self {
110            name: name.into(),
111            expression,
112            source,
113            downstream: Vec::new(),
114            source_name: String::new(),
115            source_kind: SourceKind::Unknown,
116            source_alias: None,
117            set_branch: None,
118            reference_node_name: String::new(),
119        }
120    }
121
122    /// Iterate over all nodes in the lineage graph using DFS
123    pub fn walk(&self) -> LineageWalker<'_> {
124        LineageWalker { stack: vec![self] }
125    }
126
127    /// Get all downstream column names
128    pub fn downstream_names(&self) -> Vec<String> {
129        self.downstream.iter().map(|n| n.name.clone()).collect()
130    }
131}
132
133fn source_kind_for_scope_context(
134    scope: &Scope,
135    source_name: &str,
136    reference_node_name: &str,
137) -> SourceKind {
138    source_kind_for_scope_context_with_type(
139        scope,
140        scope.scope_type,
141        source_name,
142        reference_node_name,
143    )
144}
145
146fn source_kind_for_scope_context_with_type(
147    scope: &Scope,
148    scope_type: ScopeType,
149    source_name: &str,
150    reference_node_name: &str,
151) -> SourceKind {
152    if source_name.is_empty() && reference_node_name.is_empty() {
153        return SourceKind::Root;
154    }
155    if let Some(source_info) = scope.sources.get(source_name) {
156        return source_info.kind;
157    }
158    if scope.cte_sources.contains_key(source_name) {
159        return SourceKind::Cte;
160    }
161    match scope_type {
162        ScopeType::Cte => SourceKind::Cte,
163        ScopeType::DerivedTable => SourceKind::DerivedTable,
164        ScopeType::Udtf => SourceKind::Virtual,
165        _ => SourceKind::Unknown,
166    }
167}
168
169fn apply_scope_context(
170    node: &mut LineageNode,
171    scope: &Scope,
172    source_name: &str,
173    reference_node_name: &str,
174) {
175    node.source_name = source_name.to_string();
176    node.reference_node_name = reference_node_name.to_string();
177    node.source_kind = source_kind_for_scope_context(scope, source_name, reference_node_name);
178}
179
180fn apply_scope_context_with_type(
181    node: &mut LineageNode,
182    scope: &Scope,
183    scope_type: ScopeType,
184    source_name: &str,
185    reference_node_name: &str,
186) {
187    node.source_name = source_name.to_string();
188    node.reference_node_name = reference_node_name.to_string();
189    node.source_kind = source_kind_for_scope_context_with_type(
190        scope,
191        scope_type,
192        source_name,
193        reference_node_name,
194    );
195}
196
197/// Iterator for walking the lineage graph
198pub struct LineageWalker<'a> {
199    stack: Vec<&'a LineageNode>,
200}
201
202impl<'a> Iterator for LineageWalker<'a> {
203    type Item = &'a LineageNode;
204
205    fn next(&mut self) -> Option<Self::Item> {
206        if let Some(node) = self.stack.pop() {
207            // Add children in reverse order so they're visited in order
208            for child in node.downstream.iter().rev() {
209                self.stack.push(child);
210            }
211            Some(node)
212        } else {
213            None
214        }
215    }
216}
217
218// ---------------------------------------------------------------------------
219// ColumnRef: name or positional index for column lookup
220// ---------------------------------------------------------------------------
221
222/// Column reference for lineage tracing — by name or positional index.
223enum ColumnRef<'a> {
224    Name(&'a str),
225    Index(usize),
226}
227
228// ---------------------------------------------------------------------------
229// Public API
230// ---------------------------------------------------------------------------
231
232/// Build the lineage graph for a column in a SQL query
233///
234/// # Arguments
235/// * `column` - The column name to trace lineage for
236/// * `sql` - The SQL expression (SELECT, UNION, etc.)
237/// * `dialect` - Optional dialect for parsing
238/// * `trim_selects` - If true, trim the source SELECT to only include the target column
239///
240/// # Returns
241/// The root lineage node for the specified column
242///
243/// # Example
244/// ```ignore
245/// use polyglot_sql::lineage::lineage;
246/// use polyglot_sql::parse_one;
247/// use polyglot_sql::DialectType;
248///
249/// let sql = "SELECT a, b + 1 AS c FROM t";
250/// let expr = parse_one(sql, DialectType::Generic).unwrap();
251/// let node = lineage("c", &expr, None, false).unwrap();
252/// ```
253pub fn lineage(
254    column: &str,
255    sql: &Expression,
256    dialect: Option<DialectType>,
257    trim_selects: bool,
258) -> Result<LineageNode> {
259    let prepared = prepare_lineage_expression(sql, None, dialect, false)?;
260    lineage_from_column_ref(ColumnRef::Name(column), &prepared, dialect, trim_selects)
261}
262
263/// Build the lineage graph for the column at a zero-based output ordinal.
264pub fn lineage_at(
265    ordinal: usize,
266    sql: &Expression,
267    dialect: Option<DialectType>,
268    trim_selects: bool,
269) -> Result<LineageNode> {
270    let prepared = prepare_lineage_expression(sql, None, dialect, false)?;
271    lineage_from_column_ref(ColumnRef::Index(ordinal), &prepared, dialect, trim_selects)
272}
273
274/// Build the lineage graph for a column in a SQL query using optional schema metadata.
275///
276/// When `schema` is provided, the query is first qualified with
277/// `optimizer::qualify_columns`, allowing more accurate lineage for unqualified or
278/// ambiguous column references.
279///
280/// # Arguments
281/// * `column` - The column name to trace lineage for
282/// * `sql` - The SQL expression (SELECT, UNION, etc.)
283/// * `schema` - Optional schema used for qualification
284/// * `dialect` - Optional dialect for qualification and lineage handling
285/// * `trim_selects` - If true, trim the source SELECT to only include the target column
286///
287/// # Returns
288/// The root lineage node for the specified column
289pub fn lineage_with_schema(
290    column: &str,
291    sql: &Expression,
292    schema: Option<&dyn Schema>,
293    dialect: Option<DialectType>,
294    trim_selects: bool,
295) -> Result<LineageNode> {
296    let prepared = prepare_lineage_expression(sql, schema, dialect, true)?;
297    lineage_from_column_ref(ColumnRef::Name(column), &prepared, dialect, trim_selects)
298}
299
300/// Build schema-aware lineage for the column at a zero-based output ordinal.
301pub fn lineage_at_with_schema(
302    ordinal: usize,
303    sql: &Expression,
304    schema: Option<&dyn Schema>,
305    dialect: Option<DialectType>,
306    trim_selects: bool,
307) -> Result<LineageNode> {
308    let prepared = prepare_lineage_expression(sql, schema, dialect, true)?;
309    lineage_from_column_ref(ColumnRef::Index(ordinal), &prepared, dialect, trim_selects)
310}
311
312/// Return the ordered output description of a query.
313pub fn output_columns(sql: &Expression, dialect: Option<DialectType>) -> Result<QueryOutput> {
314    let prepared = prepare_lineage_expression(sql, None, dialect, false)?;
315    query_output_from_expression(&prepared, dialect)
316}
317
318/// Return the ordered output description of a query after schema-aware expansion.
319pub fn output_columns_with_schema(
320    sql: &Expression,
321    schema: Option<&dyn Schema>,
322    dialect: Option<DialectType>,
323) -> Result<QueryOutput> {
324    let prepared = prepare_lineage_expression(sql, schema, dialect, true)?;
325    query_output_from_expression(&prepared, dialect)
326}
327
328fn prepare_lineage_expression(
329    sql: &Expression,
330    schema: Option<&dyn Schema>,
331    dialect: Option<DialectType>,
332    schema_aware: bool,
333) -> Result<Expression> {
334    let normalized = lineage_normalized_expression(sql);
335    let mut prepared = if schema_aware {
336        if let Some(schema) = schema {
337            qualify_schema_aware_expression(normalized.clone(), schema, dialect).map_err(
338                |error| {
339                    Error::internal(format!("Lineage qualification failed with schema: {error}"))
340                },
341            )?
342        } else {
343            normalized
344        }
345    } else {
346        normalized
347    };
348
349    if schema_aware {
350        annotate_types(&mut prepared, schema, dialect);
351        expand_cte_stars(&mut prepared, schema);
352    } else if has_lineage_with_clause(&prepared) {
353        expand_cte_stars(&mut prepared, None);
354    }
355
356    Ok(prepared)
357}
358
359fn lineage_from_column_ref(
360    column: ColumnRef<'_>,
361    sql: &Expression,
362    dialect: Option<DialectType>,
363    trim_selects: bool,
364) -> Result<LineageNode> {
365    let scope = build_scope(sql);
366    to_node(column, scope, dialect, "", "", "", trim_selects)
367}
368
369fn lineage_normalized_expression(sql: &Expression) -> Expression {
370    match sql {
371        Expression::Prepare(prepare) => lineage_normalized_expression(&prepare.statement),
372        Expression::CreateTable(create) => create
373            .as_select
374            .as_ref()
375            .map(|query| attach_with_to_query(query.clone(), create.with_cte.clone()))
376            .unwrap_or_else(|| sql.clone()),
377        Expression::CreateView(create) => lineage_normalized_expression(&create.query),
378        Expression::Upsert(upsert) => lineage_normalized_expression(&upsert.source),
379        Expression::Insert(insert) => insert
380            .query
381            .as_ref()
382            .map(|query| attach_with_to_query(query.clone(), insert.with.clone()))
383            .unwrap_or_else(|| sql.clone()),
384        _ => sql.clone(),
385    }
386}
387
388fn attach_with_to_query(
389    mut query: Expression,
390    with: Option<crate::expressions::With>,
391) -> Expression {
392    if let Some(with) = with {
393        attach_with_to_query_mut(&mut query, with);
394    }
395    query
396}
397
398fn attach_with_to_query_mut(query: &mut Expression, with: crate::expressions::With) {
399    match query {
400        Expression::Select(select) => {
401            if select.with.is_none() {
402                select.with = Some(with);
403            }
404        }
405        Expression::Union(union) => {
406            if union.with.is_none() {
407                union.with = Some(with);
408            }
409        }
410        Expression::Intersect(intersect) => {
411            if intersect.with.is_none() {
412                intersect.with = Some(with);
413            }
414        }
415        Expression::Except(except) => {
416            if except.with.is_none() {
417                except.with = Some(with);
418            }
419        }
420        Expression::Paren(paren) => attach_with_to_query_mut(&mut paren.this, with),
421        _ => {}
422    }
423}
424
425fn has_lineage_with_clause(expr: &Expression) -> bool {
426    match expr {
427        Expression::Select(select) => select.with.is_some(),
428        Expression::Union(union) => {
429            union.with.is_some()
430                || has_lineage_with_clause(&union.left)
431                || has_lineage_with_clause(&union.right)
432        }
433        Expression::Intersect(intersect) => {
434            intersect.with.is_some()
435                || has_lineage_with_clause(&intersect.left)
436                || has_lineage_with_clause(&intersect.right)
437        }
438        Expression::Except(except) => {
439            except.with.is_some()
440                || has_lineage_with_clause(&except.left)
441                || has_lineage_with_clause(&except.right)
442        }
443        Expression::Paren(paren) => has_lineage_with_clause(&paren.this),
444        _ => false,
445    }
446}
447
448// ---------------------------------------------------------------------------
449// CTE star expansion
450// ---------------------------------------------------------------------------
451
452/// Normalize an identifier for CTE name matching.
453///
454/// Follows SQL semantics: unquoted identifiers are case-insensitive (lowercased),
455/// quoted identifiers preserve their original case. This matches sqlglot's
456/// `normalize_identifiers` behavior.
457fn normalize_cte_name(ident: &Identifier) -> String {
458    if ident.quoted {
459        ident.name.clone()
460    } else {
461        ident.name.to_lowercase()
462    }
463}
464
465/// Expand SELECT * in CTEs by walking CTE definitions in order and propagating
466/// resolved column lists. This also handles chained CTEs in schema-free lineage
467/// and analysis paths, where full column qualification is not performed.
468///
469/// When `schema` is provided, stars from external tables (not CTEs) are also resolved
470/// by looking up column names in the schema. This enables correct expansion of patterns
471/// like `WITH cte AS (SELECT * FROM external_table) SELECT * FROM cte`.
472///
473/// CTE name matching follows SQL identifier semantics: unquoted names are compared
474/// case-insensitively (lowercased), while quoted names preserve their original case.
475/// This matches sqlglot's `normalize_identifiers` behavior.
476pub fn expand_cte_stars(expr: &mut Expression, schema: Option<&dyn Schema>) {
477    if let Expression::Prepare(prepare) = expr {
478        expand_cte_stars(&mut prepare.statement, schema);
479        return;
480    }
481
482    let resolved_cte_columns = {
483        let with = match query_with_mut(expr) {
484            Some(with) => with,
485            None => return,
486        };
487        let is_recursive_with = with.recursive;
488        let mut resolved_cte_columns: HashMap<String, Vec<String>> = HashMap::new();
489
490        for cte in &mut with.ctes {
491            let cte_name = normalize_cte_name(&cte.alias);
492            let explicit_columns = (!cte.columns.is_empty())
493                .then(|| cte.columns.iter().map(|c| c.name.clone()).collect());
494
495            // Skip recursive CTE bodies — resolving their self-references safely is
496            // more complex than ordered, non-recursive CTE propagation. Inspect every
497            // set-operation arm because the recursive reference normally appears in
498            // the right branch after a non-recursive base case.
499            if is_recursive_with && query_references_source(&cte.this, &cte_name) {
500                if let Some(columns) = explicit_columns {
501                    resolved_cte_columns.insert(cte_name, columns);
502                }
503                continue;
504            }
505
506            // Rewrite every SELECT arm, but derive the CTE's implicit output names
507            // from the leftmost arm only. Explicit CTE column aliases override those
508            // implicit names without preventing safe body expansion.
509            let implicit_columns =
510                rewrite_stars_in_query(&mut cte.this, &resolved_cte_columns, schema);
511            if let Some(columns) = explicit_columns.or(implicit_columns) {
512                resolved_cte_columns.insert(cte_name, columns);
513            }
514        }
515
516        resolved_cte_columns
517    };
518
519    // Also expand stars in every arm of the outer query. WITH can be attached
520    // directly to a root set operation, so limiting this to Expression::Select
521    // would skip the entire query.
522    rewrite_stars_in_query(expr, &resolved_cte_columns, schema);
523}
524
525/// Get the WITH clause attached to a query root, drilling through parentheses.
526fn query_with_mut(expr: &mut Expression) -> Option<&mut With> {
527    let mut current = expr;
528    loop {
529        match current {
530            Expression::Select(select) => return select.with.as_mut(),
531            Expression::Union(union) => return union.with.as_mut(),
532            Expression::Intersect(intersect) => return intersect.with.as_mut(),
533            Expression::Except(except) => return except.with.as_mut(),
534            Expression::Paren(p) => current = &mut p.this,
535            Expression::Subquery(subquery) => current = &mut subquery.this,
536            _ => return None,
537        }
538    }
539}
540
541/// Whether any SELECT arm in a query directly references `source_name`.
542///
543/// This is used to identify recursive CTEs whose self-reference commonly lives
544/// in a non-leftmost set-operation branch.
545fn query_references_source(expr: &Expression, source_name: &str) -> bool {
546    let mut stack = vec![expr];
547
548    while let Some(current) = stack.pop() {
549        match current {
550            Expression::Select(select) => {
551                if get_select_sources(select)
552                    .iter()
553                    .any(|source| source.normalized == source_name)
554                {
555                    return true;
556                }
557            }
558            Expression::Union(union) => {
559                stack.push(&union.right);
560                stack.push(&union.left);
561            }
562            Expression::Intersect(intersect) => {
563                stack.push(&intersect.right);
564                stack.push(&intersect.left);
565            }
566            Expression::Except(except) => {
567                stack.push(&except.right);
568                stack.push(&except.left);
569            }
570            Expression::Paren(paren) => stack.push(&paren.this),
571            Expression::Subquery(subquery) => stack.push(&subquery.this),
572            _ => {}
573        }
574    }
575
576    false
577}
578
579/// Rewrite stars in every SELECT arm of a query.
580///
581/// The traversal visits left branches first, so the first returned column list
582/// remains the set operation's output column list while all later arms are still
583/// rewritten independently. An explicit stack avoids adding recursion pressure
584/// for deeply nested set-operation chains.
585fn rewrite_stars_in_query(
586    expr: &mut Expression,
587    resolved_ctes: &HashMap<String, Vec<String>>,
588    schema: Option<&dyn Schema>,
589) -> Option<Vec<String>> {
590    let mut leftmost_columns = None;
591    let mut stack = vec![expr];
592
593    while let Some(current) = stack.pop() {
594        match current {
595            Expression::Select(select) => {
596                let columns = rewrite_stars_in_select(select, resolved_ctes, schema);
597                if leftmost_columns.is_none() {
598                    leftmost_columns = Some(columns);
599                }
600            }
601            Expression::Union(union) => {
602                stack.push(&mut union.right);
603                stack.push(&mut union.left);
604            }
605            Expression::Intersect(intersect) => {
606                stack.push(&mut intersect.right);
607                stack.push(&mut intersect.left);
608            }
609            Expression::Except(except) => {
610                stack.push(&mut except.right);
611                stack.push(&mut except.left);
612            }
613            Expression::Paren(paren) => stack.push(&mut paren.this),
614            Expression::Subquery(subquery) => stack.push(&mut subquery.this),
615            _ => {}
616        }
617    }
618
619    leftmost_columns
620}
621
622/// Rewrite star expressions in a SELECT using resolved CTE column lists.
623/// Falls back to `schema` for external table column lookup.
624/// Returns the list of output column names after expansion.
625fn rewrite_stars_in_select(
626    select: &mut Select,
627    resolved_ctes: &HashMap<String, Vec<String>>,
628    schema: Option<&dyn Schema>,
629) -> Vec<String> {
630    // The AST represents star expressions in two forms depending on syntax:
631    //   - `SELECT *`      → Expression::Star (unqualified star)
632    //   - `SELECT table.*` → Expression::Column { name: "*", table: Some(...) } (qualified star)
633    // Both must be checked to handle all star patterns.
634    let has_star = select
635        .expressions
636        .iter()
637        .any(|e| matches!(e, Expression::Star(_)));
638    let has_qualified_star = select
639        .expressions
640        .iter()
641        .any(|e| matches!(e, Expression::Column(c) if c.name.name == "*"));
642
643    if !has_star && !has_qualified_star {
644        // No stars — just extract column names without rewriting
645        return select
646            .expressions
647            .iter()
648            .filter_map(get_expression_output_name)
649            .collect();
650    }
651
652    let sources = get_select_sources(select);
653    let mut new_expressions = Vec::new();
654    let mut result_columns = Vec::new();
655
656    for expr in &select.expressions {
657        match expr {
658            Expression::Star(star) => {
659                let qual = star.table.as_ref();
660                let expanded = (!star.has_except_pattern())
661                    .then(|| expand_star_from_sources(qual, &sources, resolved_ctes, schema))
662                    .flatten();
663                if let Some(expanded) = expanded {
664                    for (src_alias, col_name) in &expanded {
665                        let table_id = Identifier::new(src_alias);
666                        let column = make_column_expr(col_name, Some(&table_id));
667                        if let Some(expression) =
668                            star.expand_column(col_name, column, schema.and_then(|s| s.dialect()))
669                        {
670                            if let Some(name) = get_expression_output_name(&expression) {
671                                result_columns.push(name);
672                            }
673                            new_expressions.push(expression);
674                        }
675                    }
676                } else {
677                    new_expressions.push(expr.clone());
678                    result_columns.push("*".to_string());
679                }
680            }
681            Expression::Column(c) if c.name.name == "*" => {
682                let qual = c.table.as_ref();
683                if let Some(expanded) =
684                    expand_star_from_sources(qual, &sources, resolved_ctes, schema)
685                {
686                    for (_src_alias, col_name) in &expanded {
687                        // Keep the original table qualifier for qualified stars (table.*)
688                        new_expressions.push(make_column_expr(col_name, c.table.as_ref()));
689                        result_columns.push(col_name.clone());
690                    }
691                } else {
692                    new_expressions.push(expr.clone());
693                    result_columns.push("*".to_string());
694                }
695            }
696            _ => {
697                new_expressions.push(expr.clone());
698                if let Some(name) = get_expression_output_name(expr) {
699                    result_columns.push(name);
700                }
701            }
702        }
703    }
704
705    select.expressions = new_expressions;
706    result_columns
707}
708
709/// Try to expand a star expression by looking up source columns from resolved CTEs,
710/// falling back to the schema for external tables.
711/// Returns (source_alias, column_name) pairs so the caller can set table qualifiers.
712/// `qualifier`: Optional table qualifier (for `table.*`). If None, expand all sources.
713fn expand_star_from_sources(
714    qualifier: Option<&Identifier>,
715    sources: &[SourceInfo],
716    resolved_ctes: &HashMap<String, Vec<String>>,
717    schema: Option<&dyn Schema>,
718) -> Option<Vec<(String, String)>> {
719    let mut expanded = Vec::new();
720
721    if let Some(qual) = qualifier {
722        // Qualified star: table.*
723        let qual_normalized = normalize_cte_name(qual);
724        for src in sources {
725            if src.normalized == qual_normalized || src.alias.to_lowercase() == qual_normalized {
726                if src.transforms_columns {
727                    return None;
728                }
729                // Try CTE first
730                if let Some(cols) = resolved_ctes.get(&src.normalized) {
731                    expanded.extend(cols.iter().map(|c| (src.alias.clone(), c.clone())));
732                    return Some(expanded);
733                }
734                // Fall back to schema
735                if let Some(cols) = lookup_schema_columns(schema, &src.fq_name) {
736                    expanded.extend(cols.into_iter().map(|c| (src.alias.clone(), c)));
737                    return Some(expanded);
738                }
739            }
740        }
741        None
742    } else {
743        // Unqualified star: expand all sources.
744        // Intentionally conservative: if any source can't be resolved, the entire
745        // expansion is aborted. Partial expansion would produce an incomplete column
746        // list, causing downstream lineage resolution to silently omit columns.
747        // This matches sqlglot's behavior (raises SqlglotError when schema is missing).
748        let mut any_expanded = false;
749        for src in sources {
750            if src.transforms_columns {
751                return None;
752            }
753            if let Some(cols) = resolved_ctes.get(&src.normalized) {
754                expanded.extend(cols.iter().map(|c| (src.alias.clone(), c.clone())));
755                any_expanded = true;
756            } else if let Some(cols) = lookup_schema_columns(schema, &src.fq_name) {
757                expanded.extend(cols.into_iter().map(|c| (src.alias.clone(), c)));
758                any_expanded = true;
759            } else {
760                return None;
761            }
762        }
763        if any_expanded {
764            Some(expanded)
765        } else {
766            None
767        }
768    }
769}
770
771/// Look up column names for a table from the schema.
772fn lookup_schema_columns(schema: Option<&dyn Schema>, fq_name: &str) -> Option<Vec<String>> {
773    let schema = schema?;
774    if fq_name.is_empty() {
775        return None;
776    }
777    schema
778        .column_names(fq_name)
779        .ok()
780        .filter(|cols| !cols.is_empty() && !cols.contains(&"*".to_string()))
781}
782
783/// Create a Column expression with the given name and optional table qualifier.
784fn make_column_expr(name: &str, table: Option<&Identifier>) -> Expression {
785    Expression::Column(Box::new(crate::expressions::Column {
786        name: Identifier::new(name),
787        table: table.cloned(),
788        join_mark: false,
789        trailing_comments: Vec::new(),
790        span: None,
791        inferred_type: None,
792    }))
793}
794
795/// Extract the output name of a SELECT expression.
796fn get_expression_output_name(expr: &Expression) -> Option<String> {
797    match expr {
798        Expression::Alias(a) => Some(a.alias.name.clone()),
799        Expression::Column(c) => Some(c.name.name.clone()),
800        Expression::Identifier(id) => Some(id.name.clone()),
801        Expression::Star(_) => Some("*".to_string()),
802        _ => None,
803    }
804}
805
806/// Source info extracted from a SELECT's FROM/JOIN clauses in a single pass.
807struct SourceInfo {
808    alias: String,
809    /// Whether this source was introduced through a quoted identifier.
810    ///
811    /// The schema-less star passthrough heuristic must stay conservative for
812    /// quoted sources because unresolved quoted table names can be distinct
813    /// from similarly named CTEs that older scope paths still compare
814    /// case-insensitively.
815    quoted: bool,
816    /// Normalized name for CTE lookup: unquoted → lowercased, quoted → as-is.
817    normalized: String,
818    /// Fully-qualified table name for schema lookup (e.g., "db.schema.table").
819    fq_name: String,
820    /// PIVOT/UNPIVOT outputs cannot be expanded using the input relation's columns.
821    transforms_columns: bool,
822}
823
824/// Extract source info (alias, normalized CTE name, fully-qualified name) from a
825/// SELECT's FROM and JOIN clauses in a single pass.
826fn get_select_sources(select: &Select) -> Vec<SourceInfo> {
827    let mut sources = Vec::new();
828
829    fn extract_source(expr: &Expression) -> Option<SourceInfo> {
830        fn virtual_source_info(alias: &Identifier) -> SourceInfo {
831            SourceInfo {
832                alias: alias.name.clone(),
833                quoted: alias.quoted,
834                normalized: normalize_cte_name(alias),
835                fq_name: alias.name.clone(),
836                transforms_columns: false,
837            }
838        }
839
840        fn named_virtual_source_info(alias: &str) -> SourceInfo {
841            SourceInfo {
842                alias: alias.to_string(),
843                quoted: false,
844                normalized: alias.to_lowercase(),
845                fq_name: alias.to_string(),
846                transforms_columns: false,
847            }
848        }
849
850        match expr {
851            Expression::Table(t) => {
852                let normalized = normalize_cte_name(&t.name);
853                let alias = t
854                    .alias
855                    .as_ref()
856                    .map(|a| a.name.clone())
857                    .unwrap_or_else(|| t.name.name.clone());
858                let mut parts = Vec::new();
859                if let Some(catalog) = &t.catalog {
860                    parts.push(catalog.name.clone());
861                }
862                if let Some(schema) = &t.schema {
863                    parts.push(schema.name.clone());
864                }
865                parts.push(t.name.name.clone());
866                let fq_name = parts.join(".");
867                Some(SourceInfo {
868                    alias,
869                    quoted: t.name.quoted,
870                    normalized,
871                    fq_name,
872                    transforms_columns: false,
873                })
874            }
875            Expression::Subquery(s) => {
876                let alias_identifier = s.alias.as_ref()?;
877                let alias = alias_identifier.name.clone();
878                let normalized = alias.to_lowercase();
879                let fq_name = alias.clone();
880                Some(SourceInfo {
881                    alias,
882                    quoted: alias_identifier.quoted,
883                    normalized,
884                    fq_name,
885                    transforms_columns: false,
886                })
887            }
888            Expression::Unnest(u) => u.alias.as_ref().map(virtual_source_info),
889            Expression::Alias(a) if matches!(&a.this, Expression::Unnest(_)) => {
890                Some(virtual_source_info(&a.alias))
891            }
892            Expression::Alias(a) if is_query_like_relation(&a.this) => {
893                Some(virtual_source_info(&a.alias))
894            }
895            Expression::Lateral(lateral) => lateral.alias.as_deref().map(named_virtual_source_info),
896            Expression::LateralView(lateral_view) => lateral_view
897                .table_alias
898                .as_ref()
899                .or_else(|| lateral_view.column_aliases.first())
900                .map(virtual_source_info),
901            Expression::Pivot(pivot) => {
902                let alias = pivot_lineage_source_name(
903                    &pivot.this,
904                    pivot.alias.as_ref().map(|alias| alias.name.as_str()),
905                );
906                Some(SourceInfo {
907                    alias: alias.clone(),
908                    quoted: false,
909                    normalized: alias.to_lowercase(),
910                    fq_name: alias,
911                    transforms_columns: true,
912                })
913            }
914            Expression::Unpivot(unpivot) => {
915                let alias = pivot_lineage_source_name(
916                    &unpivot.this,
917                    unpivot.alias.as_ref().map(|alias| alias.name.as_str()),
918                );
919                Some(SourceInfo {
920                    alias: alias.clone(),
921                    quoted: false,
922                    normalized: alias.to_lowercase(),
923                    fq_name: alias,
924                    transforms_columns: true,
925                })
926            }
927            Expression::Paren(p) => extract_source(&p.this),
928            _ => None,
929        }
930    }
931
932    if let Some(from) = &select.from {
933        for expr in &from.expressions {
934            if let Some(info) = extract_source(expr) {
935                sources.push(info);
936            }
937        }
938    }
939    for join in &select.joins {
940        if is_semi_or_anti_join_kind(join.kind) {
941            continue;
942        }
943        if let Some(info) = extract_source(&join.this) {
944            sources.push(info);
945        }
946    }
947    for lateral_view in &select.lateral_views {
948        if let Some(info) = extract_source(&Expression::LateralView(Box::new(lateral_view.clone())))
949        {
950            sources.push(info);
951        }
952    }
953    sources
954}
955
956fn pivot_lineage_source_name(source: &Expression, explicit_alias: Option<&str>) -> String {
957    if let Some(alias) = explicit_alias {
958        return alias.to_string();
959    }
960
961    match source {
962        Expression::Table(table) => table
963            .alias
964            .as_ref()
965            .map(|alias| alias.name.clone())
966            .unwrap_or_else(|| table.name.name.clone()),
967        Expression::Subquery(subquery) => subquery
968            .alias
969            .as_ref()
970            .map(|alias| alias.name.clone())
971            .unwrap_or_else(|| "_0".to_string()),
972        Expression::Paren(paren) => pivot_lineage_source_name(&paren.this, explicit_alias),
973        _ => "_0".to_string(),
974    }
975}
976
977/// Get all source tables from a lineage graph
978pub fn get_source_tables(node: &LineageNode) -> HashSet<String> {
979    let mut tables = HashSet::new();
980    collect_source_tables(node, &mut tables);
981    tables
982}
983
984/// Recursively collect source table names from lineage graph
985pub fn collect_source_tables(node: &LineageNode, tables: &mut HashSet<String>) {
986    if let Expression::Table(table) = &node.source {
987        tables.insert(table.name.name.clone());
988    }
989    for child in &node.downstream {
990        collect_source_tables(child, tables);
991    }
992}
993
994// ---------------------------------------------------------------------------
995// Core recursive lineage builder
996// ---------------------------------------------------------------------------
997
998/// Maximum recursion depth for lineage tracing to prevent stack overflow
999/// on circular or deeply nested CTE chains.
1000const MAX_LINEAGE_DEPTH: usize = 64;
1001
1002#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1003struct ScopeId(usize);
1004
1005struct IndexedScope {
1006    scope: Scope,
1007    subquery_scopes: Vec<ScopeId>,
1008    derived_table_scopes: Vec<ScopeId>,
1009    cte_scopes: Vec<ScopeId>,
1010    union_scopes: Vec<ScopeId>,
1011}
1012
1013struct LineageScopeContext {
1014    scopes: Vec<IndexedScope>,
1015    /// Usage analysis must not claim undeclared qualifiers as physical tables.
1016    conservative: bool,
1017}
1018
1019/// Reuse the lineage resolver for analysis of non-output column occurrences.
1020/// The caller selects the lexical owner and checks ambiguous/open sources first.
1021/// This keeps CTE, derived-table, virtual-source and set-operation tracing shared
1022/// with projection lineage, without manufacturing SELECT projections.
1023pub(crate) struct ScopedLineage {
1024    context: LineageScopeContext,
1025    root: ScopeId,
1026    ctes: Vec<ScopeId>,
1027    dialect: Option<DialectType>,
1028}
1029
1030impl ScopedLineage {
1031    /// Prepare and scope `sql` once, for the lineage of many output columns.
1032    /// Each column gets the result of `lineage_with_schema` (with a schema)
1033    /// or `lineage` (without one).
1034    pub(crate) fn for_query(
1035        sql: &Expression,
1036        schema: Option<&dyn Schema>,
1037        dialect: Option<DialectType>,
1038    ) -> Result<Self> {
1039        let prepared = prepare_lineage_expression(sql, schema, dialect, schema.is_some())?;
1040        let (context, root) = LineageScopeContext::from_scope(build_scope(&prepared));
1041        Ok(Self {
1042            context,
1043            root,
1044            ctes: Vec::new(),
1045            dialect,
1046        })
1047    }
1048
1049    pub(crate) fn named(&self, column: &str) -> Result<LineageNode> {
1050        to_node_inner(
1051            ColumnRef::Name(column),
1052            &self.context,
1053            self.root,
1054            self.dialect,
1055            "",
1056            "",
1057            "",
1058            false,
1059            &self.ctes,
1060            0,
1061        )
1062    }
1063}
1064
1065#[cfg(feature = "generate")]
1066impl ScopedLineage {
1067    pub(crate) fn new(scope: Scope, inherited_ctes: &[Scope], dialect: DialectType) -> Self {
1068        let visible_ctes = scope.cte_sources.clone();
1069        let (mut context, root) = LineageScopeContext::from_scope(scope);
1070        context.conservative = true;
1071        let mut ctes = context.indexed(root).cte_scopes.clone();
1072        for cte_scope in inherited_ctes {
1073            if let Expression::Cte(cte) = &cte_scope.expression {
1074                if visible_ctes
1075                    .get(&cte.alias.name)
1076                    .is_some_and(|source| source.expression == cte_scope.expression)
1077                    && !ctes
1078                        .iter()
1079                        .any(|id| context.scope(*id).expression == cte_scope.expression)
1080                {
1081                    ctes.push(context.insert_scope(cte_scope.clone()));
1082                }
1083            }
1084        }
1085        Self {
1086            context,
1087            root,
1088            ctes,
1089            dialect: Some(dialect),
1090        }
1091    }
1092
1093    pub(crate) fn column(&self, source: &str, column: &str) -> LineageNode {
1094        let mut node = LineageNode::new(
1095            column,
1096            Expression::qualified_column(source, column),
1097            Expression::Null(crate::expressions::Null),
1098        );
1099        resolve_qualified_column(
1100            &mut node,
1101            &self.context,
1102            self.root,
1103            self.dialect,
1104            source,
1105            column,
1106            column,
1107            false,
1108            &self.ctes,
1109            0,
1110        );
1111        node
1112    }
1113
1114    pub(crate) fn output(&self, ordinal: usize) -> Result<LineageNode> {
1115        to_node_inner(
1116            ColumnRef::Index(ordinal),
1117            &self.context,
1118            self.root,
1119            self.dialect,
1120            "",
1121            "",
1122            "",
1123            false,
1124            &self.ctes,
1125            0,
1126        )
1127    }
1128
1129    pub(crate) fn output_names(&self) -> Vec<String> {
1130        crate::ast_transforms::get_output_column_names_for_dialect(
1131            effective_scope_expression(&self.context.scope(self.root).expression),
1132            self.dialect,
1133        )
1134    }
1135
1136    pub(crate) fn subquery_output(&self, index: usize) -> Result<LineageNode> {
1137        let scope_id = *self
1138            .context
1139            .indexed(self.root)
1140            .subquery_scopes
1141            .get(index)
1142            .ok_or_else(|| Error::internal("missing scalar subquery scope"))?;
1143        to_node_inner(
1144            ColumnRef::Index(0),
1145            &self.context,
1146            scope_id,
1147            self.dialect,
1148            "",
1149            "",
1150            "",
1151            false,
1152            &self.ctes,
1153            0,
1154        )
1155    }
1156}
1157
1158impl LineageScopeContext {
1159    fn from_scope(scope: Scope) -> (Self, ScopeId) {
1160        let mut context = Self {
1161            scopes: Vec::new(),
1162            conservative: false,
1163        };
1164        let root = context.insert_scope(scope);
1165        (context, root)
1166    }
1167
1168    fn insert_scope(&mut self, mut scope: Scope) -> ScopeId {
1169        let subquery_scopes = std::mem::take(&mut scope.subquery_scopes)
1170            .into_iter()
1171            .map(|child| self.insert_scope(child))
1172            .collect();
1173        let derived_table_scopes = std::mem::take(&mut scope.derived_table_scopes)
1174            .into_iter()
1175            .map(|child| self.insert_scope(child))
1176            .collect();
1177        let cte_scopes = std::mem::take(&mut scope.cte_scopes)
1178            .into_iter()
1179            .map(|child| self.insert_scope(child))
1180            .collect();
1181        let union_scopes = std::mem::take(&mut scope.union_scopes)
1182            .into_iter()
1183            .map(|child| self.insert_scope(child))
1184            .collect();
1185
1186        let id = ScopeId(self.scopes.len());
1187        self.scopes.push(IndexedScope {
1188            scope,
1189            subquery_scopes,
1190            derived_table_scopes,
1191            cte_scopes,
1192            union_scopes,
1193        });
1194        id
1195    }
1196
1197    fn indexed(&self, id: ScopeId) -> &IndexedScope {
1198        &self.scopes[id.0]
1199    }
1200
1201    fn scope(&self, id: ScopeId) -> &Scope {
1202        &self.indexed(id).scope
1203    }
1204}
1205
1206/// Recursively build a lineage node for a column in a scope.
1207fn to_node(
1208    column: ColumnRef<'_>,
1209    scope: Scope,
1210    dialect: Option<DialectType>,
1211    scope_name: &str,
1212    source_name: &str,
1213    reference_node_name: &str,
1214    trim_selects: bool,
1215) -> Result<LineageNode> {
1216    let (context, scope_id) = LineageScopeContext::from_scope(scope);
1217    to_node_inner(
1218        column,
1219        &context,
1220        scope_id,
1221        dialect,
1222        scope_name,
1223        source_name,
1224        reference_node_name,
1225        trim_selects,
1226        &[],
1227        0,
1228    )
1229}
1230
1231fn to_node_inner(
1232    column: ColumnRef<'_>,
1233    context: &LineageScopeContext,
1234    scope_id: ScopeId,
1235    dialect: Option<DialectType>,
1236    scope_name: &str,
1237    source_name: &str,
1238    reference_node_name: &str,
1239    trim_selects: bool,
1240    ancestor_cte_scopes: &[ScopeId],
1241    depth: usize,
1242) -> Result<LineageNode> {
1243    if depth > MAX_LINEAGE_DEPTH {
1244        return Err(Error::internal(format!(
1245            "lineage recursion depth exceeded (>{MAX_LINEAGE_DEPTH}) — possible circular CTE reference for scope '{scope_name}'"
1246        )));
1247    }
1248    let scope = context.scope(scope_id);
1249    let scope_expr = &scope.expression;
1250
1251    // Build combined CTE scopes: current scope's cte_scopes + ancestors
1252    let mut all_cte_scopes = context.indexed(scope_id).cte_scopes.clone();
1253    all_cte_scopes.extend_from_slice(ancestor_cte_scopes);
1254    let descendant_cte_scopes = descendant_cte_scope_ids(&all_cte_scopes, scope_id);
1255
1256    // 0. Unwrap CTE scope — CTE scope expressions are Expression::Cte(...)
1257    //    but we need the inner query (SELECT/UNION) for column lookup.
1258    let effective_expr = effective_scope_expression(scope_expr);
1259    // A CTE's explicit output names replace its projection names by ordinal.
1260    // Resolve that boundary before unwrapping, including unnamed expressions.
1261    let alias_ordinal = match (scope_expr, &column) {
1262        (Expression::Cte(cte), ColumnRef::Name(name)) if !cte.columns.is_empty() => {
1263            cte.columns.iter().position(|alias| {
1264                normalize_column_name(&alias.name, dialect) == normalize_column_name(name, dialect)
1265            })
1266        }
1267        _ => None,
1268    };
1269    let lookup_column = alias_ordinal.map(ColumnRef::Index);
1270    let lookup_column = lookup_column.as_ref().unwrap_or(&column);
1271
1272    // 1. Set operations (UNION / INTERSECT / EXCEPT)
1273    if matches!(
1274        effective_expr,
1275        Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_)
1276    ) {
1277        let mut node = handle_set_operation(
1278            lookup_column,
1279            context,
1280            scope_id,
1281            effective_expr,
1282            matches!(scope_expr, Expression::Cte(_)).then_some(ScopeType::Root),
1283            dialect,
1284            scope_name,
1285            source_name,
1286            reference_node_name,
1287            trim_selects,
1288            &descendant_cte_scopes,
1289            depth,
1290        )?;
1291        if alias_ordinal.is_some() {
1292            if let ColumnRef::Name(name) = &column {
1293                node.name = (*name).to_string();
1294            }
1295        }
1296        return Ok(node);
1297    }
1298
1299    // 2. Find the select expression for this column
1300    let select_expr = find_select_expr(effective_expr, lookup_column, dialect)?;
1301    let column_name = resolve_column_name(&column, &select_expr);
1302
1303    // 3. Trim source if requested
1304    // Internal fact consumers need physical leaf sources, not a copy of the
1305    // entire SELECT for every output. Public lineage retains its source ASTs.
1306    let node_source = if context.conservative {
1307        Expression::Null(crate::expressions::Null)
1308    } else if trim_selects {
1309        trim_source(effective_expr, &select_expr)
1310    } else {
1311        effective_expr.clone()
1312    };
1313
1314    // 4. Create the lineage node
1315    let mut node = LineageNode::new(&column_name, select_expr.clone(), node_source);
1316    apply_scope_context(&mut node, scope, source_name, reference_node_name);
1317
1318    // 5. Star handling — add downstream for each source
1319    if let Expression::Star(star) = &select_expr {
1320        let star_table = star
1321            .table
1322            .as_ref()
1323            .map(|identifier| identifier.name.as_str());
1324        for (name, source_info) in &scope.sources {
1325            if let Some(star_table) = star_table {
1326                let table_matches = name.eq_ignore_ascii_case(star_table)
1327                    || source_info
1328                        .alias
1329                        .as_deref()
1330                        .is_some_and(|alias| alias.eq_ignore_ascii_case(star_table))
1331                    || matches!(
1332                        &source_info.expression,
1333                        Expression::Table(table_ref)
1334                            if table_name_from_table_ref(table_ref).eq_ignore_ascii_case(star_table)
1335                    );
1336                if !table_matches {
1337                    continue;
1338                }
1339            }
1340
1341            let mut child = LineageNode::new(
1342                format!("{}.*", name),
1343                Expression::Star(crate::expressions::Star {
1344                    table: star.table.clone(),
1345                    except: None,
1346                    replace: None,
1347                    rename: None,
1348                    trailing_comments: vec![],
1349                    span: None,
1350                }),
1351                source_info.expression.clone(),
1352            );
1353            apply_source_info_context(&mut child, name, source_info);
1354            node.downstream.push(child);
1355        }
1356        return Ok(node);
1357    }
1358
1359    // 6. Subqueries in select — trace through scalar subqueries
1360    for query in query_expressions_in_scope(&select_expr) {
1361        for &sq_scope_id in &context.indexed(scope_id).subquery_scopes {
1362            if context.scope(sq_scope_id).expression == *query {
1363                if let Ok(child) = to_node_inner(
1364                    ColumnRef::Index(0),
1365                    context,
1366                    sq_scope_id,
1367                    dialect,
1368                    &column_name,
1369                    "",
1370                    "",
1371                    trim_selects,
1372                    &descendant_cte_scopes,
1373                    depth + 1,
1374                ) {
1375                    node.downstream.push(child);
1376                }
1377                break;
1378            }
1379        }
1380    }
1381
1382    // 7. Column references — trace each column to its source
1383    let col_refs = find_column_refs_in_expr_with_select(&select_expr, effective_expr, dialect);
1384    for col_ref in col_refs {
1385        let col_name = &col_ref.column;
1386        if let Some(ref table_id) = col_ref.table {
1387            let tbl = &table_id.name;
1388            resolve_qualified_column(
1389                &mut node,
1390                context,
1391                scope_id,
1392                dialect,
1393                tbl,
1394                col_name,
1395                &column_name,
1396                trim_selects,
1397                &all_cte_scopes,
1398                depth,
1399            );
1400        } else {
1401            if let Some(alias_expr) =
1402                find_prior_select_alias_expr(effective_expr, &select_expr, col_name, dialect)
1403            {
1404                for alias_ref in
1405                    find_column_refs_in_expr_with_select(&alias_expr, effective_expr, dialect)
1406                {
1407                    if let Some(ref table_id) = alias_ref.table {
1408                        resolve_qualified_column(
1409                            &mut node,
1410                            context,
1411                            scope_id,
1412                            dialect,
1413                            &table_id.name,
1414                            &alias_ref.column,
1415                            &column_name,
1416                            trim_selects,
1417                            &all_cte_scopes,
1418                            depth,
1419                        );
1420                    } else {
1421                        resolve_unqualified_column(
1422                            &mut node,
1423                            context,
1424                            scope_id,
1425                            dialect,
1426                            &alias_ref.column,
1427                            &column_name,
1428                            trim_selects,
1429                            &all_cte_scopes,
1430                            depth,
1431                        );
1432                    }
1433                }
1434                continue;
1435            }
1436
1437            resolve_unqualified_column(
1438                &mut node,
1439                context,
1440                scope_id,
1441                dialect,
1442                col_name,
1443                &column_name,
1444                trim_selects,
1445                &all_cte_scopes,
1446                depth,
1447            );
1448        }
1449    }
1450
1451    Ok(node)
1452}
1453
1454fn descendant_cte_scope_ids(all_cte_scopes: &[ScopeId], current_scope: ScopeId) -> Vec<ScopeId> {
1455    all_cte_scopes
1456        .iter()
1457        .copied()
1458        .filter(|scope| *scope != current_scope)
1459        .collect()
1460}
1461
1462fn effective_scope_expression(expr: &Expression) -> &Expression {
1463    match expr {
1464        Expression::Cte(cte) => effective_scope_expression(&cte.this),
1465        Expression::Subquery(subquery) => effective_scope_expression(&subquery.this),
1466        Expression::Paren(paren) => effective_scope_expression(&paren.this),
1467        other => other,
1468    }
1469}
1470
1471fn query_expressions_in_scope(expr: &Expression) -> Vec<&Expression> {
1472    let mut queries = Vec::new();
1473    let mut seen = HashSet::new();
1474
1475    for node in find_all_in_scope(
1476        expr,
1477        |node| {
1478            matches!(
1479                node,
1480                Expression::Subquery(subquery) if subquery.alias.is_none()
1481            ) || matches!(
1482                node,
1483                Expression::Exists(_) | Expression::In(_) | Expression::Any(_) | Expression::All(_)
1484            )
1485        },
1486        false,
1487    ) {
1488        let query = match node {
1489            Expression::Subquery(subquery) if subquery.alias.is_none() => Some(&subquery.this),
1490            Expression::Exists(exists) => Some(&exists.this),
1491            Expression::In(in_expr) => in_expr.query.as_ref(),
1492            Expression::Any(quantified) | Expression::All(quantified) => Some(&quantified.subquery),
1493            _ => None,
1494        };
1495
1496        if let Some(query) = query {
1497            let key = query as *const Expression as usize;
1498            if seen.insert(key) {
1499                queries.push(query);
1500            }
1501        }
1502    }
1503
1504    queries
1505}
1506
1507// ---------------------------------------------------------------------------
1508// Set operation handling
1509// ---------------------------------------------------------------------------
1510
1511fn set_branch_metadata(expression: &Expression, ordinal: usize) -> SetBranch {
1512    match expression {
1513        Expression::Union(union) => SetBranch {
1514            operator: SetOperator::Union,
1515            ordinal,
1516            all: union.all,
1517        },
1518        Expression::Intersect(intersect) => SetBranch {
1519            operator: SetOperator::Intersect,
1520            ordinal,
1521            all: intersect.all,
1522        },
1523        Expression::Except(except) => SetBranch {
1524            operator: SetOperator::Except,
1525            ordinal,
1526            all: except.all,
1527        },
1528        _ => unreachable!("set-operation metadata requires a set operation"),
1529    }
1530}
1531
1532fn handle_set_operation(
1533    column: &ColumnRef<'_>,
1534    context: &LineageScopeContext,
1535    scope_id: ScopeId,
1536    scope_expr: &Expression,
1537    scope_type_override: Option<ScopeType>,
1538    dialect: Option<DialectType>,
1539    scope_name: &str,
1540    source_name: &str,
1541    reference_node_name: &str,
1542    trim_selects: bool,
1543    ancestor_cte_scopes: &[ScopeId],
1544    depth: usize,
1545) -> Result<LineageNode> {
1546    let scope = context.scope(scope_id);
1547    let trace_wildcard_by_name =
1548        matches!(column, ColumnRef::Name(name) if normalize_column_name(name, dialect) == "*");
1549
1550    let aligned_layout = match crate::set_operation::set_operation_layout(scope_expr, dialect) {
1551        Ok(layout) => layout,
1552        Err(error) if error.is_indeterminate() => None,
1553        Err(error) => return Err(Error::invalid_input(error.to_string())),
1554    };
1555
1556    // Determine column index
1557    let col_index = match column {
1558        ColumnRef::Name(_) if trace_wildcard_by_name => 0,
1559        ColumnRef::Name(name) => column_to_index(scope_expr, name, dialect)?,
1560        ColumnRef::Index(i) => *i,
1561    };
1562
1563    if aligned_layout
1564        .as_ref()
1565        .is_some_and(|layout| col_index >= layout.outputs.len())
1566    {
1567        return Err(ordinal_resolution_error(
1568            col_index,
1569            ColumnResolutionReason::NotFound,
1570        ));
1571    }
1572
1573    let col_name = match column {
1574        ColumnRef::Name(name) => name.to_string(),
1575        ColumnRef::Index(_) => aligned_layout
1576            .as_ref()
1577            .and_then(|layout| layout.outputs.get(col_index))
1578            .map(|output| output.identifier.name.clone())
1579            .unwrap_or_else(|| format!("_{col_index}")),
1580    };
1581
1582    let mut node = LineageNode::new(&col_name, scope_expr.clone(), scope_expr.clone());
1583    if let Some(scope_type) = scope_type_override {
1584        apply_scope_context_with_type(
1585            &mut node,
1586            scope,
1587            scope_type,
1588            source_name,
1589            reference_node_name,
1590        );
1591    } else {
1592        apply_scope_context(&mut node, scope, source_name, reference_node_name);
1593    }
1594
1595    let mut resolution_failure = None;
1596
1597    // Recurse into each set-operation branch. Resolution failures are branch-local,
1598    // but genuine parser/internal errors must not be silently discarded.
1599    for (branch_ordinal, &branch_scope_id) in
1600        context.indexed(scope_id).union_scopes.iter().enumerate()
1601    {
1602        let branch_column = if trace_wildcard_by_name {
1603            match column {
1604                ColumnRef::Name(name) => ColumnRef::Name(name),
1605                ColumnRef::Index(_) => unreachable!("wildcard tracing is name-based"),
1606            }
1607        } else if let Some(layout) = &aligned_layout {
1608            let output = &layout.outputs[col_index];
1609            let branch_index = if branch_ordinal == 0 {
1610                output.left_ordinal
1611            } else {
1612                output.right_ordinal
1613            };
1614            let Some(branch_index) = branch_index else {
1615                // This branch contributes a dialect-inserted NULL for the
1616                // result slot, so it has no source expression to trace.
1617                continue;
1618            };
1619            ColumnRef::Index(branch_index)
1620        } else {
1621            ColumnRef::Index(col_index)
1622        };
1623
1624        match to_node_inner(
1625            branch_column,
1626            context,
1627            branch_scope_id,
1628            dialect,
1629            scope_name,
1630            "",
1631            "",
1632            trim_selects,
1633            ancestor_cte_scopes,
1634            depth + 1,
1635        ) {
1636            Ok(mut child) => {
1637                child.set_branch = Some(set_branch_metadata(scope_expr, branch_ordinal));
1638                node.downstream.push(child);
1639            }
1640            Err(Error::ColumnResolution { reason, .. }) => {
1641                resolution_failure = Some(merge_resolution_reason(resolution_failure, reason));
1642            }
1643            Err(error) => return Err(error),
1644        }
1645    }
1646
1647    if node.downstream.is_empty() {
1648        if let Some(reason) = resolution_failure {
1649            let target = match column {
1650                ColumnRef::Name(name) => ColumnResolutionTarget::Name {
1651                    name: name.to_string(),
1652                },
1653                ColumnRef::Index(ordinal) => ColumnResolutionTarget::Ordinal { ordinal: *ordinal },
1654            };
1655            return Err(column_resolution_error(target, reason));
1656        }
1657    }
1658
1659    Ok(node)
1660}
1661
1662fn merge_resolution_reason(
1663    current: Option<ColumnResolutionReason>,
1664    candidate: ColumnResolutionReason,
1665) -> ColumnResolutionReason {
1666    match (current, candidate) {
1667        (Some(ColumnResolutionReason::Ambiguous), _) | (_, ColumnResolutionReason::Ambiguous) => {
1668            ColumnResolutionReason::Ambiguous
1669        }
1670        (Some(ColumnResolutionReason::Indeterminate), _)
1671        | (_, ColumnResolutionReason::Indeterminate) => ColumnResolutionReason::Indeterminate,
1672        _ => ColumnResolutionReason::NotFound,
1673    }
1674}
1675
1676// ---------------------------------------------------------------------------
1677// Column resolution helpers
1678// ---------------------------------------------------------------------------
1679
1680fn resolve_qualified_column(
1681    node: &mut LineageNode,
1682    context: &LineageScopeContext,
1683    scope_id: ScopeId,
1684    dialect: Option<DialectType>,
1685    table: &str,
1686    col_name: &str,
1687    parent_name: &str,
1688    trim_selects: bool,
1689    all_cte_scopes: &[ScopeId],
1690    depth: usize,
1691) {
1692    let scope = context.scope(scope_id);
1693    // Resolve CTE alias: if `table` is a FROM alias for a CTE (e.g., `FROM my_cte AS t`),
1694    // resolve it to the actual CTE name so the CTE scope lookup succeeds.
1695    let resolved_cte_name = resolve_cte_alias(scope, table);
1696    let effective_table = resolved_cte_name.as_deref().unwrap_or(table);
1697
1698    if let Some(source_info) = scope
1699        .sources
1700        .get(table)
1701        .or_else(|| scope.sources.get(effective_table))
1702    {
1703        match &source_info.expression {
1704            Expression::Pivot(pivot) => {
1705                if attach_pivot_dependencies(
1706                    node,
1707                    context,
1708                    scope_id,
1709                    dialect,
1710                    pivot,
1711                    col_name,
1712                    trim_selects,
1713                    all_cte_scopes,
1714                    depth,
1715                ) {
1716                    return;
1717                }
1718            }
1719            Expression::Unpivot(unpivot) => {
1720                if attach_unpivot_dependencies(
1721                    node,
1722                    context,
1723                    scope_id,
1724                    dialect,
1725                    unpivot,
1726                    col_name,
1727                    trim_selects,
1728                    all_cte_scopes,
1729                    depth,
1730                ) {
1731                    return;
1732                }
1733            }
1734            _ => {}
1735        }
1736    }
1737
1738    // Check if table is a CTE reference — check both the current scope's cte_sources
1739    // and ancestor CTE scopes (for sibling CTEs in parent WITH clauses).
1740    let is_cte = scope.cte_sources.contains_key(effective_table)
1741        || all_cte_scopes.iter().any(
1742            |scope_id| matches!(&context.scope(*scope_id).expression, Expression::Cte(cte) if cte.alias.name == effective_table),
1743        );
1744    if is_cte {
1745        if let Some(child_scope_id) =
1746            find_child_scope_in(context, all_cte_scopes, scope_id, effective_table)
1747        {
1748            if let Ok(child) = to_node_inner(
1749                ColumnRef::Name(col_name),
1750                context,
1751                child_scope_id,
1752                dialect,
1753                parent_name,
1754                effective_table,
1755                parent_name,
1756                trim_selects,
1757                all_cte_scopes,
1758                depth + 1,
1759            ) {
1760                node.downstream.push(child);
1761                return;
1762            }
1763        }
1764
1765        if let Some(source_info) = scope
1766            .sources
1767            .get(table)
1768            .or_else(|| scope.sources.get(effective_table))
1769            .filter(|source_info| source_info.kind == SourceKind::Cte)
1770        {
1771            node.downstream.push(make_table_column_node_from_source(
1772                effective_table,
1773                col_name,
1774                source_info,
1775            ));
1776            return;
1777        }
1778    }
1779
1780    // Check if table is a derived table (is_scope = true in sources)
1781    if let Some(source_info) = scope.sources.get(table) {
1782        if source_info.is_scope {
1783            if let Some(child_scope_id) = find_child_scope(context, scope_id, table) {
1784                if let Ok(child) = to_node_inner(
1785                    ColumnRef::Name(col_name),
1786                    context,
1787                    child_scope_id,
1788                    dialect,
1789                    parent_name,
1790                    table,
1791                    parent_name,
1792                    trim_selects,
1793                    all_cte_scopes,
1794                    depth + 1,
1795                ) {
1796                    node.downstream.push(child);
1797                    return;
1798                }
1799            }
1800        }
1801    }
1802
1803    // Base table source found in current scope: preserve alias in the display name
1804    // but store the resolved table expression and name for downstream consumers.
1805    if let Some(source_info) = scope.sources.get(table) {
1806        if !source_info.is_scope {
1807            let mut child = make_table_column_node_from_source(table, col_name, source_info);
1808            if source_info.kind == SourceKind::Virtual {
1809                attach_virtual_source_dependencies(
1810                    &mut child,
1811                    context,
1812                    scope_id,
1813                    dialect,
1814                    table,
1815                    &source_info.expression,
1816                    trim_selects,
1817                    all_cte_scopes,
1818                    depth,
1819                );
1820            }
1821            node.downstream.push(child);
1822            return;
1823        }
1824    }
1825
1826    // Base table or unresolved — terminal node
1827    let mut child = make_table_column_node(table, col_name);
1828    if context.conservative {
1829        child.source_kind = SourceKind::Unknown;
1830    }
1831    node.downstream.push(child);
1832}
1833
1834fn attach_pivot_dependencies(
1835    node: &mut LineageNode,
1836    context: &LineageScopeContext,
1837    scope_id: ScopeId,
1838    dialect: Option<DialectType>,
1839    pivot: &crate::expressions::Pivot,
1840    col_name: &str,
1841    trim_selects: bool,
1842    all_cte_scopes: &[ScopeId],
1843    depth: usize,
1844) -> bool {
1845    if pivot.unpivot {
1846        return false;
1847    }
1848
1849    let scope = context.scope(scope_id);
1850    let mapping = pivot_lineage_column_mapping(pivot, scope, dialect);
1851    let Some(input_columns) = mapping.get(&normalize_column_name(col_name, dialect)) else {
1852        if pivot_implicit_source_column(pivot, col_name) {
1853            let col_ref = SimpleColumnRef {
1854                table: None,
1855                column: col_name.to_string(),
1856            };
1857            attach_pivot_input_column(
1858                node,
1859                context,
1860                scope_id,
1861                dialect,
1862                &pivot.this,
1863                &col_ref,
1864                trim_selects,
1865                all_cte_scopes,
1866                depth,
1867            );
1868            return true;
1869        }
1870        return false;
1871    };
1872
1873    // A generated output aggregates its inputs for each value of the pivot field.
1874    let aggregation = pivot_output_aggregation(pivot, scope, col_name, dialect);
1875    let field_columns = match aggregation {
1876        Some(_) => pivot_field_columns(pivot, dialect),
1877        None => Vec::new(),
1878    };
1879    let mut step = aggregation.map(|aggregation| {
1880        LineageNode::new(node.name.clone(), aggregation.clone(), node.source.clone())
1881    });
1882    let target = step.as_mut().unwrap_or(&mut *node);
1883    for col_ref in input_columns.iter().chain(&field_columns) {
1884        attach_pivot_input_column(
1885            target,
1886            context,
1887            scope_id,
1888            dialect,
1889            &pivot.this,
1890            col_ref,
1891            trim_selects,
1892            all_cte_scopes,
1893            depth,
1894        );
1895    }
1896    node.downstream.extend(step);
1897    true
1898}
1899
1900fn attach_unpivot_dependencies(
1901    node: &mut LineageNode,
1902    context: &LineageScopeContext,
1903    scope_id: ScopeId,
1904    dialect: Option<DialectType>,
1905    unpivot: &crate::expressions::Unpivot,
1906    col_name: &str,
1907    trim_selects: bool,
1908    all_cte_scopes: &[ScopeId],
1909    depth: usize,
1910) -> bool {
1911    let mapping = unpivot_column_mapping(unpivot, dialect);
1912    let normalized = normalize_column_name(col_name, dialect);
1913    let Some(input_columns) = mapping.get(&normalized) else {
1914        // A column outside the UNPIVOT passes through from the input.
1915        let unpivoted = unpivot
1916            .columns
1917            .iter()
1918            .flat_map(unpivot_entry_columns)
1919            .any(|column| normalize_column_name(&column.column, dialect) == normalized);
1920        if unpivoted {
1921            return false;
1922        }
1923        let col_ref = SimpleColumnRef {
1924            table: None,
1925            column: col_name.to_string(),
1926        };
1927        attach_pivot_input_column(
1928            node,
1929            context,
1930            scope_id,
1931            dialect,
1932            &unpivot.this,
1933            &col_ref,
1934            trim_selects,
1935            all_cte_scopes,
1936            depth,
1937        );
1938        return true;
1939    };
1940
1941    for col_ref in input_columns {
1942        attach_pivot_input_column(
1943            node,
1944            context,
1945            scope_id,
1946            dialect,
1947            &unpivot.this,
1948            col_ref,
1949            trim_selects,
1950            all_cte_scopes,
1951            depth,
1952        );
1953    }
1954    true
1955}
1956
1957fn pivot_column_mapping(
1958    pivot: &crate::expressions::Pivot,
1959    dialect: Option<DialectType>,
1960) -> HashMap<String, Vec<SimpleColumnRef>> {
1961    let aggregations = pivot_aggregation_expressions(pivot);
1962    let output_columns = pivot_generated_output_columns(pivot, dialect);
1963    if aggregations.is_empty() || output_columns.is_empty() {
1964        return HashMap::new();
1965    }
1966
1967    let mut mapping = HashMap::new();
1968    for (agg_index, agg) in aggregations.iter().enumerate() {
1969        let input_columns = find_column_refs_in_expr(agg, dialect);
1970        if input_columns.is_empty() {
1971            continue;
1972        }
1973        for col_index in (agg_index..output_columns.len()).step_by(aggregations.len()) {
1974            mapping.insert(
1975                normalize_column_name(&output_columns[col_index], dialect),
1976                input_columns.clone(),
1977            );
1978        }
1979    }
1980    mapping
1981}
1982
1983fn pivot_lineage_column_mapping(
1984    pivot: &crate::expressions::Pivot,
1985    scope: &Scope,
1986    dialect: Option<DialectType>,
1987) -> HashMap<String, Vec<SimpleColumnRef>> {
1988    let mut mapping = pivot_column_mapping(pivot, dialect);
1989    let Some(pre_pivot_columns) = pre_pivot_output_columns(&pivot.this, scope) else {
1990        return mapping;
1991    };
1992
1993    let output_columns = pivot_output_columns(pivot, &pre_pivot_columns, dialect);
1994    if output_columns.is_empty() {
1995        return mapping;
1996    }
1997
1998    let base_mapping = mapping.clone();
1999    for (post_name, pre_name) in output_columns {
2000        let normalized_pre = normalize_column_name(&pre_name, dialect);
2001        let normalized_post = normalize_column_name(&post_name, dialect);
2002
2003        if let Some(input_columns) = base_mapping.get(&normalized_pre) {
2004            mapping.insert(normalized_post, input_columns.clone());
2005        } else {
2006            mapping.insert(
2007                normalized_post,
2008                vec![SimpleColumnRef {
2009                    table: None,
2010                    column: pre_name,
2011                }],
2012            );
2013        }
2014    }
2015
2016    mapping
2017}
2018
2019/// The aggregation that computes the generated pivot output `col_name`.
2020fn pivot_output_aggregation<'a>(
2021    pivot: &'a crate::expressions::Pivot,
2022    scope: &Scope,
2023    col_name: &str,
2024    dialect: Option<DialectType>,
2025) -> Option<&'a Expression> {
2026    let mut name = normalize_column_name(col_name, dialect);
2027    // Pivot alias columns rename the outputs by position.
2028    if let Some(pre_columns) = pre_pivot_output_columns(&pivot.this, scope) {
2029        if let Some((_, pre)) = pivot_output_columns(pivot, &pre_columns, dialect)
2030            .into_iter()
2031            .find(|(post, _)| normalize_column_name(post, dialect) == name)
2032        {
2033            name = normalize_column_name(&pre, dialect);
2034        }
2035    }
2036    let aggregations = pivot_aggregation_expressions(pivot);
2037    pivot_generated_output_columns(pivot, dialect)
2038        .iter()
2039        .position(|output| normalize_column_name(output, dialect) == name)
2040        .and_then(|index| aggregations.get(index % aggregations.len()))
2041}
2042
2043/// The columns of the pivot field (`FOR method IN (...)`).
2044fn pivot_field_columns(
2045    pivot: &crate::expressions::Pivot,
2046    dialect: Option<DialectType>,
2047) -> Vec<SimpleColumnRef> {
2048    pivot
2049        .fields
2050        .iter()
2051        .filter_map(|field| match field {
2052            Expression::In(in_expr) => Some(&in_expr.this),
2053            _ => None,
2054        })
2055        .flat_map(|expr| find_column_refs_in_expr(expr, dialect))
2056        .collect()
2057}
2058
2059fn pre_pivot_output_columns(source: &Expression, scope: &Scope) -> Option<Vec<String>> {
2060    match source {
2061        Expression::Subquery(subquery) => known_output_columns(&subquery.this),
2062        Expression::Table(table) if table.schema.is_none() && table.catalog.is_none() => scope
2063            .cte_sources
2064            .get(&table.name.name)
2065            .and_then(|source| known_output_columns(&source.expression)),
2066        Expression::Paren(paren) => pre_pivot_output_columns(&paren.this, scope),
2067        _ => None,
2068    }
2069}
2070
2071fn known_output_columns(expression: &Expression) -> Option<Vec<String>> {
2072    let expression = match expression {
2073        Expression::Cte(cte) => &cte.this,
2074        Expression::Subquery(subquery) => &subquery.this,
2075        other => other,
2076    };
2077    let columns = crate::ast_transforms::get_output_column_names(expression);
2078    if columns.is_empty() || columns.iter().any(|column| column == "*") {
2079        None
2080    } else {
2081        Some(columns)
2082    }
2083}
2084
2085fn pivot_output_columns(
2086    pivot: &crate::expressions::Pivot,
2087    pre_pivot_columns: &[String],
2088    dialect: Option<DialectType>,
2089) -> Vec<(String, String)> {
2090    let generated_outputs = pivot_generated_output_columns(pivot, dialect);
2091    let excluded = pivot_excluded_source_columns(pivot, dialect);
2092
2093    if excluded.is_empty() || generated_outputs.is_empty() {
2094        return Vec::new();
2095    }
2096
2097    let mut pre_rename: Vec<String> = pre_pivot_columns
2098        .iter()
2099        .filter(|column| !excluded.contains(&normalize_column_name(column, dialect)))
2100        .cloned()
2101        .collect();
2102    pre_rename.extend(generated_outputs);
2103
2104    let post_rename = if pivot.alias_columns.is_empty() {
2105        pre_rename.clone()
2106    } else {
2107        let mut names: Vec<String> = pivot
2108            .alias_columns
2109            .iter()
2110            .map(|column| column.name.clone())
2111            .collect();
2112        names.extend(pre_rename.iter().skip(names.len()).cloned());
2113        names
2114    };
2115
2116    post_rename.into_iter().zip(pre_rename).collect()
2117}
2118
2119fn pivot_excluded_source_columns(
2120    pivot: &crate::expressions::Pivot,
2121    dialect: Option<DialectType>,
2122) -> HashSet<String> {
2123    pivot
2124        .fields
2125        .iter()
2126        .chain(pivot.expressions.iter())
2127        .chain(pivot.using.iter())
2128        .flat_map(|expr| find_column_refs_in_expr(expr, dialect))
2129        .map(|column| normalize_column_name(&column.column, dialect))
2130        .collect()
2131}
2132
2133fn pivot_generated_output_columns(
2134    pivot: &crate::expressions::Pivot,
2135    _dialect: Option<DialectType>,
2136) -> Vec<String> {
2137    let fields = pivot_field_output_names(pivot);
2138    if fields.is_empty() {
2139        return Vec::new();
2140    }
2141
2142    let aggregations = pivot_aggregation_expressions(pivot);
2143    if aggregations.is_empty() {
2144        return Vec::new();
2145    }
2146
2147    let needs_suffix = aggregations.len() > 1;
2148    let mut outputs = Vec::new();
2149    for field in fields {
2150        for aggregation in aggregations {
2151            if let Some(suffix) = pivot_aggregation_output_suffix(aggregation, needs_suffix) {
2152                outputs.push(format!("{field}_{suffix}"));
2153            } else {
2154                outputs.push(field.clone());
2155            }
2156        }
2157    }
2158    outputs
2159}
2160
2161fn pivot_aggregation_expressions(pivot: &crate::expressions::Pivot) -> &[Expression] {
2162    if pivot.using.is_empty() {
2163        &pivot.expressions
2164    } else {
2165        &pivot.using
2166    }
2167}
2168
2169fn pivot_aggregation_output_suffix(expr: &Expression, needs_suffix: bool) -> Option<String> {
2170    match expr {
2171        Expression::Alias(alias) => Some(alias.alias.name.clone()),
2172        _ if needs_suffix => pivot_generated_aggregation_suffix(expr),
2173        _ => None,
2174    }
2175}
2176
2177#[cfg(feature = "generate")]
2178fn pivot_generated_aggregation_suffix(expr: &Expression) -> Option<String> {
2179    Generator::sql(expr).ok().map(|sql| sql.to_lowercase())
2180}
2181
2182#[cfg(not(feature = "generate"))]
2183fn pivot_generated_aggregation_suffix(expr: &Expression) -> Option<String> {
2184    pivot_expr_output_name(expr).or_else(|| Some(expr.variant_name().to_string()))
2185}
2186
2187fn pivot_field_output_names(pivot: &crate::expressions::Pivot) -> Vec<String> {
2188    let mut names = Vec::new();
2189    for field in &pivot.fields {
2190        if let Expression::In(in_expr) = field {
2191            for expr in &in_expr.expressions {
2192                if let Some(name) = pivot_expr_output_name(expr) {
2193                    names.push(name);
2194                }
2195            }
2196        }
2197    }
2198    names
2199}
2200
2201fn pivot_expr_output_name(expr: &Expression) -> Option<String> {
2202    match expr {
2203        Expression::PivotAlias(alias) => pivot_expr_output_name(&alias.alias),
2204        Expression::Alias(alias) => Some(alias.alias.name.clone()),
2205        Expression::Identifier(identifier) => Some(identifier.name.clone()),
2206        Expression::Column(column) => Some(column.name.name.clone()),
2207        Expression::Literal(literal) => Some(literal.value_str().to_string()),
2208        Expression::Var(var) => Some(var.this.clone()),
2209        Expression::Tuple(tuple) => tuple.expressions.first().and_then(pivot_expr_output_name),
2210        _ => None,
2211    }
2212}
2213
2214fn pivot_implicit_source_column(pivot: &crate::expressions::Pivot, col_name: &str) -> bool {
2215    let pivot_columns: HashSet<String> = pivot
2216        .fields
2217        .iter()
2218        .filter_map(|field| match field {
2219            Expression::In(in_expr) => Some(&in_expr.this),
2220            _ => None,
2221        })
2222        .flat_map(|expr| find_column_refs_in_expr(expr, None))
2223        .map(|col| col.column.to_lowercase())
2224        .collect();
2225    let aggregation_columns: HashSet<String> = pivot
2226        .expressions
2227        .iter()
2228        .flat_map(|expr| find_column_refs_in_expr(expr, None))
2229        .map(|col| col.column.to_lowercase())
2230        .collect();
2231
2232    let normalized = col_name.to_lowercase();
2233    !pivot_columns.contains(&normalized) && !aggregation_columns.contains(&normalized)
2234}
2235
2236fn unpivot_column_mapping(
2237    unpivot: &crate::expressions::Unpivot,
2238    dialect: Option<DialectType>,
2239) -> HashMap<String, Vec<SimpleColumnRef>> {
2240    let value_columns: Vec<String> = std::iter::once(unpivot.value_column.name.clone())
2241        .chain(
2242            unpivot
2243                .extra_value_columns
2244                .iter()
2245                .map(|column| column.name.clone()),
2246        )
2247        .collect();
2248    let mut all_input_columns = Vec::new();
2249    let mut value_input_columns: Vec<Vec<SimpleColumnRef>> = vec![Vec::new(); value_columns.len()];
2250
2251    for entry in &unpivot.columns {
2252        let columns = unpivot_entry_columns(entry);
2253        all_input_columns.extend(columns.clone());
2254        if columns.len() == value_columns.len() {
2255            for (idx, col_ref) in columns.into_iter().enumerate() {
2256                value_input_columns[idx].push(col_ref);
2257            }
2258        } else {
2259            for inputs in &mut value_input_columns {
2260                inputs.extend(columns.clone());
2261            }
2262        }
2263    }
2264
2265    let mut mapping = HashMap::new();
2266    mapping.insert(
2267        normalize_column_name(&unpivot.name_column.name, dialect),
2268        all_input_columns.clone(),
2269    );
2270    for (idx, value_column) in value_columns.iter().enumerate() {
2271        mapping.insert(
2272            normalize_column_name(value_column, dialect),
2273            value_input_columns.get(idx).cloned().unwrap_or_default(),
2274        );
2275    }
2276    mapping
2277}
2278
2279fn unpivot_entry_columns(expr: &Expression) -> Vec<SimpleColumnRef> {
2280    match expr {
2281        Expression::PivotAlias(alias) => unpivot_entry_columns(&alias.this),
2282        Expression::Tuple(tuple) => tuple
2283            .expressions
2284            .iter()
2285            .flat_map(unpivot_entry_columns)
2286            .collect(),
2287        Expression::Column(column) => vec![SimpleColumnRef {
2288            table: column.table.clone(),
2289            column: column.name.name.clone(),
2290        }],
2291        Expression::Identifier(identifier) => vec![SimpleColumnRef {
2292            table: None,
2293            column: identifier.name.clone(),
2294        }],
2295        _ => find_column_refs_in_expr(expr, None),
2296    }
2297}
2298
2299fn attach_pivot_input_column(
2300    node: &mut LineageNode,
2301    context: &LineageScopeContext,
2302    scope_id: ScopeId,
2303    dialect: Option<DialectType>,
2304    source_expr: &Expression,
2305    col_ref: &SimpleColumnRef,
2306    trim_selects: bool,
2307    all_cte_scopes: &[ScopeId],
2308    depth: usize,
2309) {
2310    let scope = context.scope(scope_id);
2311    match source_expr {
2312        Expression::Table(table) => {
2313            let table_name = table.name.name.as_str();
2314            if table.schema.is_none()
2315                && table.catalog.is_none()
2316                && scope.cte_sources.contains_key(table_name)
2317            {
2318                // The visible source under this name is the pivot wrapper.
2319                // Resolve its input CTE directly instead of entering that same
2320                // wrapper again through resolve_qualified_column.
2321                if let Some(input_scope) =
2322                    find_child_scope_in(context, all_cte_scopes, scope_id, table_name)
2323                {
2324                    if let Ok(child) = to_node_inner(
2325                        ColumnRef::Name(&col_ref.column),
2326                        context,
2327                        input_scope,
2328                        dialect,
2329                        &node.name,
2330                        table_name,
2331                        &node.name,
2332                        trim_selects,
2333                        all_cte_scopes,
2334                        depth + 1,
2335                    ) {
2336                        node.downstream.push(child);
2337                    }
2338                }
2339            } else {
2340                let mut source = ScopeSourceInfo::new(
2341                    Expression::Table(Box::new(table.as_ref().clone())),
2342                    false,
2343                    SourceKind::Table,
2344                );
2345                if let Some(alias) = &table.alias {
2346                    source = source.with_alias(alias.name.clone());
2347                }
2348                let source_key = table
2349                    .alias
2350                    .as_ref()
2351                    .map(|alias| alias.name.as_str())
2352                    .unwrap_or(table.name.name.as_str());
2353                node.downstream.push(make_table_column_node_from_source(
2354                    source_key,
2355                    &col_ref.column,
2356                    &source,
2357                ));
2358            }
2359        }
2360        Expression::Subquery(subquery) => {
2361            let Some(source_scope_id) =
2362                find_derived_scope_for_query(context, scope_id, &subquery.this)
2363            else {
2364                return;
2365            };
2366            let child = if let Some(table) = &col_ref.table {
2367                let mut child_node = LineageNode::new(
2368                    &col_ref.column,
2369                    subquery.this.clone(),
2370                    subquery.this.clone(),
2371                );
2372                resolve_qualified_column(
2373                    &mut child_node,
2374                    context,
2375                    source_scope_id,
2376                    dialect,
2377                    &table.name,
2378                    &col_ref.column,
2379                    &node.name.clone(),
2380                    trim_selects,
2381                    all_cte_scopes,
2382                    depth + 1,
2383                );
2384                Ok(child_node)
2385            } else {
2386                to_node_inner(
2387                    ColumnRef::Name(&col_ref.column),
2388                    context,
2389                    source_scope_id,
2390                    dialect,
2391                    "",
2392                    "",
2393                    "",
2394                    trim_selects,
2395                    all_cte_scopes,
2396                    depth + 1,
2397                )
2398            };
2399            if let Ok(child) = child {
2400                node.downstream.push(child);
2401            }
2402        }
2403        Expression::Paren(paren) => attach_pivot_input_column(
2404            node,
2405            context,
2406            scope_id,
2407            dialect,
2408            &paren.this,
2409            col_ref,
2410            trim_selects,
2411            all_cte_scopes,
2412            depth,
2413        ),
2414        _ => {
2415            if let Some(table) = &col_ref.table {
2416                resolve_qualified_column(
2417                    node,
2418                    context,
2419                    scope_id,
2420                    dialect,
2421                    &table.name,
2422                    &col_ref.column,
2423                    &node.name.clone(),
2424                    trim_selects,
2425                    all_cte_scopes,
2426                    depth + 1,
2427                );
2428            } else {
2429                node.downstream
2430                    .push(make_table_column_node("_", &col_ref.column));
2431            }
2432        }
2433    }
2434}
2435
2436/// Resolve a FROM alias to the original CTE name.
2437///
2438/// When a query uses `FROM my_cte AS alias`, the scope's `sources` map contains
2439/// `"alias"` → CTE expression, but `cte_sources` only contains `"my_cte"`.
2440/// This function checks if `name` is such an alias and returns the CTE name.
2441fn resolve_cte_alias(scope: &Scope, name: &str) -> Option<String> {
2442    // If it's already a known CTE name, no resolution needed
2443    if scope.cte_sources.contains_key(name) {
2444        return None;
2445    }
2446    // Check if the source's expression is a CTE — if so, extract the CTE name
2447    if let Some(source_info) = scope.sources.get(name) {
2448        if source_info.is_scope {
2449            if let Expression::Cte(cte) = &source_info.expression {
2450                let cte_name = &cte.alias.name;
2451                if scope.cte_sources.contains_key(cte_name) {
2452                    return Some(cte_name.clone());
2453                }
2454            }
2455        }
2456    }
2457    None
2458}
2459
2460fn resolve_unqualified_column(
2461    node: &mut LineageNode,
2462    context: &LineageScopeContext,
2463    scope_id: ScopeId,
2464    dialect: Option<DialectType>,
2465    col_name: &str,
2466    parent_name: &str,
2467    trim_selects: bool,
2468    all_cte_scopes: &[ScopeId],
2469    depth: usize,
2470) {
2471    let scope = context.scope(scope_id);
2472    // Try to find which source this column belongs to.
2473    // Build the source list from the actual FROM/JOIN clauses to avoid
2474    // mixing in CTE definitions that are in scope but not referenced.
2475    let from_source_names = source_names_from_from_join(scope);
2476
2477    if let Some(tbl) = unique_virtual_source_for_column(scope, &from_source_names, col_name) {
2478        resolve_qualified_column(
2479            node,
2480            context,
2481            scope_id,
2482            dialect,
2483            &tbl,
2484            col_name,
2485            parent_name,
2486            trim_selects,
2487            all_cte_scopes,
2488            depth,
2489        );
2490        return;
2491    }
2492
2493    if from_source_names.len() == 1 {
2494        let tbl = &from_source_names[0];
2495        resolve_qualified_column(
2496            node,
2497            context,
2498            scope_id,
2499            dialect,
2500            tbl,
2501            col_name,
2502            parent_name,
2503            trim_selects,
2504            all_cte_scopes,
2505            depth,
2506        );
2507        return;
2508    }
2509
2510    // Multiple sources — can't resolve without schema info, add unqualified node
2511    let child = LineageNode::new(
2512        col_name.to_string(),
2513        Expression::Column(Box::new(crate::expressions::Column {
2514            name: crate::expressions::Identifier::new(col_name.to_string()),
2515            table: None,
2516            join_mark: false,
2517            trailing_comments: vec![],
2518            span: None,
2519            inferred_type: None,
2520        })),
2521        node.source.clone(),
2522    );
2523    node.downstream.push(child);
2524}
2525
2526fn unique_virtual_source_for_column(
2527    scope: &Scope,
2528    source_names: &[String],
2529    col_name: &str,
2530) -> Option<String> {
2531    let mut matches = source_names.iter().filter_map(|source_name| {
2532        let source = scope.sources.get(source_name)?;
2533        if source.kind == SourceKind::Virtual
2534            && virtual_source_output_columns(source)
2535                .any(|column| column.eq_ignore_ascii_case(col_name))
2536        {
2537            Some(source_name.clone())
2538        } else {
2539            None
2540        }
2541    });
2542
2543    let first = matches.next()?;
2544    if matches.next().is_none() {
2545        Some(first)
2546    } else {
2547        None
2548    }
2549}
2550
2551fn virtual_source_output_columns(
2552    source_info: &ScopeSourceInfo,
2553) -> Box<dyn Iterator<Item = String> + '_> {
2554    match &source_info.expression {
2555        Expression::Unnest(unnest) => Box::new(unnest_output_columns(unnest)),
2556        Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) => {
2557            Box::new(alias_output_columns(alias))
2558        }
2559        Expression::Lateral(lateral) => Box::new(lateral_output_columns(lateral)),
2560        Expression::LateralView(lateral_view) => {
2561            Box::new(lateral_view_output_columns(lateral_view))
2562        }
2563        _ => Box::new(source_info.alias.clone().into_iter()),
2564    }
2565}
2566
2567fn unnest_output_types(unnest: &crate::expressions::UnnestFunc) -> Vec<DataType> {
2568    let element_type = |expression: &Expression| match expression.inferred_type() {
2569        Some(DataType::Array { element_type, .. }) => (**element_type).clone(),
2570        _ => DataType::Unknown,
2571    };
2572
2573    let mut types = vec![unnest
2574        .inferred_type
2575        .clone()
2576        .unwrap_or_else(|| element_type(&unnest.this))];
2577    types.extend(unnest.expressions.iter().map(element_type));
2578    if unnest.with_ordinality || unnest.offset_alias.is_some() {
2579        types.push(DataType::BigInt { length: None });
2580    }
2581    types
2582}
2583
2584fn virtual_source_column_type(source_info: &ScopeSourceInfo, column: &str) -> Option<DataType> {
2585    let find_type = |names: Vec<String>, types: Vec<DataType>| {
2586        names
2587            .iter()
2588            .position(|name| name.eq_ignore_ascii_case(column))
2589            .and_then(|index| types.get(index).cloned())
2590    };
2591
2592    match &source_info.expression {
2593        Expression::Unnest(unnest) => find_type(
2594            unnest_output_columns(unnest).collect(),
2595            unnest_output_types(unnest),
2596        ),
2597        Expression::Alias(alias) => match &alias.this {
2598            Expression::Unnest(unnest) => find_type(
2599                alias_output_columns(alias).collect(),
2600                unnest_output_types(unnest),
2601            ),
2602            _ => None,
2603        },
2604        Expression::Lateral(lateral) => match lateral.this.as_ref() {
2605            Expression::Unnest(unnest) => find_type(
2606                lateral_output_columns(lateral).collect(),
2607                unnest_output_types(unnest),
2608            ),
2609            _ => None,
2610        },
2611        _ => None,
2612    }
2613}
2614
2615fn unnest_output_columns(
2616    unnest: &crate::expressions::UnnestFunc,
2617) -> impl Iterator<Item = String> + '_ {
2618    unnest
2619        .alias
2620        .iter()
2621        .map(|alias| alias.name.clone())
2622        .chain(unnest.offset_alias.iter().map(|alias| alias.name.clone()))
2623}
2624
2625fn alias_output_columns(
2626    alias: &crate::expressions::Alias,
2627) -> Box<dyn Iterator<Item = String> + '_> {
2628    if alias.column_aliases.is_empty() {
2629        Box::new(std::iter::once(alias.alias.name.clone()))
2630    } else {
2631        Box::new(
2632            alias
2633                .column_aliases
2634                .iter()
2635                .map(|column| column.name.clone()),
2636        )
2637    }
2638}
2639
2640fn lateral_output_columns(
2641    lateral: &crate::expressions::Lateral,
2642) -> Box<dyn Iterator<Item = String> + '_> {
2643    if lateral.column_aliases.is_empty() {
2644        default_virtual_output_columns(&lateral.this)
2645    } else {
2646        Box::new(lateral.column_aliases.iter().cloned())
2647    }
2648}
2649
2650fn lateral_view_output_columns(
2651    lateral_view: &crate::expressions::LateralView,
2652) -> Box<dyn Iterator<Item = String> + '_> {
2653    Box::new(
2654        lateral_view
2655            .column_aliases
2656            .iter()
2657            .map(|column| column.name.clone()),
2658    )
2659}
2660
2661fn default_virtual_output_columns(expr: &Expression) -> Box<dyn Iterator<Item = String> + '_> {
2662    match expr {
2663        Expression::Unnest(unnest) => Box::new(unnest_output_columns(unnest)),
2664        Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) => {
2665            alias_output_columns(alias)
2666        }
2667        Expression::Function(function) if function.name.eq_ignore_ascii_case("FLATTEN") => {
2668            Box::new(
2669                ["seq", "key", "path", "index", "value", "this"]
2670                    .into_iter()
2671                    .map(String::from),
2672            )
2673        }
2674        _ => Box::new(std::iter::empty()),
2675    }
2676}
2677
2678fn attach_virtual_source_dependencies(
2679    node: &mut LineageNode,
2680    context: &LineageScopeContext,
2681    scope_id: ScopeId,
2682    dialect: Option<DialectType>,
2683    source_alias: &str,
2684    source_expr: &Expression,
2685    trim_selects: bool,
2686    all_cte_scopes: &[ScopeId],
2687    depth: usize,
2688) {
2689    let scope = context.scope(scope_id);
2690    let parent_name = node.name.clone();
2691    let mut seen = HashSet::new();
2692    for col_ref in find_column_refs_in_expr(source_expr, dialect) {
2693        let key = (
2694            col_ref.table.as_ref().map(|t| t.name.clone()),
2695            col_ref.column.clone(),
2696        );
2697        if !seen.insert(key) {
2698            continue;
2699        }
2700
2701        if let Some(table_id) = col_ref.table {
2702            let table = table_id.name;
2703            if table == source_alias {
2704                continue;
2705            }
2706            resolve_qualified_column(
2707                node,
2708                context,
2709                scope_id,
2710                dialect,
2711                &table,
2712                &col_ref.column,
2713                &parent_name,
2714                trim_selects,
2715                all_cte_scopes,
2716                depth + 1,
2717            );
2718        } else {
2719            let non_virtual_sources = non_virtual_source_names_from_from_join(scope);
2720            if non_virtual_sources.len() == 1 {
2721                resolve_qualified_column(
2722                    node,
2723                    context,
2724                    scope_id,
2725                    dialect,
2726                    &non_virtual_sources[0],
2727                    &col_ref.column,
2728                    &parent_name,
2729                    trim_selects,
2730                    all_cte_scopes,
2731                    depth + 1,
2732                );
2733            }
2734        }
2735    }
2736}
2737
2738fn source_names_from_from_join(scope: &Scope) -> Vec<String> {
2739    fn source_name(expr: &Expression) -> Option<String> {
2740        match expr {
2741            Expression::Table(table) => Some(
2742                table
2743                    .alias
2744                    .as_ref()
2745                    .map(|a| a.name.clone())
2746                    .unwrap_or_else(|| table.name.name.clone()),
2747            ),
2748            Expression::Subquery(subquery) => {
2749                subquery.alias.as_ref().map(|alias| alias.name.clone())
2750            }
2751            Expression::Unnest(unnest) => unnest.alias.as_ref().map(|alias| alias.name.clone()),
2752            Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) => {
2753                Some(alias.alias.name.clone())
2754            }
2755            Expression::Alias(alias) if is_query_like_relation(&alias.this) => {
2756                Some(alias.alias.name.clone())
2757            }
2758            Expression::Lateral(lateral) => lateral.alias.clone(),
2759            Expression::LateralView(lateral_view) => lateral_view
2760                .table_alias
2761                .as_ref()
2762                .or_else(|| lateral_view.column_aliases.first())
2763                .map(|alias| alias.name.clone()),
2764            Expression::Pivot(pivot) => Some(pivot_lineage_source_name(
2765                &pivot.this,
2766                pivot.alias.as_ref().map(|alias| alias.name.as_str()),
2767            )),
2768            Expression::Unpivot(unpivot) => Some(pivot_lineage_source_name(
2769                &unpivot.this,
2770                unpivot.alias.as_ref().map(|alias| alias.name.as_str()),
2771            )),
2772            Expression::Paren(paren) => source_name(&paren.this),
2773            _ => None,
2774        }
2775    }
2776
2777    let effective_expr = match &scope.expression {
2778        Expression::Cte(cte) => &cte.this,
2779        expr => expr,
2780    };
2781
2782    let mut names = Vec::new();
2783    let mut seen = std::collections::HashSet::new();
2784
2785    if let Expression::Select(select) = effective_expr {
2786        if let Some(from) = &select.from {
2787            for expr in &from.expressions {
2788                if let Some(name) = source_name(expr) {
2789                    if !name.is_empty() && seen.insert(name.clone()) {
2790                        names.push(name);
2791                    }
2792                }
2793            }
2794        }
2795        for join in &select.joins {
2796            if is_semi_or_anti_join_kind(join.kind) {
2797                continue;
2798            }
2799            if let Some(name) = source_name(&join.this) {
2800                if !name.is_empty() && seen.insert(name.clone()) {
2801                    names.push(name);
2802                }
2803            }
2804        }
2805        for lateral_view in &select.lateral_views {
2806            if let Some(name) =
2807                source_name(&Expression::LateralView(Box::new(lateral_view.clone())))
2808            {
2809                if !name.is_empty() && seen.insert(name.clone()) {
2810                    names.push(name);
2811                }
2812            }
2813        }
2814    }
2815
2816    names
2817}
2818
2819fn is_semi_or_anti_join_kind(kind: JoinKind) -> bool {
2820    matches!(
2821        kind,
2822        JoinKind::Semi
2823            | JoinKind::Anti
2824            | JoinKind::LeftSemi
2825            | JoinKind::LeftAnti
2826            | JoinKind::RightSemi
2827            | JoinKind::RightAnti
2828    )
2829}
2830
2831fn is_query_like_relation(expr: &Expression) -> bool {
2832    match expr {
2833        Expression::Select(_)
2834        | Expression::Subquery(_)
2835        | Expression::Union(_)
2836        | Expression::Intersect(_)
2837        | Expression::Except(_) => true,
2838        Expression::Paren(paren) => is_query_like_relation(&paren.this),
2839        _ => false,
2840    }
2841}
2842
2843fn derived_source_query(expr: &Expression) -> Option<&Expression> {
2844    match expr {
2845        Expression::Subquery(subquery) => Some(&subquery.this),
2846        Expression::Alias(alias) if is_query_like_relation(&alias.this) => Some(&alias.this),
2847        Expression::Select(_)
2848        | Expression::Union(_)
2849        | Expression::Intersect(_)
2850        | Expression::Except(_) => Some(expr),
2851        Expression::Paren(paren) => derived_source_query(&paren.this),
2852        _ => None,
2853    }
2854}
2855
2856fn expressions_equivalent_after_wrappers(left: &Expression, right: &Expression) -> bool {
2857    left == right || effective_scope_expression(left) == effective_scope_expression(right)
2858}
2859
2860fn non_virtual_source_names_from_from_join(scope: &Scope) -> Vec<String> {
2861    source_names_from_from_join(scope)
2862        .into_iter()
2863        .filter(|name| {
2864            !matches!(
2865                scope.sources.get(name).map(|source| source.kind),
2866                Some(SourceKind::Virtual)
2867            )
2868        })
2869        .collect()
2870}
2871
2872// ---------------------------------------------------------------------------
2873// Helper functions
2874// ---------------------------------------------------------------------------
2875
2876#[derive(Debug, Clone)]
2877struct OutputLayoutEntry {
2878    column: OutputColumn,
2879    projection_index: usize,
2880}
2881
2882fn query_output_from_expression(
2883    expression: &Expression,
2884    dialect: Option<DialectType>,
2885) -> Result<QueryOutput> {
2886    match crate::set_operation::set_operation_layout(expression, dialect) {
2887        Ok(Some(layout)) => {
2888            return Ok(QueryOutput {
2889                columns: layout
2890                    .outputs
2891                    .into_iter()
2892                    .enumerate()
2893                    .map(|(ordinal, output)| OutputColumn::Named {
2894                        name: output.identifier.name,
2895                        ordinal: Some(ordinal),
2896                    })
2897                    .collect(),
2898                ordinal_complete: true,
2899            })
2900        }
2901        Ok(None) => {}
2902        Err(error) if error.is_indeterminate() => {}
2903        Err(error) => return Err(Error::invalid_input(error.to_string())),
2904    }
2905
2906    let select = leftmost_output_select(expression).ok_or_else(|| {
2907        Error::invalid_input("output_columns requires a SELECT or set-operation query")
2908    })?;
2909    let entries = output_layout(select);
2910    let ordinal_complete = !entries
2911        .iter()
2912        .any(|entry| matches!(entry.column, OutputColumn::Wildcard { .. }));
2913
2914    Ok(QueryOutput {
2915        columns: entries.into_iter().map(|entry| entry.column).collect(),
2916        ordinal_complete,
2917    })
2918}
2919
2920fn leftmost_output_select(expression: &Expression) -> Option<&Select> {
2921    match expression {
2922        Expression::Select(select) => Some(select),
2923        Expression::Union(set_op) => leftmost_output_select(&set_op.left),
2924        Expression::Intersect(set_op) => leftmost_output_select(&set_op.left),
2925        Expression::Except(set_op) => leftmost_output_select(&set_op.left),
2926        Expression::Subquery(subquery) => leftmost_output_select(&subquery.this),
2927        Expression::Cte(cte) => leftmost_output_select(&cte.this),
2928        Expression::Paren(paren) => leftmost_output_select(&paren.this),
2929        _ => None,
2930    }
2931}
2932
2933fn output_layout(select: &Select) -> Vec<OutputLayoutEntry> {
2934    let mut entries = Vec::new();
2935    let mut next_ordinal = Some(0usize);
2936
2937    for (projection_index, projection) in select.expressions.iter().enumerate() {
2938        let projection = unwrap_output_annotation(projection);
2939
2940        if let Some(qualifier) = output_wildcard_qualifier(projection) {
2941            entries.push(OutputLayoutEntry {
2942                column: OutputColumn::Wildcard {
2943                    qualifier,
2944                    start_ordinal: next_ordinal,
2945                },
2946                projection_index,
2947            });
2948            next_ordinal = None;
2949            continue;
2950        }
2951
2952        if let Expression::Aliases(aliases) = projection {
2953            if !aliases.expressions.is_empty() {
2954                for alias in &aliases.expressions {
2955                    let ordinal = take_output_ordinal(&mut next_ordinal);
2956                    let column = get_alias_or_name(alias)
2957                        .map(|name| OutputColumn::Named { name, ordinal })
2958                        .unwrap_or(OutputColumn::Unnamed { ordinal });
2959                    entries.push(OutputLayoutEntry {
2960                        column,
2961                        projection_index,
2962                    });
2963                }
2964                continue;
2965            }
2966        }
2967
2968        let ordinal = take_output_ordinal(&mut next_ordinal);
2969        let column = get_alias_or_name(projection)
2970            .map(|name| OutputColumn::Named { name, ordinal })
2971            .unwrap_or(OutputColumn::Unnamed { ordinal });
2972        entries.push(OutputLayoutEntry {
2973            column,
2974            projection_index,
2975        });
2976    }
2977
2978    entries
2979}
2980
2981fn take_output_ordinal(next_ordinal: &mut Option<usize>) -> Option<usize> {
2982    let ordinal = *next_ordinal;
2983    if let Some(value) = ordinal {
2984        *next_ordinal = Some(value + 1);
2985    }
2986    ordinal
2987}
2988
2989fn unwrap_output_annotation(mut expression: &Expression) -> &Expression {
2990    while let Expression::Annotated(annotated) = expression {
2991        expression = &annotated.this;
2992    }
2993    expression
2994}
2995
2996/// Return `Some(qualifier)` for a wildcard, where the inner option is the qualifier.
2997fn output_wildcard_qualifier(expression: &Expression) -> Option<Option<String>> {
2998    match expression {
2999        Expression::Star(star) => Some(star.table.as_ref().map(|table| table.name.clone())),
3000        Expression::Column(column) if column.name.name == "*" => {
3001            Some(column.table.as_ref().map(|table| table.name.clone()))
3002        }
3003        _ => None,
3004    }
3005}
3006
3007fn column_resolution_error(
3008    target: ColumnResolutionTarget,
3009    reason: ColumnResolutionReason,
3010) -> Error {
3011    Error::column_resolution(target, reason)
3012}
3013
3014fn name_resolution_error(name: &str, reason: ColumnResolutionReason) -> Error {
3015    column_resolution_error(
3016        ColumnResolutionTarget::Name {
3017            name: name.to_string(),
3018        },
3019        reason,
3020    )
3021}
3022
3023fn ordinal_resolution_error(ordinal: usize, reason: ColumnResolutionReason) -> Error {
3024    column_resolution_error(ColumnResolutionTarget::Ordinal { ordinal }, reason)
3025}
3026
3027fn find_select_expr_by_name(
3028    select: &Select,
3029    name: &str,
3030    dialect: Option<DialectType>,
3031) -> Result<Expression> {
3032    let normalized_name = normalize_column_name(name, dialect);
3033    let layout = output_layout(select);
3034    let mut matches = Vec::new();
3035
3036    for entry in &layout {
3037        let is_match = match &entry.column {
3038            OutputColumn::Named {
3039                name: output_name, ..
3040            } => normalize_column_name(output_name, dialect) == normalized_name,
3041            OutputColumn::Wildcard { .. } => normalized_name == "*",
3042            OutputColumn::Unnamed { .. } => false,
3043        };
3044        if is_match {
3045            matches.push(entry.projection_index);
3046        }
3047    }
3048    match matches.as_slice() {
3049        [projection_index] => return Ok(select.expressions[*projection_index].clone()),
3050        [_, ..] => {
3051            return Err(name_resolution_error(
3052                name,
3053                ColumnResolutionReason::Ambiguous,
3054            ))
3055        }
3056        [] => {}
3057    }
3058
3059    if let Some(expression) = synthesize_star_passthrough_expr(select, name, dialect) {
3060        return Ok(expression);
3061    }
3062
3063    let reason = if layout
3064        .iter()
3065        .any(|entry| matches!(entry.column, OutputColumn::Wildcard { .. }))
3066    {
3067        ColumnResolutionReason::Indeterminate
3068    } else {
3069        ColumnResolutionReason::NotFound
3070    };
3071    Err(name_resolution_error(name, reason))
3072}
3073
3074fn find_select_expr_by_ordinal(select: &Select, ordinal: usize) -> Result<Expression> {
3075    let layout = output_layout(select);
3076
3077    for entry in &layout {
3078        match &entry.column {
3079            OutputColumn::Named {
3080                ordinal: Some(candidate),
3081                ..
3082            }
3083            | OutputColumn::Unnamed {
3084                ordinal: Some(candidate),
3085            } if *candidate == ordinal => {
3086                return Ok(select.expressions[entry.projection_index].clone())
3087            }
3088            OutputColumn::Wildcard { start_ordinal, .. } => {
3089                if match start_ordinal {
3090                    Some(start) => ordinal >= *start,
3091                    None => true,
3092                } {
3093                    return Err(ordinal_resolution_error(
3094                        ordinal,
3095                        ColumnResolutionReason::Indeterminate,
3096                    ));
3097                }
3098            }
3099            _ => {}
3100        }
3101    }
3102
3103    Err(ordinal_resolution_error(
3104        ordinal,
3105        ColumnResolutionReason::NotFound,
3106    ))
3107}
3108
3109fn output_name_to_ordinal(
3110    expression: &Expression,
3111    name: &str,
3112    dialect: Option<DialectType>,
3113) -> Result<usize> {
3114    match crate::set_operation::set_operation_layout(expression, dialect) {
3115        Ok(Some(layout)) => {
3116            let lookup = Identifier::new(name);
3117            let lookup_key = crate::set_operation::identifier_key(&lookup, dialect);
3118            let matches: Vec<_> = layout
3119                .outputs
3120                .iter()
3121                .enumerate()
3122                .filter(|(_, output)| {
3123                    (output.identifier.quoted && output.identifier.name == name)
3124                        || crate::set_operation::identifier_key(&output.identifier, dialect)
3125                            == lookup_key
3126                })
3127                .map(|(ordinal, _)| ordinal)
3128                .collect();
3129
3130            return match matches.as_slice() {
3131                [ordinal] => Ok(*ordinal),
3132                [_, ..] => Err(name_resolution_error(
3133                    name,
3134                    ColumnResolutionReason::Ambiguous,
3135                )),
3136                [] => Err(name_resolution_error(
3137                    name,
3138                    ColumnResolutionReason::NotFound,
3139                )),
3140            };
3141        }
3142        Ok(None) => {}
3143        Err(error) if error.is_indeterminate() => {}
3144        Err(error) => return Err(Error::invalid_input(error.to_string())),
3145    }
3146
3147    let select = leftmost_output_select(expression).ok_or_else(|| {
3148        Error::invalid_input("column resolution requires a SELECT or set-operation query")
3149    })?;
3150    let normalized_name = normalize_column_name(name, dialect);
3151    let layout = output_layout(select);
3152    let mut matches = Vec::new();
3153
3154    for entry in &layout {
3155        if let OutputColumn::Named {
3156            name: output_name,
3157            ordinal,
3158        } = &entry.column
3159        {
3160            if normalize_column_name(output_name, dialect) == normalized_name {
3161                matches.push(*ordinal);
3162            }
3163        }
3164    }
3165
3166    if matches.len() > 1 {
3167        return Err(name_resolution_error(
3168            name,
3169            ColumnResolutionReason::Ambiguous,
3170        ));
3171    }
3172    if let Some(ordinal) = matches.into_iter().next() {
3173        return ordinal
3174            .ok_or_else(|| name_resolution_error(name, ColumnResolutionReason::Indeterminate));
3175    }
3176
3177    let reason = if layout
3178        .iter()
3179        .any(|entry| matches!(entry.column, OutputColumn::Wildcard { .. }))
3180    {
3181        ColumnResolutionReason::Indeterminate
3182    } else {
3183        ColumnResolutionReason::NotFound
3184    };
3185    Err(name_resolution_error(name, reason))
3186}
3187
3188/// Get the alias or name of an expression
3189fn get_alias_or_name(expr: &Expression) -> Option<String> {
3190    match expr {
3191        Expression::Alias(alias) => Some(alias.alias.name.clone()),
3192        Expression::Column(col) => Some(col.name.name.clone()),
3193        Expression::Identifier(id) => Some(id.name.clone()),
3194        Expression::Star(_) => Some("*".to_string()),
3195        // Annotated wraps an expression with trailing comments (e.g. `SELECT\n-- comment\na`).
3196        // Unwrap to get the actual column/alias name from the inner expression.
3197        Expression::Annotated(a) => get_alias_or_name(&a.this),
3198        _ => None,
3199    }
3200}
3201
3202fn find_prior_select_alias_expr(
3203    scope_expr: &Expression,
3204    target_expr: &Expression,
3205    alias_name: &str,
3206    dialect: Option<DialectType>,
3207) -> Option<Expression> {
3208    let Expression::Select(select) = scope_expr else {
3209        return None;
3210    };
3211
3212    let normalized_alias = normalize_column_name(alias_name, dialect);
3213    for expr in &select.expressions {
3214        if expr == target_expr {
3215            return None;
3216        }
3217
3218        if let Expression::Alias(alias) = expr {
3219            if normalize_column_name(&alias.alias.name, dialect) == normalized_alias {
3220                return Some(alias.this.clone());
3221            }
3222        }
3223    }
3224
3225    None
3226}
3227
3228/// Resolve the display name for a column reference.
3229fn resolve_column_name(column: &ColumnRef<'_>, select_expr: &Expression) -> String {
3230    match column {
3231        ColumnRef::Name(n) => n.to_string(),
3232        ColumnRef::Index(_) => get_alias_or_name(select_expr).unwrap_or_else(|| "?".to_string()),
3233    }
3234}
3235
3236/// Find the select expression matching a column reference.
3237fn find_select_expr(
3238    scope_expr: &Expression,
3239    column: &ColumnRef<'_>,
3240    dialect: Option<DialectType>,
3241) -> Result<Expression> {
3242    if let Expression::Select(ref select) = scope_expr {
3243        match column {
3244            ColumnRef::Name(name) => find_select_expr_by_name(select, name, dialect),
3245            ColumnRef::Index(ordinal) => find_select_expr_by_ordinal(select, *ordinal),
3246        }
3247    } else {
3248        Err(Error::invalid_input(
3249            "column resolution requires a SELECT expression",
3250        ))
3251    }
3252}
3253
3254fn synthesize_star_passthrough_expr(
3255    select: &Select,
3256    name: &str,
3257    dialect: Option<DialectType>,
3258) -> Option<Expression> {
3259    let sources = get_select_sources(select);
3260    if sources.is_empty() {
3261        return None;
3262    }
3263
3264    let mut candidates = Vec::new();
3265    let mut seen = HashSet::new();
3266    let mut replacement = None;
3267
3268    for expr in &select.expressions {
3269        let aliases = match star_passthrough_source_aliases(expr, &sources) {
3270            StarPassthroughSources::None => continue,
3271            StarPassthroughSources::Ambiguous => return None,
3272            StarPassthroughSources::Aliases(aliases) => aliases,
3273        };
3274
3275        // Apply the star modifiers: the output `name` may read another column, or none.
3276        let column = match passthrough_star(expr) {
3277            Some(star) if star.has_except_pattern() => return None,
3278            Some(star) => {
3279                let column = match star.source_column(name, dialect) {
3280                    Some(column) => column,
3281                    None => continue,
3282                };
3283                if let Some(alias) = star.replacement(&column, dialect) {
3284                    if replacement.is_some() {
3285                        return None;
3286                    }
3287                    let mut alias = alias.clone();
3288                    if let Some(renamed) = star.renamed(&column, dialect) {
3289                        alias.alias = renamed;
3290                    }
3291                    replacement = Some(Expression::Alias(Box::new(alias)));
3292                    continue;
3293                }
3294                column
3295            }
3296            None => name.to_string(),
3297        };
3298
3299        for alias in aliases {
3300            if seen.insert((alias.clone(), column.clone())) {
3301                candidates.push((alias, column.clone()));
3302            }
3303        }
3304    }
3305
3306    match (replacement, candidates.as_slice()) {
3307        (Some(replacement), []) => Some(replacement),
3308        (None, [(alias, column)]) => {
3309            let table = Identifier::new(alias.clone());
3310            Some(make_column_expr(column, Some(&table)))
3311        }
3312        _ => None,
3313    }
3314}
3315
3316fn passthrough_star(expr: &Expression) -> Option<&Star> {
3317    match expr {
3318        Expression::Star(star) => Some(star),
3319        Expression::Annotated(annotated) => passthrough_star(&annotated.this),
3320        _ => None,
3321    }
3322}
3323
3324enum StarPassthroughSources {
3325    None,
3326    Ambiguous,
3327    Aliases(Vec<String>),
3328}
3329
3330fn star_passthrough_source_aliases(
3331    expr: &Expression,
3332    sources: &[SourceInfo],
3333) -> StarPassthroughSources {
3334    match expr {
3335        Expression::Star(star) => star_source_aliases(star.table.as_ref(), sources),
3336        Expression::Column(column) if column.name.name == "*" => {
3337            star_source_aliases(column.table.as_ref(), sources)
3338        }
3339        Expression::Annotated(annotated) => {
3340            star_passthrough_source_aliases(&annotated.this, sources)
3341        }
3342        _ => StarPassthroughSources::None,
3343    }
3344}
3345
3346fn star_source_aliases(
3347    qualifier: Option<&Identifier>,
3348    sources: &[SourceInfo],
3349) -> StarPassthroughSources {
3350    if let Some(qualifier) = qualifier {
3351        let mut aliases = Vec::new();
3352
3353        for source in sources {
3354            if source_matches_star_qualifier(source, qualifier) {
3355                aliases.push(source.alias.clone());
3356            }
3357        }
3358
3359        return match aliases.len() {
3360            0 => StarPassthroughSources::None,
3361            1 => StarPassthroughSources::Aliases(aliases),
3362            _ => StarPassthroughSources::Ambiguous,
3363        };
3364    }
3365
3366    match sources {
3367        // Do not synthesize a source column for unresolved quoted table stars.
3368        // This keeps quoted CTE/table case semantics intact while still allowing
3369        // the schema-less fallback for common unquoted SELECT * passthroughs.
3370        [source] if source.quoted => StarPassthroughSources::None,
3371        [source] => StarPassthroughSources::Aliases(vec![source.alias.clone()]),
3372        [] => StarPassthroughSources::None,
3373        _ => StarPassthroughSources::Ambiguous,
3374    }
3375}
3376
3377fn source_matches_star_qualifier(source: &SourceInfo, qualifier: &Identifier) -> bool {
3378    if source.normalized == normalize_cte_name(qualifier) {
3379        return true;
3380    }
3381
3382    if qualifier.quoted {
3383        source.alias == qualifier.name
3384    } else {
3385        source.alias.eq_ignore_ascii_case(&qualifier.name)
3386    }
3387}
3388
3389/// Find the positional index of a column name in a set operation's first SELECT branch.
3390fn column_to_index(
3391    set_op_expr: &Expression,
3392    name: &str,
3393    dialect: Option<DialectType>,
3394) -> Result<usize> {
3395    output_name_to_ordinal(set_op_expr, name, dialect)
3396}
3397
3398fn normalize_column_name(name: &str, dialect: Option<DialectType>) -> String {
3399    normalize_name(name, dialect, false, true)
3400}
3401
3402/// If trim_selects is enabled, return a copy of the SELECT with only the target column.
3403fn trim_source(select_expr: &Expression, target_expr: &Expression) -> Expression {
3404    if let Expression::Select(select) = select_expr {
3405        let mut trimmed = select.as_ref().clone();
3406        trimmed.expressions = vec![target_expr.clone()];
3407        Expression::Select(Box::new(trimmed))
3408    } else {
3409        select_expr.clone()
3410    }
3411}
3412
3413/// Find the child scope (CTE or derived table) for a given source name.
3414fn find_child_scope(
3415    context: &LineageScopeContext,
3416    scope_id: ScopeId,
3417    source_name: &str,
3418) -> Option<ScopeId> {
3419    let indexed = context.indexed(scope_id);
3420    let scope = &indexed.scope;
3421
3422    // Check CTE scopes
3423    if scope.cte_sources.contains_key(source_name) {
3424        for &cte_scope_id in &indexed.cte_scopes {
3425            let cte_scope = context.scope(cte_scope_id);
3426            if let Expression::Cte(cte) = &cte_scope.expression {
3427                if cte.alias.name == source_name {
3428                    return Some(cte_scope_id);
3429                }
3430            }
3431        }
3432    }
3433
3434    // Check derived table scopes
3435    if let Some(source_info) = scope.sources.get(source_name) {
3436        if source_info.is_scope && !scope.cte_sources.contains_key(source_name) {
3437            if let Some(query) = derived_source_query(&source_info.expression) {
3438                for &dt_scope_id in &indexed.derived_table_scopes {
3439                    let dt_scope = context.scope(dt_scope_id);
3440                    if expressions_equivalent_after_wrappers(&dt_scope.expression, query) {
3441                        return Some(dt_scope_id);
3442                    }
3443                }
3444            }
3445        }
3446    }
3447
3448    None
3449}
3450
3451/// Find a CTE scope by name, searching through a combined list of CTE scopes.
3452/// This handles nested CTEs where the current scope doesn't have the CTE scope
3453/// as a direct child but knows about it via cte_sources.
3454fn find_child_scope_in(
3455    context: &LineageScopeContext,
3456    all_cte_scopes: &[ScopeId],
3457    scope_id: ScopeId,
3458    source_name: &str,
3459) -> Option<ScopeId> {
3460    let indexed = context.indexed(scope_id);
3461    let scope = &indexed.scope;
3462
3463    // First try the scope's own cte_scopes
3464    for &cte_scope_id in &indexed.cte_scopes {
3465        let cte_scope = context.scope(cte_scope_id);
3466        if let Expression::Cte(cte) = &cte_scope.expression {
3467            if cte.alias.name == source_name {
3468                return Some(cte_scope_id);
3469            }
3470        }
3471    }
3472
3473    // Then search through all ancestor CTE scopes
3474    for &cte_scope_id in all_cte_scopes {
3475        let cte_scope = context.scope(cte_scope_id);
3476        if let Expression::Cte(cte) = &cte_scope.expression {
3477            if cte.alias.name == source_name {
3478                return Some(cte_scope_id);
3479            }
3480        }
3481    }
3482
3483    // Fall back to derived table scopes
3484    if let Some(source_info) = scope.sources.get(source_name) {
3485        if source_info.is_scope {
3486            if let Some(query) = derived_source_query(&source_info.expression) {
3487                for &dt_scope_id in &indexed.derived_table_scopes {
3488                    let dt_scope = context.scope(dt_scope_id);
3489                    if expressions_equivalent_after_wrappers(&dt_scope.expression, query) {
3490                        return Some(dt_scope_id);
3491                    }
3492                }
3493            }
3494        }
3495    }
3496
3497    None
3498}
3499
3500fn find_derived_scope_for_query(
3501    context: &LineageScopeContext,
3502    scope_id: ScopeId,
3503    query: &Expression,
3504) -> Option<ScopeId> {
3505    context
3506        .indexed(scope_id)
3507        .derived_table_scopes
3508        .iter()
3509        .copied()
3510        .find(|derived_scope_id| {
3511            expressions_equivalent_after_wrappers(
3512                &context.scope(*derived_scope_id).expression,
3513                query,
3514            )
3515        })
3516}
3517
3518/// Create a terminal lineage node for a table.column reference.
3519fn make_table_column_node(table: &str, column: &str) -> LineageNode {
3520    let mut node = LineageNode::new(
3521        format!("{}.{}", table, column),
3522        Expression::Column(Box::new(crate::expressions::Column {
3523            name: crate::expressions::Identifier::new(column.to_string()),
3524            table: Some(crate::expressions::Identifier::new(table.to_string())),
3525            join_mark: false,
3526            trailing_comments: vec![],
3527            span: None,
3528            inferred_type: None,
3529        })),
3530        Expression::Table(Box::new(crate::expressions::TableRef::new(table))),
3531    );
3532    node.source_name = table.to_string();
3533    node.source_kind = SourceKind::Table;
3534    node
3535}
3536
3537fn table_name_from_table_ref(table_ref: &crate::expressions::TableRef) -> String {
3538    let mut parts: Vec<String> = Vec::new();
3539    if let Some(catalog) = &table_ref.catalog {
3540        parts.push(catalog.name.clone());
3541    }
3542    if let Some(schema) = &table_ref.schema {
3543        parts.push(schema.name.clone());
3544    }
3545    parts.push(table_ref.name.name.clone());
3546    parts.join(".")
3547}
3548
3549fn apply_source_info_context(
3550    node: &mut LineageNode,
3551    source_key: &str,
3552    source_info: &ScopeSourceInfo,
3553) {
3554    node.source_kind = source_info.kind;
3555    node.source_name =
3556        source_info
3557            .lineage_name
3558            .clone()
3559            .unwrap_or_else(|| match &source_info.expression {
3560                Expression::Table(table_ref) => table_name_from_table_ref(table_ref),
3561                _ => source_key.to_string(),
3562            });
3563    node.source_alias = source_info.alias.clone();
3564}
3565
3566fn make_table_column_node_from_source(
3567    source_key: &str,
3568    column: &str,
3569    source_info: &ScopeSourceInfo,
3570) -> LineageNode {
3571    let lineage_name = source_info.lineage_name.as_deref().unwrap_or(source_key);
3572    let inferred_type = (source_info.kind == SourceKind::Virtual)
3573        .then(|| virtual_source_column_type(source_info, column))
3574        .flatten();
3575    let mut node = LineageNode::new(
3576        format!("{}.{}", lineage_name, column),
3577        Expression::Column(Box::new(crate::expressions::Column {
3578            name: crate::expressions::Identifier::new(column.to_string()),
3579            table: Some(crate::expressions::Identifier::new(
3580                lineage_name.to_string(),
3581            )),
3582            join_mark: false,
3583            trailing_comments: vec![],
3584            span: None,
3585            inferred_type,
3586        })),
3587        source_info.expression.clone(),
3588    );
3589
3590    apply_source_info_context(&mut node, source_key, source_info);
3591
3592    node
3593}
3594
3595/// Simple column reference extracted from an expression
3596#[derive(Debug, Clone)]
3597struct SimpleColumnRef {
3598    table: Option<crate::expressions::Identifier>,
3599    column: String,
3600}
3601
3602/// Find all column references in an expression (does not recurse into subqueries).
3603fn find_column_refs_in_expr(
3604    expr: &Expression,
3605    dialect: Option<DialectType>,
3606) -> Vec<SimpleColumnRef> {
3607    let mut refs = Vec::new();
3608    collect_column_refs(expr, dialect, &mut refs, None);
3609    refs
3610}
3611
3612fn find_column_refs_in_expr_with_select(
3613    expr: &Expression,
3614    select_expr: &Expression,
3615    dialect: Option<DialectType>,
3616) -> Vec<SimpleColumnRef> {
3617    let named_windows = match select_expr {
3618        Expression::Select(select) => select.windows.as_deref(),
3619        _ => None,
3620    };
3621    let mut refs = Vec::new();
3622    collect_column_refs(expr, dialect, &mut refs, named_windows);
3623    refs
3624}
3625
3626fn is_bigquery_safe_namespace_receiver(expr: &Expression) -> bool {
3627    match expr {
3628        Expression::Column(col) => {
3629            col.table.is_none() && !col.name.quoted && col.name.name.eq_ignore_ascii_case("SAFE")
3630        }
3631        Expression::Identifier(id) => !id.quoted && id.name.eq_ignore_ascii_case("SAFE"),
3632        _ => false,
3633    }
3634}
3635
3636fn collect_column_refs(
3637    expr: &Expression,
3638    dialect: Option<DialectType>,
3639    refs: &mut Vec<SimpleColumnRef>,
3640    named_windows: Option<&[NamedWindow]>,
3641) {
3642    let mut stack: Vec<&Expression> = vec![expr];
3643
3644    while let Some(current) = stack.pop() {
3645        match current {
3646            // === Leaf: collect Column references ===
3647            Expression::Column(col) => {
3648                refs.push(SimpleColumnRef {
3649                    table: col.table.clone(),
3650                    column: col.name.name.clone(),
3651                });
3652            }
3653
3654            // === Boundary: don't recurse into subqueries (handled separately) ===
3655            Expression::Subquery(_) | Expression::Exists(_) => {}
3656
3657            // === BinaryOp variants: left, right ===
3658            Expression::And(op)
3659            | Expression::Or(op)
3660            | Expression::Eq(op)
3661            | Expression::Neq(op)
3662            | Expression::Lt(op)
3663            | Expression::Lte(op)
3664            | Expression::Gt(op)
3665            | Expression::Gte(op)
3666            | Expression::Add(op)
3667            | Expression::Sub(op)
3668            | Expression::Mul(op)
3669            | Expression::Div(op)
3670            | Expression::Mod(op)
3671            | Expression::BitwiseAnd(op)
3672            | Expression::BitwiseOr(op)
3673            | Expression::BitwiseXor(op)
3674            | Expression::BitwiseLeftShift(op)
3675            | Expression::BitwiseRightShift(op)
3676            | Expression::Concat(op)
3677            | Expression::Adjacent(op)
3678            | Expression::TsMatch(op)
3679            | Expression::PropertyEQ(op)
3680            | Expression::ArrayContainsAll(op)
3681            | Expression::ArrayContainedBy(op)
3682            | Expression::ArrayOverlaps(op)
3683            | Expression::JSONBContainsAllTopKeys(op)
3684            | Expression::JSONBContainsAnyTopKeys(op)
3685            | Expression::JSONBDeleteAtPath(op)
3686            | Expression::ExtendsLeft(op)
3687            | Expression::ExtendsRight(op)
3688            | Expression::Is(op)
3689            | Expression::MemberOf(op)
3690            | Expression::NullSafeEq(op)
3691            | Expression::NullSafeNeq(op)
3692            | Expression::Glob(op)
3693            | Expression::Match(op) => {
3694                stack.push(&op.left);
3695                stack.push(&op.right);
3696            }
3697
3698            // === UnaryOp variants: this ===
3699            Expression::Not(u) | Expression::Neg(u) | Expression::BitwiseNot(u) => {
3700                stack.push(&u.this);
3701            }
3702
3703            // === UnaryFunc variants: this ===
3704            Expression::Upper(f)
3705            | Expression::Lower(f)
3706            | Expression::Length(f)
3707            | Expression::LTrim(f)
3708            | Expression::RTrim(f)
3709            | Expression::Reverse(f)
3710            | Expression::Abs(f)
3711            | Expression::Sqrt(f)
3712            | Expression::Cbrt(f)
3713            | Expression::Ln(f)
3714            | Expression::Exp(f)
3715            | Expression::Sign(f)
3716            | Expression::Date(f)
3717            | Expression::Time(f)
3718            | Expression::DateFromUnixDate(f)
3719            | Expression::UnixDate(f)
3720            | Expression::UnixSeconds(f)
3721            | Expression::UnixMillis(f)
3722            | Expression::UnixMicros(f)
3723            | Expression::TimeStrToDate(f)
3724            | Expression::DateToDi(f)
3725            | Expression::DiToDate(f)
3726            | Expression::TsOrDiToDi(f)
3727            | Expression::TsOrDsToDatetime(f)
3728            | Expression::TsOrDsToTimestamp(f)
3729            | Expression::YearOfWeek(f)
3730            | Expression::YearOfWeekIso(f)
3731            | Expression::Initcap(f)
3732            | Expression::Ascii(f)
3733            | Expression::Chr(f)
3734            | Expression::Soundex(f)
3735            | Expression::ByteLength(f)
3736            | Expression::Hex(f)
3737            | Expression::LowerHex(f)
3738            | Expression::Unicode(f)
3739            | Expression::Radians(f)
3740            | Expression::Degrees(f)
3741            | Expression::Sin(f)
3742            | Expression::Cos(f)
3743            | Expression::Tan(f)
3744            | Expression::Asin(f)
3745            | Expression::Acos(f)
3746            | Expression::Atan(f)
3747            | Expression::IsNan(f)
3748            | Expression::IsInf(f)
3749            | Expression::ArrayLength(f)
3750            | Expression::ArraySize(f)
3751            | Expression::Cardinality(f)
3752            | Expression::ArrayReverse(f)
3753            | Expression::ArrayDistinct(f)
3754            | Expression::ArrayFlatten(f)
3755            | Expression::ArrayCompact(f)
3756            | Expression::Explode(f)
3757            | Expression::ExplodeOuter(f)
3758            | Expression::ToArray(f)
3759            | Expression::MapFromEntries(f)
3760            | Expression::MapKeys(f)
3761            | Expression::MapValues(f)
3762            | Expression::JsonArrayLength(f)
3763            | Expression::JsonKeys(f)
3764            | Expression::JsonType(f)
3765            | Expression::ParseJson(f)
3766            | Expression::ToJson(f)
3767            | Expression::Typeof(f)
3768            | Expression::BitwiseCount(f)
3769            | Expression::Year(f)
3770            | Expression::Month(f)
3771            | Expression::Day(f)
3772            | Expression::Hour(f)
3773            | Expression::Minute(f)
3774            | Expression::Second(f)
3775            | Expression::DayOfWeek(f)
3776            | Expression::DayOfWeekIso(f)
3777            | Expression::DayOfMonth(f)
3778            | Expression::DayOfYear(f)
3779            | Expression::WeekOfYear(f)
3780            | Expression::Quarter(f)
3781            | Expression::Epoch(f)
3782            | Expression::EpochMs(f)
3783            | Expression::TimeStrToUnix(f)
3784            | Expression::SHA(f)
3785            | Expression::SHA1Digest(f)
3786            | Expression::TimeToUnix(f)
3787            | Expression::JSONBool(f)
3788            | Expression::Int64(f)
3789            | Expression::MD5NumberLower64(f)
3790            | Expression::MD5NumberUpper64(f)
3791            | Expression::DateStrToDate(f)
3792            | Expression::DateToDateStr(f) => {
3793                stack.push(&f.this);
3794            }
3795
3796            // === BinaryFunc variants: this, expression ===
3797            Expression::Power(f)
3798            | Expression::NullIf(f)
3799            | Expression::IfNull(f)
3800            | Expression::Nvl(f)
3801            | Expression::UnixToTimeStr(f)
3802            | Expression::Contains(f)
3803            | Expression::StartsWith(f)
3804            | Expression::EndsWith(f)
3805            | Expression::Levenshtein(f)
3806            | Expression::ModFunc(f)
3807            | Expression::Atan2(f)
3808            | Expression::IntDiv(f)
3809            | Expression::AddMonths(f)
3810            | Expression::MonthsBetween(f)
3811            | Expression::NextDay(f)
3812            | Expression::ArrayContains(f)
3813            | Expression::ArrayPosition(f)
3814            | Expression::ArrayAppend(f)
3815            | Expression::ArrayPrepend(f)
3816            | Expression::ArrayUnion(f)
3817            | Expression::ArrayExcept(f)
3818            | Expression::ArrayRemove(f)
3819            | Expression::StarMap(f)
3820            | Expression::MapFromArrays(f)
3821            | Expression::MapContainsKey(f)
3822            | Expression::ElementAt(f)
3823            | Expression::JsonMergePatch(f)
3824            | Expression::JSONBContains(f)
3825            | Expression::JSONBExtract(f) => {
3826                stack.push(&f.this);
3827                stack.push(&f.expression);
3828            }
3829
3830            // === VarArgFunc variants: expressions ===
3831            Expression::Greatest(f)
3832            | Expression::Least(f)
3833            | Expression::Coalesce(f)
3834            | Expression::ArrayConcat(f)
3835            | Expression::ArrayIntersect(f)
3836            | Expression::ArrayZip(f)
3837            | Expression::MapConcat(f)
3838            | Expression::JsonArray(f) => {
3839                for e in &f.expressions {
3840                    stack.push(e);
3841                }
3842            }
3843
3844            // === AggFunc variants: this, filter, having_max, limit ===
3845            Expression::Sum(f)
3846            | Expression::Avg(f)
3847            | Expression::Min(f)
3848            | Expression::Max(f)
3849            | Expression::ArrayAgg(f)
3850            | Expression::CountIf(f)
3851            | Expression::Stddev(f)
3852            | Expression::StddevPop(f)
3853            | Expression::StddevSamp(f)
3854            | Expression::Variance(f)
3855            | Expression::VarPop(f)
3856            | Expression::VarSamp(f)
3857            | Expression::Median(f)
3858            | Expression::Mode(f)
3859            | Expression::First(f)
3860            | Expression::Last(f)
3861            | Expression::AnyValue(f)
3862            | Expression::ApproxDistinct(f)
3863            | Expression::ApproxCountDistinct(f)
3864            | Expression::LogicalAnd(f)
3865            | Expression::LogicalOr(f)
3866            | Expression::Skewness(f)
3867            | Expression::ArrayConcatAgg(f)
3868            | Expression::ArrayUniqueAgg(f)
3869            | Expression::BoolXorAgg(f)
3870            | Expression::BitwiseAndAgg(f)
3871            | Expression::BitwiseOrAgg(f)
3872            | Expression::BitwiseXorAgg(f) => {
3873                stack.push(&f.this);
3874                if let Some(ref filter) = f.filter {
3875                    stack.push(filter);
3876                }
3877                if let Some((ref expr, _)) = f.having_max {
3878                    stack.push(expr);
3879                }
3880                if let Some(ref limit) = f.limit {
3881                    stack.push(limit);
3882                }
3883            }
3884
3885            // === Generic Function / AggregateFunction: args ===
3886            Expression::Function(func) => {
3887                for arg in &func.args {
3888                    stack.push(arg);
3889                }
3890            }
3891            Expression::AggregateFunction(func) => {
3892                for arg in &func.args {
3893                    stack.push(arg);
3894                }
3895                if let Some(ref filter) = func.filter {
3896                    stack.push(filter);
3897                }
3898                if let Some(ref limit) = func.limit {
3899                    stack.push(limit);
3900                }
3901            }
3902
3903            // === WindowFunction: this (skip Over for lineage purposes) ===
3904            Expression::WindowFunction(wf) => {
3905                stack.push(&wf.this);
3906                for e in &wf.over.partition_by {
3907                    stack.push(e);
3908                }
3909                for e in &wf.over.order_by {
3910                    stack.push(&e.this);
3911                }
3912                if let Some(keep) = &wf.keep {
3913                    for e in &keep.order_by {
3914                        stack.push(&e.this);
3915                    }
3916                }
3917                if let (Some(window_name), Some(named_windows)) =
3918                    (&wf.over.window_name, named_windows)
3919                {
3920                    for named_window in named_windows {
3921                        if named_window
3922                            .name
3923                            .name
3924                            .eq_ignore_ascii_case(&window_name.name)
3925                        {
3926                            for e in &named_window.spec.partition_by {
3927                                stack.push(e);
3928                            }
3929                            for e in &named_window.spec.order_by {
3930                                stack.push(&e.this);
3931                            }
3932                        }
3933                    }
3934                }
3935            }
3936
3937            // === Containers and special expressions ===
3938            Expression::Alias(a) => {
3939                stack.push(&a.this);
3940            }
3941            Expression::Cast(c) | Expression::TryCast(c) | Expression::SafeCast(c) => {
3942                stack.push(&c.this);
3943                if let Some(ref fmt) = c.format {
3944                    stack.push(fmt);
3945                }
3946                if let Some(ref def) = c.default {
3947                    stack.push(def);
3948                }
3949            }
3950            Expression::Paren(p) => {
3951                stack.push(&p.this);
3952            }
3953            Expression::Annotated(a) => {
3954                stack.push(&a.this);
3955            }
3956            Expression::Case(case) => {
3957                if let Some(ref operand) = case.operand {
3958                    stack.push(operand);
3959                }
3960                for (cond, result) in &case.whens {
3961                    stack.push(cond);
3962                    stack.push(result);
3963                }
3964                if let Some(ref else_expr) = case.else_ {
3965                    stack.push(else_expr);
3966                }
3967            }
3968            Expression::Collation(c) => {
3969                stack.push(&c.this);
3970            }
3971            Expression::In(i) => {
3972                stack.push(&i.this);
3973                for e in &i.expressions {
3974                    stack.push(e);
3975                }
3976                if let Some(ref q) = i.query {
3977                    stack.push(q);
3978                }
3979                if let Some(ref u) = i.unnest {
3980                    stack.push(u);
3981                }
3982            }
3983            Expression::Between(b) => {
3984                stack.push(&b.this);
3985                stack.push(&b.low);
3986                stack.push(&b.high);
3987            }
3988            Expression::IsNull(n) => {
3989                stack.push(&n.this);
3990            }
3991            Expression::IsTrue(t) | Expression::IsFalse(t) => {
3992                stack.push(&t.this);
3993            }
3994            Expression::IsJson(j) => {
3995                stack.push(&j.this);
3996            }
3997            Expression::Like(l) | Expression::ILike(l) => {
3998                stack.push(&l.left);
3999                stack.push(&l.right);
4000                if let Some(ref esc) = l.escape {
4001                    stack.push(esc);
4002                }
4003            }
4004            Expression::SimilarTo(s) => {
4005                stack.push(&s.this);
4006                stack.push(&s.pattern);
4007                if let Some(ref esc) = s.escape {
4008                    stack.push(esc);
4009                }
4010            }
4011            Expression::Ordered(o) => {
4012                stack.push(&o.this);
4013            }
4014            Expression::Array(a) => {
4015                for e in &a.expressions {
4016                    stack.push(e);
4017                }
4018            }
4019            Expression::Tuple(t) => {
4020                for e in &t.expressions {
4021                    stack.push(e);
4022                }
4023            }
4024            Expression::Struct(s) => {
4025                for (_, e) in &s.fields {
4026                    stack.push(e);
4027                }
4028            }
4029            Expression::Subscript(s) => {
4030                stack.push(&s.this);
4031                stack.push(&s.index);
4032            }
4033            Expression::Dot(d) => {
4034                stack.push(&d.this);
4035            }
4036            Expression::MethodCall(m) => {
4037                if !matches!(dialect, Some(DialectType::BigQuery))
4038                    || !is_bigquery_safe_namespace_receiver(&m.this)
4039                {
4040                    stack.push(&m.this);
4041                }
4042                for arg in &m.args {
4043                    stack.push(arg);
4044                }
4045            }
4046            Expression::ArraySlice(s) => {
4047                stack.push(&s.this);
4048                if let Some(ref start) = s.start {
4049                    stack.push(start);
4050                }
4051                if let Some(ref end) = s.end {
4052                    stack.push(end);
4053                }
4054            }
4055            Expression::Lambda(l) => {
4056                stack.push(&l.body);
4057            }
4058            Expression::NamedArgument(n) => {
4059                stack.push(&n.value);
4060            }
4061            Expression::Lateral(l) => {
4062                stack.push(&l.this);
4063                if let Some(ref view) = l.view {
4064                    stack.push(view);
4065                }
4066                if let Some(ref outer) = l.outer {
4067                    stack.push(outer);
4068                }
4069                if let Some(ref ordinality) = l.ordinality {
4070                    stack.push(ordinality);
4071                }
4072            }
4073            Expression::LateralView(lv) => {
4074                stack.push(&lv.this);
4075            }
4076            Expression::TryCatch(t) => {
4077                for stmt in &t.try_body {
4078                    stack.push(stmt);
4079                }
4080                if let Some(catch_body) = &t.catch_body {
4081                    for stmt in catch_body {
4082                        stack.push(stmt);
4083                    }
4084                }
4085            }
4086            Expression::BracedWildcard(e) | Expression::ReturnStmt(e) => {
4087                stack.push(e);
4088            }
4089
4090            // === Custom function structs ===
4091            Expression::Substring(f) => {
4092                stack.push(&f.this);
4093                stack.push(&f.start);
4094                if let Some(ref len) = f.length {
4095                    stack.push(len);
4096                }
4097            }
4098            Expression::Trim(f) => {
4099                stack.push(&f.this);
4100                if let Some(ref chars) = f.characters {
4101                    stack.push(chars);
4102                }
4103            }
4104            Expression::Replace(f) => {
4105                stack.push(&f.this);
4106                stack.push(&f.old);
4107                stack.push(&f.new);
4108            }
4109            Expression::IfFunc(f) => {
4110                stack.push(&f.condition);
4111                stack.push(&f.true_value);
4112                if let Some(ref fv) = f.false_value {
4113                    stack.push(fv);
4114                }
4115            }
4116            Expression::Nvl2(f) => {
4117                stack.push(&f.this);
4118                stack.push(&f.true_value);
4119                stack.push(&f.false_value);
4120            }
4121            Expression::ConcatWs(f) => {
4122                stack.push(&f.separator);
4123                for e in &f.expressions {
4124                    stack.push(e);
4125                }
4126            }
4127            Expression::Count(f) => {
4128                if let Some(ref this) = f.this {
4129                    stack.push(this);
4130                }
4131                if let Some(ref filter) = f.filter {
4132                    stack.push(filter);
4133                }
4134            }
4135            Expression::GroupConcat(f) => {
4136                stack.push(&f.this);
4137                if let Some(ref sep) = f.separator {
4138                    stack.push(sep);
4139                }
4140                if let Some(ref filter) = f.filter {
4141                    stack.push(filter);
4142                }
4143            }
4144            Expression::StringAgg(f) => {
4145                stack.push(&f.this);
4146                if let Some(ref sep) = f.separator {
4147                    stack.push(sep);
4148                }
4149                if let Some(ref filter) = f.filter {
4150                    stack.push(filter);
4151                }
4152                if let Some(ref limit) = f.limit {
4153                    stack.push(limit);
4154                }
4155            }
4156            Expression::ListAgg(f) => {
4157                stack.push(&f.this);
4158                if let Some(ref sep) = f.separator {
4159                    stack.push(sep);
4160                }
4161                if let Some(ref filter) = f.filter {
4162                    stack.push(filter);
4163                }
4164            }
4165            Expression::SumIf(f) => {
4166                stack.push(&f.this);
4167                stack.push(&f.condition);
4168                if let Some(ref filter) = f.filter {
4169                    stack.push(filter);
4170                }
4171            }
4172            Expression::DateAdd(f) | Expression::DateSub(f) => {
4173                stack.push(&f.this);
4174                stack.push(&f.interval);
4175            }
4176            Expression::DateDiff(f) => {
4177                stack.push(&f.this);
4178                stack.push(&f.expression);
4179            }
4180            Expression::DateTrunc(f) | Expression::TimestampTrunc(f) => {
4181                stack.push(&f.this);
4182            }
4183            Expression::Extract(f) => {
4184                stack.push(&f.this);
4185            }
4186            Expression::Round(f) => {
4187                stack.push(&f.this);
4188                if let Some(ref d) = f.decimals {
4189                    stack.push(d);
4190                }
4191            }
4192            Expression::Floor(f) => {
4193                stack.push(&f.this);
4194                if let Some(ref s) = f.scale {
4195                    stack.push(s);
4196                }
4197                if let Some(ref t) = f.to {
4198                    stack.push(t);
4199                }
4200            }
4201            Expression::Ceil(f) => {
4202                stack.push(&f.this);
4203                if let Some(ref d) = f.decimals {
4204                    stack.push(d);
4205                }
4206                if let Some(ref t) = f.to {
4207                    stack.push(t);
4208                }
4209            }
4210            Expression::Log(f) => {
4211                stack.push(&f.this);
4212                if let Some(ref b) = f.base {
4213                    stack.push(b);
4214                }
4215            }
4216            Expression::AtTimeZone(f) => {
4217                stack.push(&f.this);
4218                stack.push(&f.zone);
4219            }
4220            Expression::Lead(f) | Expression::Lag(f) => {
4221                stack.push(&f.this);
4222                if let Some(ref off) = f.offset {
4223                    stack.push(off);
4224                }
4225                if let Some(ref def) = f.default {
4226                    stack.push(def);
4227                }
4228            }
4229            Expression::FirstValue(f) | Expression::LastValue(f) => {
4230                stack.push(&f.this);
4231            }
4232            Expression::NthValue(f) => {
4233                stack.push(&f.this);
4234                stack.push(&f.offset);
4235            }
4236            Expression::Position(f) => {
4237                stack.push(&f.substring);
4238                stack.push(&f.string);
4239                if let Some(ref start) = f.start {
4240                    stack.push(start);
4241                }
4242            }
4243            Expression::Decode(f) => {
4244                stack.push(&f.this);
4245                for (search, result) in &f.search_results {
4246                    stack.push(search);
4247                    stack.push(result);
4248                }
4249                if let Some(ref def) = f.default {
4250                    stack.push(def);
4251                }
4252            }
4253            Expression::CharFunc(f) => {
4254                for arg in &f.args {
4255                    stack.push(arg);
4256                }
4257            }
4258            Expression::ArraySort(f) => {
4259                stack.push(&f.this);
4260                if let Some(ref cmp) = f.comparator {
4261                    stack.push(cmp);
4262                }
4263            }
4264            Expression::ArrayJoin(f) | Expression::ArrayToString(f) => {
4265                stack.push(&f.this);
4266                stack.push(&f.separator);
4267                if let Some(ref nr) = f.null_replacement {
4268                    stack.push(nr);
4269                }
4270            }
4271            Expression::ArrayFilter(f) => {
4272                stack.push(&f.this);
4273                stack.push(&f.filter);
4274            }
4275            Expression::ArrayTransform(f) => {
4276                stack.push(&f.this);
4277                stack.push(&f.transform);
4278            }
4279            Expression::Sequence(f)
4280            | Expression::Generate(f)
4281            | Expression::ExplodingGenerateSeries(f) => {
4282                stack.push(&f.start);
4283                stack.push(&f.stop);
4284                if let Some(ref step) = f.step {
4285                    stack.push(step);
4286                }
4287            }
4288            Expression::JsonExtract(f)
4289            | Expression::JsonExtractScalar(f)
4290            | Expression::JsonQuery(f)
4291            | Expression::JsonValue(f) => {
4292                stack.push(&f.this);
4293                stack.push(&f.path);
4294            }
4295            Expression::JsonExtractPath(f) | Expression::JsonRemove(f) => {
4296                stack.push(&f.this);
4297                for p in &f.paths {
4298                    stack.push(p);
4299                }
4300            }
4301            Expression::JsonObject(f) => {
4302                for (k, v) in &f.pairs {
4303                    stack.push(k);
4304                    stack.push(v);
4305                }
4306            }
4307            Expression::JsonSet(f) | Expression::JsonInsert(f) => {
4308                stack.push(&f.this);
4309                for (path, val) in &f.path_values {
4310                    stack.push(path);
4311                    stack.push(val);
4312                }
4313            }
4314            Expression::Overlay(f) => {
4315                stack.push(&f.this);
4316                stack.push(&f.replacement);
4317                stack.push(&f.from);
4318                if let Some(ref len) = f.length {
4319                    stack.push(len);
4320                }
4321            }
4322            Expression::Convert(f) => {
4323                stack.push(&f.this);
4324                if let Some(ref style) = f.style {
4325                    stack.push(style);
4326                }
4327            }
4328            Expression::ApproxPercentile(f) => {
4329                stack.push(&f.this);
4330                stack.push(&f.percentile);
4331                if let Some(ref acc) = f.accuracy {
4332                    stack.push(acc);
4333                }
4334                if let Some(ref filter) = f.filter {
4335                    stack.push(filter);
4336                }
4337            }
4338            Expression::Percentile(f)
4339            | Expression::PercentileCont(f)
4340            | Expression::PercentileDisc(f) => {
4341                stack.push(&f.this);
4342                stack.push(&f.percentile);
4343                if let Some(ref filter) = f.filter {
4344                    stack.push(filter);
4345                }
4346            }
4347            Expression::WithinGroup(f) => {
4348                stack.push(&f.this);
4349                for e in &f.order_by {
4350                    stack.push(&e.this);
4351                }
4352            }
4353            Expression::Left(f) | Expression::Right(f) => {
4354                stack.push(&f.this);
4355                stack.push(&f.length);
4356            }
4357            Expression::Repeat(f) => {
4358                stack.push(&f.this);
4359                stack.push(&f.times);
4360            }
4361            Expression::Lpad(f) | Expression::Rpad(f) => {
4362                stack.push(&f.this);
4363                stack.push(&f.length);
4364                if let Some(ref fill) = f.fill {
4365                    stack.push(fill);
4366                }
4367            }
4368            Expression::Split(f) => {
4369                stack.push(&f.this);
4370                stack.push(&f.delimiter);
4371            }
4372            Expression::RegexpLike(f) => {
4373                stack.push(&f.this);
4374                stack.push(&f.pattern);
4375                if let Some(ref flags) = f.flags {
4376                    stack.push(flags);
4377                }
4378            }
4379            Expression::RegexpReplace(f) => {
4380                stack.push(&f.this);
4381                stack.push(&f.pattern);
4382                stack.push(&f.replacement);
4383                if let Some(ref flags) = f.flags {
4384                    stack.push(flags);
4385                }
4386            }
4387            Expression::RegexpExtract(f) => {
4388                stack.push(&f.this);
4389                stack.push(&f.pattern);
4390                if let Some(ref group) = f.group {
4391                    stack.push(group);
4392                }
4393            }
4394            Expression::ToDate(f) => {
4395                stack.push(&f.this);
4396                if let Some(ref fmt) = f.format {
4397                    stack.push(fmt);
4398                }
4399            }
4400            Expression::ToTimestamp(f) => {
4401                stack.push(&f.this);
4402                if let Some(ref fmt) = f.format {
4403                    stack.push(fmt);
4404                }
4405            }
4406            Expression::DateFormat(f) | Expression::FormatDate(f) => {
4407                stack.push(&f.this);
4408                stack.push(&f.format);
4409            }
4410            Expression::LastDay(f) => {
4411                stack.push(&f.this);
4412            }
4413            Expression::FromUnixtime(f) => {
4414                stack.push(&f.this);
4415                if let Some(ref fmt) = f.format {
4416                    stack.push(fmt);
4417                }
4418            }
4419            Expression::UnixTimestamp(f) => {
4420                if let Some(ref this) = f.this {
4421                    stack.push(this);
4422                }
4423                if let Some(ref fmt) = f.format {
4424                    stack.push(fmt);
4425                }
4426            }
4427            Expression::MakeDate(f) => {
4428                stack.push(&f.year);
4429                stack.push(&f.month);
4430                stack.push(&f.day);
4431            }
4432            Expression::MakeTimestamp(f) => {
4433                stack.push(&f.year);
4434                stack.push(&f.month);
4435                stack.push(&f.day);
4436                stack.push(&f.hour);
4437                stack.push(&f.minute);
4438                stack.push(&f.second);
4439                if let Some(ref tz) = f.timezone {
4440                    stack.push(tz);
4441                }
4442            }
4443            Expression::TruncFunc(f) => {
4444                stack.push(&f.this);
4445                if let Some(ref d) = f.decimals {
4446                    stack.push(d);
4447                }
4448            }
4449            Expression::ArrayFunc(f) => {
4450                for e in &f.expressions {
4451                    stack.push(e);
4452                }
4453            }
4454            Expression::Unnest(f) => {
4455                stack.push(&f.this);
4456                for e in &f.expressions {
4457                    stack.push(e);
4458                }
4459            }
4460            Expression::StructFunc(f) => {
4461                for (_, e) in &f.fields {
4462                    stack.push(e);
4463                }
4464            }
4465            Expression::StructExtract(f) => {
4466                stack.push(&f.this);
4467            }
4468            Expression::NamedStruct(f) => {
4469                for (k, v) in &f.pairs {
4470                    stack.push(k);
4471                    stack.push(v);
4472                }
4473            }
4474            Expression::MapFunc(f) => {
4475                for k in &f.keys {
4476                    stack.push(k);
4477                }
4478                for v in &f.values {
4479                    stack.push(v);
4480                }
4481            }
4482            Expression::TransformKeys(f) | Expression::TransformValues(f) => {
4483                stack.push(&f.this);
4484                stack.push(&f.transform);
4485            }
4486            Expression::JsonArrayAgg(f) => {
4487                stack.push(&f.this);
4488                if let Some(ref filter) = f.filter {
4489                    stack.push(filter);
4490                }
4491            }
4492            Expression::JsonObjectAgg(f) => {
4493                stack.push(&f.key);
4494                stack.push(&f.value);
4495                if let Some(ref filter) = f.filter {
4496                    stack.push(filter);
4497                }
4498            }
4499            Expression::NTile(f) => {
4500                if let Some(ref n) = f.num_buckets {
4501                    stack.push(n);
4502                }
4503            }
4504            Expression::Rand(f) => {
4505                if let Some(ref s) = f.seed {
4506                    stack.push(s);
4507                }
4508                if let Some(ref lo) = f.lower {
4509                    stack.push(lo);
4510                }
4511                if let Some(ref hi) = f.upper {
4512                    stack.push(hi);
4513                }
4514            }
4515            Expression::Any(q) | Expression::All(q) => {
4516                stack.push(&q.this);
4517                stack.push(&q.subquery);
4518            }
4519            Expression::Overlaps(o) => {
4520                if let Some(ref this) = o.this {
4521                    stack.push(this);
4522                }
4523                if let Some(ref expr) = o.expression {
4524                    stack.push(expr);
4525                }
4526                if let Some(ref ls) = o.left_start {
4527                    stack.push(ls);
4528                }
4529                if let Some(ref le) = o.left_end {
4530                    stack.push(le);
4531                }
4532                if let Some(ref rs) = o.right_start {
4533                    stack.push(rs);
4534                }
4535                if let Some(ref re) = o.right_end {
4536                    stack.push(re);
4537                }
4538            }
4539            Expression::Interval(i) => {
4540                if let Some(ref this) = i.this {
4541                    stack.push(this);
4542                }
4543            }
4544            Expression::TimeStrToTime(f) => {
4545                stack.push(&f.this);
4546                if let Some(ref zone) = f.zone {
4547                    stack.push(zone);
4548                }
4549            }
4550            Expression::JSONBExtractScalar(f) => {
4551                stack.push(&f.this);
4552                stack.push(&f.expression);
4553                if let Some(ref jt) = f.json_type {
4554                    stack.push(jt);
4555                }
4556            }
4557            Expression::JSONExtract(f) => {
4558                stack.push(&f.this);
4559                stack.push(&f.expression);
4560                for e in &f.expressions {
4561                    stack.push(e);
4562                }
4563                if let Some(ref option) = f.option {
4564                    stack.push(option);
4565                }
4566                if let Some(ref on_condition) = f.on_condition {
4567                    stack.push(on_condition);
4568                }
4569            }
4570
4571            // === True leaves and non-expression-bearing nodes ===
4572            // Literals, Identifier, Star, DataType, Placeholder, Boolean, Null,
4573            // CurrentDate/Time/Timestamp, RowNumber, Rank, DenseRank, PercentRank,
4574            // CumeDist, Random, Pi, SessionUser, DDL statements, clauses, etc.
4575            _ => {}
4576        }
4577    }
4578}
4579
4580// ---------------------------------------------------------------------------
4581// Tests
4582// ---------------------------------------------------------------------------
4583
4584#[cfg(test)]
4585mod tests {
4586    use super::*;
4587    use crate::dialects::{Dialect, DialectType};
4588    use crate::expressions::DataType;
4589    use crate::optimizer::annotate_types::annotate_types;
4590    use crate::parse_one;
4591    use crate::schema::{MappingSchema, Schema};
4592
4593    fn parse(sql: &str) -> Expression {
4594        let dialect = Dialect::get(DialectType::Generic);
4595        let ast = dialect.parse(sql).unwrap();
4596        ast.into_iter().next().unwrap()
4597    }
4598
4599    fn parse_dialect(sql: &str, dialect_type: DialectType) -> Expression {
4600        let dialect = Dialect::get(dialect_type);
4601        let ast = dialect.parse(sql).unwrap();
4602        ast.into_iter().next().unwrap()
4603    }
4604
4605    fn lineage_names(node: &LineageNode) -> Vec<String> {
4606        node.walk().map(|n| n.name.clone()).collect()
4607    }
4608
4609    fn assert_lineage_contains(node: &LineageNode, expected: &str) {
4610        let names = lineage_names(node);
4611        assert!(
4612            names.iter().any(|name| name == expected),
4613            "expected {expected} in lineage, got {names:?}"
4614        );
4615    }
4616
4617    const ISSUE_368_SQL: &str = "with
4618base as (
4619  select 1 as col_a
4620),
4621literal_branch as (
4622  select 2 as col_a
4623),
4624unioned as (
4625  select * from base
4626  union all
4627  select * from literal_branch
4628)
4629select col_a from unioned";
4630
4631    #[test]
4632    fn test_simple_lineage() {
4633        let expr = parse("SELECT a FROM t");
4634        let node = lineage("a", &expr, None, false).unwrap();
4635
4636        assert_eq!(node.name, "a");
4637        assert!(!node.downstream.is_empty(), "Should have downstream nodes");
4638        // Should trace to t.a
4639        let names = node.downstream_names();
4640        assert!(
4641            names.iter().any(|n| n == "t.a"),
4642            "Expected t.a in downstream, got: {:?}",
4643            names
4644        );
4645    }
4646
4647    #[test]
4648    fn test_lineage_walk() {
4649        let root = LineageNode {
4650            name: "col_a".to_string(),
4651            expression: Expression::Null(crate::expressions::Null),
4652            source: Expression::Null(crate::expressions::Null),
4653            downstream: vec![LineageNode::new(
4654                "t.a",
4655                Expression::Null(crate::expressions::Null),
4656                Expression::Null(crate::expressions::Null),
4657            )],
4658            source_name: String::new(),
4659            source_kind: SourceKind::Unknown,
4660            source_alias: None,
4661            set_branch: None,
4662            reference_node_name: String::new(),
4663        };
4664
4665        let names: Vec<_> = root.walk().map(|n| n.name.clone()).collect();
4666        assert_eq!(names.len(), 2);
4667        assert_eq!(names[0], "col_a");
4668        assert_eq!(names[1], "t.a");
4669    }
4670
4671    #[test]
4672    fn test_aliased_column() {
4673        let expr = parse("SELECT a + 1 AS b FROM t");
4674        let node = lineage("b", &expr, None, false).unwrap();
4675
4676        assert_eq!(node.name, "b");
4677        // Should trace through the expression to t.a
4678        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
4679        assert!(
4680            all_names.iter().any(|n| n.contains("a")),
4681            "Expected to trace to column a, got: {:?}",
4682            all_names
4683        );
4684    }
4685
4686    #[test]
4687    fn test_qualified_column() {
4688        let expr = parse("SELECT t.a FROM t");
4689        let node = lineage("a", &expr, None, false).unwrap();
4690
4691        assert_eq!(node.name, "a");
4692        let names = node.downstream_names();
4693        assert!(
4694            names.iter().any(|n| n == "t.a"),
4695            "Expected t.a, got: {:?}",
4696            names
4697        );
4698    }
4699
4700    #[test]
4701    fn test_unqualified_column() {
4702        let expr = parse("SELECT a FROM t");
4703        let node = lineage("a", &expr, None, false).unwrap();
4704
4705        // Unqualified but single source → resolved to t.a
4706        let names = node.downstream_names();
4707        assert!(
4708            names.iter().any(|n| n == "t.a"),
4709            "Expected t.a, got: {:?}",
4710            names
4711        );
4712    }
4713
4714    #[test]
4715    fn test_lineage_with_schema_qualifies_root_expression_issue_40() {
4716        let query = "SELECT name FROM users";
4717        let dialect = Dialect::get(DialectType::BigQuery);
4718        let expr = dialect
4719            .parse(query)
4720            .unwrap()
4721            .into_iter()
4722            .next()
4723            .expect("expected one expression");
4724
4725        let mut schema = MappingSchema::with_dialect(DialectType::BigQuery);
4726        schema
4727            .add_table("users", &[("name".into(), DataType::Text)], None)
4728            .expect("schema setup");
4729
4730        let node_without_schema = lineage("name", &expr, Some(DialectType::BigQuery), false)
4731            .expect("lineage without schema");
4732        let mut expr_without = node_without_schema.expression.clone();
4733        annotate_types(
4734            &mut expr_without,
4735            Some(&schema),
4736            Some(DialectType::BigQuery),
4737        );
4738        assert_eq!(
4739            expr_without.inferred_type(),
4740            None,
4741            "Expected unresolved root type without schema-aware lineage qualification"
4742        );
4743
4744        let node_with_schema = lineage_with_schema(
4745            "name",
4746            &expr,
4747            Some(&schema),
4748            Some(DialectType::BigQuery),
4749            false,
4750        )
4751        .expect("lineage with schema");
4752        let mut expr_with = node_with_schema.expression.clone();
4753        annotate_types(&mut expr_with, Some(&schema), Some(DialectType::BigQuery));
4754
4755        assert_eq!(expr_with.inferred_type(), Some(&DataType::Text));
4756    }
4757
4758    #[test]
4759    fn test_lineage_with_schema_tolerates_partial_schema_for_known_column() {
4760        let expr = parse_dialect("SELECT order_id, amount FROM t", DialectType::DuckDB);
4761        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
4762        schema
4763            .add_table(
4764                "t",
4765                &[("amount".into(), DataType::BigInt { length: None })],
4766                None,
4767            )
4768            .expect("schema setup");
4769
4770        let node = lineage_with_schema(
4771            "amount",
4772            &expr,
4773            Some(&schema),
4774            Some(DialectType::DuckDB),
4775            false,
4776        )
4777        .expect("lineage_with_schema should tolerate unrelated unknown columns");
4778
4779        assert_lineage_contains(&node, "t.amount");
4780    }
4781
4782    #[test]
4783    fn test_lineage_with_schema_tolerates_partial_schema_for_unknown_column() {
4784        let expr = parse_dialect("SELECT order_id, amount FROM t", DialectType::DuckDB);
4785        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
4786        schema
4787            .add_table(
4788                "t",
4789                &[("amount".into(), DataType::BigInt { length: None })],
4790                None,
4791            )
4792            .expect("schema setup");
4793
4794        let node = lineage_with_schema(
4795            "order_id",
4796            &expr,
4797            Some(&schema),
4798            Some(DialectType::DuckDB),
4799            false,
4800        )
4801        .expect("lineage_with_schema should keep unknown selected columns");
4802
4803        assert_lineage_contains(&node, "t.order_id");
4804    }
4805
4806    #[test]
4807    fn test_lineage_with_schema_tolerates_partial_schema_for_join_conditions() {
4808        let expr = parse_dialect(
4809            "SELECT a.order_id, b.amount FROM t a JOIN u b ON a.id = b.id",
4810            DialectType::DuckDB,
4811        );
4812        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
4813        schema
4814            .add_table(
4815                "t",
4816                &[("order_id".into(), DataType::BigInt { length: None })],
4817                None,
4818            )
4819            .expect("schema setup");
4820        schema
4821            .add_table(
4822                "u",
4823                &[("amount".into(), DataType::BigInt { length: None })],
4824                None,
4825            )
4826            .expect("schema setup");
4827
4828        let node = lineage_with_schema(
4829            "amount",
4830            &expr,
4831            Some(&schema),
4832            Some(DialectType::DuckDB),
4833            false,
4834        )
4835        .expect("lineage_with_schema should tolerate unknown join keys");
4836
4837        assert_lineage_contains(&node, "b.amount");
4838    }
4839
4840    #[test]
4841    fn test_lineage_with_schema_correlated_scalar_subquery() {
4842        let query = "SELECT id, (SELECT AVG(val) FROM t2 WHERE t2.id = t1.id) AS avg_val FROM t1";
4843        let dialect = Dialect::get(DialectType::BigQuery);
4844        let expr = dialect
4845            .parse(query)
4846            .unwrap()
4847            .into_iter()
4848            .next()
4849            .expect("expected one expression");
4850
4851        let mut schema = MappingSchema::with_dialect(DialectType::BigQuery);
4852        schema
4853            .add_table(
4854                "t1",
4855                &[("id".into(), DataType::BigInt { length: None })],
4856                None,
4857            )
4858            .expect("schema setup");
4859        schema
4860            .add_table(
4861                "t2",
4862                &[
4863                    ("id".into(), DataType::BigInt { length: None }),
4864                    ("val".into(), DataType::BigInt { length: None }),
4865                ],
4866                None,
4867            )
4868            .expect("schema setup");
4869
4870        let node = lineage_with_schema(
4871            "id",
4872            &expr,
4873            Some(&schema),
4874            Some(DialectType::BigQuery),
4875            false,
4876        )
4877        .expect("lineage_with_schema should handle correlated scalar subqueries");
4878
4879        assert_eq!(node.name, "id");
4880    }
4881
4882    #[test]
4883    fn test_lineage_with_schema_join_using() {
4884        let query = "SELECT a FROM t1 JOIN t2 USING(a)";
4885        let dialect = Dialect::get(DialectType::BigQuery);
4886        let expr = dialect
4887            .parse(query)
4888            .unwrap()
4889            .into_iter()
4890            .next()
4891            .expect("expected one expression");
4892
4893        let mut schema = MappingSchema::with_dialect(DialectType::BigQuery);
4894        schema
4895            .add_table(
4896                "t1",
4897                &[("a".into(), DataType::BigInt { length: None })],
4898                None,
4899            )
4900            .expect("schema setup");
4901        schema
4902            .add_table(
4903                "t2",
4904                &[("a".into(), DataType::BigInt { length: None })],
4905                None,
4906            )
4907            .expect("schema setup");
4908
4909        let node = lineage_with_schema(
4910            "a",
4911            &expr,
4912            Some(&schema),
4913            Some(DialectType::BigQuery),
4914            false,
4915        )
4916        .expect("lineage_with_schema should handle JOIN USING");
4917
4918        assert_eq!(node.name, "a");
4919    }
4920
4921    #[test]
4922    fn test_lineage_with_schema_natural_join_merged_column() {
4923        let query = "SELECT shared_key AS output_key FROM source_table \
4924                     NATURAL JOIN (SELECT shared_key FROM source_table) derived";
4925        let dialect = Dialect::get(DialectType::DuckDB);
4926        let expr = dialect
4927            .parse(query)
4928            .unwrap()
4929            .into_iter()
4930            .next()
4931            .expect("expected one expression");
4932
4933        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
4934        schema
4935            .add_table(
4936                "source_table",
4937                &[(
4938                    "shared_key".into(),
4939                    DataType::VarChar {
4940                        length: None,
4941                        parenthesized_length: false,
4942                    },
4943                )],
4944                None,
4945            )
4946            .expect("schema setup");
4947
4948        let node = lineage_with_schema(
4949            "output_key",
4950            &expr,
4951            Some(&schema),
4952            Some(DialectType::DuckDB),
4953            false,
4954        )
4955        .expect("lineage_with_schema should resolve a NATURAL JOIN merged column");
4956
4957        let terminal_nodes: Vec<_> = node
4958            .walk()
4959            .filter(|candidate| candidate.downstream.is_empty())
4960            .collect();
4961        assert_eq!(
4962            terminal_nodes.len(),
4963            2,
4964            "both NATURAL JOIN inputs should remain visible: {node:#?}"
4965        );
4966        assert!(terminal_nodes.iter().all(|candidate| {
4967            candidate.source_kind == SourceKind::Table
4968                && candidate.source_name == "source_table"
4969                && candidate.name.ends_with(".shared_key")
4970        }));
4971    }
4972
4973    #[test]
4974    fn test_lineage_with_schema_propagates_type_through_anonymous_derived_table() {
4975        let expr = parse_dialect(
4976            "SELECT source_value FROM (SELECT source_value FROM source_table)",
4977            DialectType::DuckDB,
4978        );
4979        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
4980        let expected = DataType::VarChar {
4981            length: None,
4982            parenthesized_length: false,
4983        };
4984        schema
4985            .add_table(
4986                "source_table",
4987                &[("source_value".into(), expected.clone())],
4988                None,
4989            )
4990            .expect("schema setup");
4991
4992        let node = lineage_with_schema(
4993            "source_value",
4994            &expr,
4995            Some(&schema),
4996            Some(DialectType::DuckDB),
4997            false,
4998        )
4999        .expect("lineage_with_schema should resolve an anonymous derived table");
5000
5001        assert_eq!(node.expression.inferred_type(), Some(&expected));
5002        let terminal_nodes: Vec<_> = node
5003            .walk()
5004            .filter(|candidate| candidate.downstream.is_empty())
5005            .collect();
5006        assert_eq!(terminal_nodes.len(), 1, "{node:#?}");
5007        assert_eq!(terminal_nodes[0].source_kind, SourceKind::Table);
5008        assert_eq!(terminal_nodes[0].source_name, "source_table");
5009        assert_eq!(terminal_nodes[0].name, "source_table.source_value");
5010    }
5011
5012    #[test]
5013    fn test_issue_412_duckdb_date_trunc_preserves_timestamp_lineage_type() {
5014        let expr = parse_dialect(
5015            "SELECT DATE_TRUNC('month', event_timestamp) AS truncated_timestamp \
5016             FROM source_table",
5017            DialectType::DuckDB,
5018        );
5019        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
5020        let timestamp = DataType::Timestamp {
5021            precision: None,
5022            timezone: false,
5023        };
5024        schema
5025            .add_table(
5026                "source_table",
5027                &[("event_timestamp".into(), timestamp.clone())],
5028                None,
5029            )
5030            .expect("schema setup");
5031
5032        let node =
5033            lineage_at_with_schema(0, &expr, Some(&schema), Some(DialectType::DuckDB), false)
5034                .expect("schema-aware lineage");
5035
5036        assert_eq!(node.expression.inferred_type(), Some(&timestamp));
5037        let Expression::Alias(alias) = &node.expression else {
5038            panic!("expected aliased DATE_TRUNC projection");
5039        };
5040        assert_eq!(alias.this.inferred_type(), Some(&timestamp));
5041        let Expression::Function(function) = &alias.this else {
5042            panic!("expected generic DATE_TRUNC function");
5043        };
5044        assert_eq!(function.args[1].inferred_type(), Some(&timestamp));
5045    }
5046
5047    #[test]
5048    fn test_lineage_with_schema_qualified_table_name() {
5049        let query = "SELECT a FROM raw.t1";
5050        let dialect = Dialect::get(DialectType::BigQuery);
5051        let expr = dialect
5052            .parse(query)
5053            .unwrap()
5054            .into_iter()
5055            .next()
5056            .expect("expected one expression");
5057
5058        let mut schema = MappingSchema::with_dialect(DialectType::BigQuery);
5059        schema
5060            .add_table(
5061                "raw.t1",
5062                &[("a".into(), DataType::BigInt { length: None })],
5063                None,
5064            )
5065            .expect("schema setup");
5066
5067        let node = lineage_with_schema(
5068            "a",
5069            &expr,
5070            Some(&schema),
5071            Some(DialectType::BigQuery),
5072            false,
5073        )
5074        .expect("lineage_with_schema should handle dotted schema.table names");
5075
5076        assert_eq!(node.name, "a");
5077    }
5078
5079    #[test]
5080    fn test_lineage_with_schema_none_matches_lineage() {
5081        let expr = parse("SELECT a FROM t");
5082        let baseline = lineage("a", &expr, None, false).expect("lineage baseline");
5083        let with_none =
5084            lineage_with_schema("a", &expr, None, None, false).expect("lineage_with_schema");
5085
5086        assert_eq!(with_none.name, baseline.name);
5087        assert_eq!(with_none.downstream_names(), baseline.downstream_names());
5088    }
5089
5090    #[test]
5091    fn test_lineage_with_schema_bigquery_mixed_case_column_names_issue_60() {
5092        let dialect = Dialect::get(DialectType::BigQuery);
5093        let expr = dialect
5094            .parse("SELECT Name AS name FROM teams")
5095            .unwrap()
5096            .into_iter()
5097            .next()
5098            .expect("expected one expression");
5099
5100        let mut schema = MappingSchema::with_dialect(DialectType::BigQuery);
5101        schema
5102            .add_table(
5103                "teams",
5104                &[("Name".into(), DataType::String { length: None })],
5105                None,
5106            )
5107            .expect("schema setup");
5108
5109        let node = lineage_with_schema(
5110            "name",
5111            &expr,
5112            Some(&schema),
5113            Some(DialectType::BigQuery),
5114            false,
5115        )
5116        .expect("lineage_with_schema should resolve mixed-case BigQuery columns");
5117
5118        let names = node.downstream_names();
5119        assert!(
5120            names.iter().any(|n| n == "teams.Name"),
5121            "Expected teams.Name in downstream, got: {:?}",
5122            names
5123        );
5124    }
5125
5126    #[test]
5127    fn test_lineage_bigquery_mixed_case_alias_lookup() {
5128        let dialect = Dialect::get(DialectType::BigQuery);
5129        let expr = dialect
5130            .parse("SELECT Name AS Name FROM teams")
5131            .unwrap()
5132            .into_iter()
5133            .next()
5134            .expect("expected one expression");
5135
5136        let node = lineage("name", &expr, Some(DialectType::BigQuery), false)
5137            .expect("lineage should resolve mixed-case aliases in BigQuery");
5138
5139        assert_eq!(node.name, "name");
5140    }
5141
5142    #[test]
5143    fn test_lineage_bigquery_unnest_alias_source_issue_209() {
5144        let expr = parse_one(
5145            r#"
5146SELECT date_val AS week_start
5147FROM UNNEST(GENERATE_DATE_ARRAY('2024-01-01', '2024-12-31', INTERVAL 1 WEEK)) AS date_val
5148"#,
5149            DialectType::BigQuery,
5150        )
5151        .expect("parse");
5152
5153        let node = lineage("week_start", &expr, Some(DialectType::BigQuery), false)
5154            .expect("lineage should resolve UNNEST alias as a source");
5155        let child = node
5156            .downstream
5157            .first()
5158            .expect("week_start should have downstream lineage");
5159
5160        assert_eq!(child.name, "_0.date_val");
5161        assert_eq!(child.source_name, "_0");
5162        assert_eq!(child.source_kind, SourceKind::Virtual);
5163        assert_eq!(child.source_alias.as_deref(), Some("date_val"));
5164
5165        let Expression::Column(column) = &child.expression else {
5166            panic!(
5167                "expected downstream column expression, got {:?}",
5168                child.expression
5169            );
5170        };
5171        assert_eq!(column.name.name, "date_val");
5172        assert_eq!(
5173            column.table.as_ref().map(|table| table.name.as_str()),
5174            Some("_0")
5175        );
5176        assert!(
5177            matches!(&child.source, Expression::Alias(alias) if matches!(&alias.this, Expression::Unnest(_)) && alias.alias.name == "date_val"),
5178            "expected UNNEST source expression, got {:?}",
5179            child.source
5180        );
5181    }
5182
5183    #[test]
5184    fn test_lineage_real_table_named_like_unnest_alias_is_not_virtual() {
5185        let expr =
5186            parse_one("SELECT date_val.id FROM date_val", DialectType::BigQuery).expect("parse");
5187
5188        let node = lineage("id", &expr, Some(DialectType::BigQuery), false).expect("lineage");
5189        let child = node.downstream.first().expect("id should have lineage");
5190
5191        assert_eq!(child.name, "date_val.id");
5192        assert_eq!(child.source_name, "date_val");
5193        assert_eq!(child.source_kind, SourceKind::Table);
5194        assert_eq!(child.source_alias, None);
5195    }
5196
5197    #[test]
5198    fn test_lineage_multiple_bigquery_unnest_sources_get_stable_virtual_names() {
5199        let expr = parse_one(
5200            r#"
5201SELECT a.a AS first_value, b.b AS second_value
5202FROM UNNEST(GENERATE_ARRAY(1, 2)) AS a
5203JOIN UNNEST(GENERATE_ARRAY(3, 4)) AS b ON TRUE
5204"#,
5205            DialectType::BigQuery,
5206        )
5207        .expect("parse");
5208
5209        let first =
5210            lineage("first_value", &expr, Some(DialectType::BigQuery), false).expect("lineage");
5211        let second =
5212            lineage("second_value", &expr, Some(DialectType::BigQuery), false).expect("lineage");
5213
5214        let first_child = first.downstream.first().expect("first source");
5215        let second_child = second.downstream.first().expect("second source");
5216
5217        assert_eq!(first_child.name, "_0.a");
5218        assert_eq!(first_child.source_name, "_0");
5219        assert_eq!(first_child.source_alias.as_deref(), Some("a"));
5220        assert_eq!(first_child.source_kind, SourceKind::Virtual);
5221
5222        assert_eq!(second_child.name, "_1.b");
5223        assert_eq!(second_child.source_name, "_1");
5224        assert_eq!(second_child.source_alias.as_deref(), Some("b"));
5225        assert_eq!(second_child.source_kind, SourceKind::Virtual);
5226    }
5227
5228    #[test]
5229    fn test_lineage_table_backed_unnest_points_to_real_source_column() {
5230        let expr = parse_one(
5231            r#"
5232SELECT item.item AS item
5233FROM t JOIN UNNEST(t.items) AS item ON TRUE
5234"#,
5235            DialectType::BigQuery,
5236        )
5237        .expect("parse");
5238
5239        let node = lineage("item", &expr, Some(DialectType::BigQuery), false).expect("lineage");
5240        let virtual_child = node.downstream.first().expect("virtual item source");
5241        assert_eq!(virtual_child.name, "_0.item");
5242        assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5243
5244        let real_child = virtual_child
5245            .downstream
5246            .first()
5247            .expect("UNNEST(t.items) should depend on t.items");
5248        assert_eq!(real_child.name, "t.items");
5249        assert_eq!(real_child.source_name, "t");
5250        assert_eq!(real_child.source_kind, SourceKind::Table);
5251    }
5252
5253    #[test]
5254    fn test_lineage_table_backed_unnest_unqualified_column_resolves_to_virtual_source() {
5255        let expr = parse_one(
5256            r#"
5257SELECT item AS item
5258FROM t JOIN UNNEST(t.items) AS item ON TRUE
5259"#,
5260            DialectType::BigQuery,
5261        )
5262        .expect("parse");
5263
5264        let node = lineage("item", &expr, Some(DialectType::BigQuery), false).expect("lineage");
5265        let virtual_child = node.downstream.first().expect("virtual item source");
5266        assert_eq!(virtual_child.name, "_0.item");
5267        assert_eq!(virtual_child.source_name, "_0");
5268        assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5269        assert_eq!(virtual_child.source_alias.as_deref(), Some("item"));
5270
5271        let real_child = virtual_child
5272            .downstream
5273            .first()
5274            .expect("UNNEST(t.items) should depend on t.items");
5275        assert_eq!(real_child.name, "t.items");
5276        assert_eq!(real_child.source_name, "t");
5277        assert_eq!(real_child.source_kind, SourceKind::Table);
5278    }
5279
5280    #[test]
5281    fn test_lineage_unnest_alias_columns_resolve_to_virtual_sources_across_dialects() {
5282        let cases = [
5283            (
5284                DialectType::PostgreSQL,
5285                "SELECT x AS out FROM t CROSS JOIN LATERAL UNNEST(items) AS u(x)",
5286            ),
5287            (
5288                DialectType::Presto,
5289                "SELECT x AS out FROM t CROSS JOIN UNNEST(items) AS u(x)",
5290            ),
5291            (
5292                DialectType::Trino,
5293                "SELECT x AS out FROM t CROSS JOIN UNNEST(items) AS u(x)",
5294            ),
5295        ];
5296
5297        for (dialect, sql) in cases {
5298            let expr = parse_one(sql, dialect).unwrap_or_else(|e| panic!("parse {dialect:?}: {e}"));
5299            let node = lineage("out", &expr, Some(dialect), false)
5300                .unwrap_or_else(|e| panic!("lineage {dialect:?}: {e}"));
5301            let virtual_child = node
5302                .downstream
5303                .first()
5304                .unwrap_or_else(|| panic!("expected virtual child for {dialect:?}"));
5305
5306            assert_eq!(
5307                virtual_child.name, "_0.x",
5308                "unexpected virtual child for {dialect:?}"
5309            );
5310            assert_eq!(virtual_child.source_name, "_0");
5311            assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5312            assert_eq!(virtual_child.source_alias.as_deref(), Some("u"));
5313
5314            let real_child = virtual_child
5315                .downstream
5316                .first()
5317                .unwrap_or_else(|| panic!("expected table dependency for {dialect:?}"));
5318            assert_eq!(real_child.name, "t.items");
5319            assert_eq!(real_child.source_kind, SourceKind::Table);
5320        }
5321    }
5322
5323    #[test]
5324    fn test_lineage_with_schema_propagates_unnest_element_type() {
5325        let expr = parse_dialect(
5326            "SELECT u.tag FROM events AS e, UNNEST(e.tags) AS u(tag)",
5327            DialectType::DuckDB,
5328        );
5329        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
5330        schema
5331            .add_table(
5332                "events",
5333                &[(
5334                    "tags".into(),
5335                    DataType::Array {
5336                        element_type: Box::new(DataType::VarChar {
5337                            length: None,
5338                            parenthesized_length: false,
5339                        }),
5340                        dimension: None,
5341                    },
5342                )],
5343                None,
5344            )
5345            .expect("schema setup");
5346
5347        let node = lineage_with_schema(
5348            "tag",
5349            &expr,
5350            Some(&schema),
5351            Some(DialectType::DuckDB),
5352            false,
5353        )
5354        .expect("lineage_with_schema");
5355        let expected = DataType::VarChar {
5356            length: None,
5357            parenthesized_length: false,
5358        };
5359
5360        assert_eq!(node.expression.inferred_type(), Some(&expected));
5361        let virtual_child = node
5362            .downstream
5363            .iter()
5364            .find(|child| child.source_kind == SourceKind::Virtual)
5365            .expect("virtual UNNEST output");
5366        assert_eq!(virtual_child.expression.inferred_type(), Some(&expected));
5367    }
5368
5369    #[test]
5370    fn test_lineage_with_schema_resolves_struct_fields_issue_408() {
5371        let struct_type = DataType::Struct {
5372            fields: vec![crate::expressions::StructField::new(
5373                "field_value".into(),
5374                DataType::Text,
5375            )],
5376            nested: true,
5377        };
5378        let mut schema = MappingSchema::with_dialect(DialectType::DuckDB);
5379        schema
5380            .add_table(
5381                "source_table",
5382                &[
5383                    ("composite_value".into(), struct_type.clone()),
5384                    (
5385                        "nested_items".into(),
5386                        DataType::Array {
5387                            element_type: Box::new(struct_type),
5388                            dimension: None,
5389                        },
5390                    ),
5391                ],
5392                None,
5393            )
5394            .expect("schema setup");
5395
5396        let direct = parse_dialect(
5397            "SELECT composite_value.field_value AS output_value FROM source_table",
5398            DialectType::DuckDB,
5399        );
5400        let direct_node = lineage_with_schema(
5401            "output_value",
5402            &direct,
5403            Some(&schema),
5404            Some(DialectType::DuckDB),
5405            false,
5406        )
5407        .expect("direct struct lineage");
5408        assert_eq!(
5409            direct_node.expression.inferred_type(),
5410            Some(&DataType::Text)
5411        );
5412        assert_lineage_contains(&direct_node, "source_table.composite_value");
5413
5414        let unnested = parse_dialect(
5415            "SELECT item.field_value AS output_value FROM source_table s \
5416             CROSS JOIN UNNEST(s.nested_items) AS expanded(item)",
5417            DialectType::DuckDB,
5418        );
5419        let unnest_node = lineage_with_schema(
5420            "output_value",
5421            &unnested,
5422            Some(&schema),
5423            Some(DialectType::DuckDB),
5424            false,
5425        )
5426        .expect("UNNEST struct lineage");
5427        assert_eq!(
5428            unnest_node.expression.inferred_type(),
5429            Some(&DataType::Text)
5430        );
5431        assert!(
5432            unnest_node.walk().any(|node| {
5433                node.source_kind == SourceKind::Table
5434                    && node.source_name == "source_table"
5435                    && node.name.ends_with(".nested_items")
5436            }),
5437            "expected physical nested_items lineage, got {:?}",
5438            lineage_names(&unnest_node)
5439        );
5440    }
5441
5442    #[test]
5443    fn test_lineage_lateral_view_columns_resolve_to_virtual_sources() {
5444        let cases = [
5445            (
5446                DialectType::Spark,
5447                "SELECT x AS out FROM t LATERAL VIEW EXPLODE(items) u AS x",
5448            ),
5449            (
5450                DialectType::Hive,
5451                "SELECT x AS out FROM t LATERAL VIEW EXPLODE(items) u AS x",
5452            ),
5453        ];
5454
5455        for (dialect, sql) in cases {
5456            let expr = parse_one(sql, dialect).unwrap_or_else(|e| panic!("parse {dialect:?}: {e}"));
5457            let node = lineage("out", &expr, Some(dialect), false)
5458                .unwrap_or_else(|e| panic!("lineage {dialect:?}: {e}"));
5459            let virtual_child = node
5460                .downstream
5461                .first()
5462                .unwrap_or_else(|| panic!("expected virtual child for {dialect:?}"));
5463
5464            assert_eq!(virtual_child.name, "_0.x");
5465            assert_eq!(virtual_child.source_name, "_0");
5466            assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5467            assert_eq!(virtual_child.source_alias.as_deref(), Some("u"));
5468
5469            let real_child = virtual_child
5470                .downstream
5471                .first()
5472                .unwrap_or_else(|| panic!("expected table dependency for {dialect:?}"));
5473            assert_eq!(real_child.name, "t.items");
5474            assert_eq!(real_child.source_kind, SourceKind::Table);
5475        }
5476    }
5477
5478    #[test]
5479    fn test_lineage_snowflake_lateral_flatten_is_virtual_source() {
5480        let expr = parse_one(
5481            "SELECT f.value AS value FROM raw_events, LATERAL FLATTEN(INPUT => payload:items) AS f",
5482            DialectType::Snowflake,
5483        )
5484        .expect("parse");
5485
5486        let node = lineage("value", &expr, Some(DialectType::Snowflake), false).expect("lineage");
5487        let virtual_child = node.downstream.first().expect("virtual flatten source");
5488        assert_eq!(virtual_child.name, "_0.value");
5489        assert_eq!(virtual_child.source_name, "_0");
5490        assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5491        assert_eq!(virtual_child.source_alias.as_deref(), Some("f"));
5492
5493        let real_child = virtual_child
5494            .downstream
5495            .first()
5496            .expect("FLATTEN input should depend on raw_events.payload");
5497        assert_eq!(real_child.name, "raw_events.payload");
5498        assert_eq!(real_child.source_kind, SourceKind::Table);
5499    }
5500
5501    #[test]
5502    fn test_lineage_with_schema_snowflake_datediff_date_part_issue_61() {
5503        let expr = parse_one(
5504            "SELECT DATEDIFF(day, date_utc, CURRENT_DATE()) AS recency FROM fact.some_daily_metrics",
5505            DialectType::Snowflake,
5506        )
5507        .expect("parse");
5508
5509        let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5510        schema
5511            .add_table(
5512                "fact.some_daily_metrics",
5513                &[("date_utc".to_string(), DataType::Date)],
5514                None,
5515            )
5516            .expect("schema setup");
5517
5518        let node = lineage_with_schema(
5519            "recency",
5520            &expr,
5521            Some(&schema),
5522            Some(DialectType::Snowflake),
5523            false,
5524        )
5525        .expect("lineage_with_schema should not treat date part as a column");
5526
5527        let names = node.downstream_names();
5528        assert!(
5529            names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5530            "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5531            names
5532        );
5533        assert!(
5534            !names.iter().any(|n| n.ends_with(".day") || n == "day"),
5535            "Did not expect date part to appear as lineage column, got: {:?}",
5536            names
5537        );
5538    }
5539
5540    #[test]
5541    fn test_snowflake_datediff_parses_to_typed_ast() {
5542        let expr = parse_one(
5543            "SELECT DATEDIFF(day, date_utc, CURRENT_DATE()) AS recency FROM fact.some_daily_metrics",
5544            DialectType::Snowflake,
5545        )
5546        .expect("parse");
5547
5548        match expr {
5549            Expression::Select(select) => match &select.expressions[0] {
5550                Expression::Alias(alias) => match &alias.this {
5551                    Expression::DateDiff(f) => {
5552                        assert_eq!(f.unit, Some(crate::expressions::IntervalUnit::Day));
5553                    }
5554                    other => panic!("expected DateDiff, got {other:?}"),
5555                },
5556                other => panic!("expected Alias, got {other:?}"),
5557            },
5558            other => panic!("expected Select, got {other:?}"),
5559        }
5560    }
5561
5562    #[test]
5563    fn test_lineage_with_schema_snowflake_dateadd_date_part_issue_followup() {
5564        let expr = parse_one(
5565            "SELECT DATEADD(day, 1, date_utc) AS next_day FROM fact.some_daily_metrics",
5566            DialectType::Snowflake,
5567        )
5568        .expect("parse");
5569
5570        let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5571        schema
5572            .add_table(
5573                "fact.some_daily_metrics",
5574                &[("date_utc".to_string(), DataType::Date)],
5575                None,
5576            )
5577            .expect("schema setup");
5578
5579        let node = lineage_with_schema(
5580            "next_day",
5581            &expr,
5582            Some(&schema),
5583            Some(DialectType::Snowflake),
5584            false,
5585        )
5586        .expect("lineage_with_schema should not treat DATEADD date part as a column");
5587
5588        let names = node.downstream_names();
5589        assert!(
5590            names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5591            "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5592            names
5593        );
5594        assert!(
5595            !names.iter().any(|n| n.ends_with(".day") || n == "day"),
5596            "Did not expect date part to appear as lineage column, got: {:?}",
5597            names
5598        );
5599    }
5600
5601    #[test]
5602    fn test_lineage_with_schema_snowflake_date_part_identifier_issue_followup() {
5603        let expr = parse_one(
5604            "SELECT DATE_PART(day, date_utc) AS day_part FROM fact.some_daily_metrics",
5605            DialectType::Snowflake,
5606        )
5607        .expect("parse");
5608
5609        let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5610        schema
5611            .add_table(
5612                "fact.some_daily_metrics",
5613                &[("date_utc".to_string(), DataType::Date)],
5614                None,
5615            )
5616            .expect("schema setup");
5617
5618        let node = lineage_with_schema(
5619            "day_part",
5620            &expr,
5621            Some(&schema),
5622            Some(DialectType::Snowflake),
5623            false,
5624        )
5625        .expect("lineage_with_schema should not treat DATE_PART identifier as a column");
5626
5627        let names = node.downstream_names();
5628        assert!(
5629            names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5630            "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5631            names
5632        );
5633        assert!(
5634            !names.iter().any(|n| n.ends_with(".day") || n == "day"),
5635            "Did not expect date part to appear as lineage column, got: {:?}",
5636            names
5637        );
5638    }
5639
5640    #[test]
5641    fn test_lineage_with_schema_snowflake_date_part_string_literal_control() {
5642        let expr = parse_one(
5643            "SELECT DATE_PART('day', date_utc) AS day_part FROM fact.some_daily_metrics",
5644            DialectType::Snowflake,
5645        )
5646        .expect("parse");
5647
5648        let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5649        schema
5650            .add_table(
5651                "fact.some_daily_metrics",
5652                &[("date_utc".to_string(), DataType::Date)],
5653                None,
5654            )
5655            .expect("schema setup");
5656
5657        let node = lineage_with_schema(
5658            "day_part",
5659            &expr,
5660            Some(&schema),
5661            Some(DialectType::Snowflake),
5662            false,
5663        )
5664        .expect("quoted DATE_PART should continue to work");
5665
5666        let names = node.downstream_names();
5667        assert!(
5668            names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5669            "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5670            names
5671        );
5672    }
5673
5674    #[test]
5675    fn test_snowflake_dateadd_date_part_identifier_stays_generic_function() {
5676        let expr = parse_one(
5677            "SELECT DATEADD(day, 1, date_utc) AS next_day FROM fact.some_daily_metrics",
5678            DialectType::Snowflake,
5679        )
5680        .expect("parse");
5681
5682        match expr {
5683            Expression::Select(select) => match &select.expressions[0] {
5684                Expression::Alias(alias) => match &alias.this {
5685                    Expression::Function(f) => {
5686                        assert_eq!(f.name.to_uppercase(), "DATEADD");
5687                        assert!(matches!(&f.args[0], Expression::Var(v) if v.this == "day"));
5688                    }
5689                    other => panic!("expected generic DATEADD function, got {other:?}"),
5690                },
5691                other => panic!("expected Alias, got {other:?}"),
5692            },
5693            other => panic!("expected Select, got {other:?}"),
5694        }
5695    }
5696
5697    #[test]
5698    fn test_snowflake_date_part_identifier_stays_generic_function_with_var_arg() {
5699        let expr = parse_one(
5700            "SELECT DATE_PART(day, date_utc) AS day_part FROM fact.some_daily_metrics",
5701            DialectType::Snowflake,
5702        )
5703        .expect("parse");
5704
5705        match expr {
5706            Expression::Select(select) => match &select.expressions[0] {
5707                Expression::Alias(alias) => match &alias.this {
5708                    Expression::Function(f) => {
5709                        assert_eq!(f.name.to_uppercase(), "DATE_PART");
5710                        assert!(matches!(&f.args[0], Expression::Var(v) if v.this == "day"));
5711                    }
5712                    other => panic!("expected generic DATE_PART function, got {other:?}"),
5713                },
5714                other => panic!("expected Alias, got {other:?}"),
5715            },
5716            other => panic!("expected Select, got {other:?}"),
5717        }
5718    }
5719
5720    #[test]
5721    fn test_snowflake_date_part_string_literal_stays_generic_function() {
5722        let expr = parse_one(
5723            "SELECT DATE_PART('day', date_utc) AS day_part FROM fact.some_daily_metrics",
5724            DialectType::Snowflake,
5725        )
5726        .expect("parse");
5727
5728        match expr {
5729            Expression::Select(select) => match &select.expressions[0] {
5730                Expression::Alias(alias) => match &alias.this {
5731                    Expression::Function(f) => {
5732                        assert_eq!(f.name.to_uppercase(), "DATE_PART");
5733                    }
5734                    other => panic!("expected generic DATE_PART function, got {other:?}"),
5735                },
5736                other => panic!("expected Alias, got {other:?}"),
5737            },
5738            other => panic!("expected Select, got {other:?}"),
5739        }
5740    }
5741
5742    #[test]
5743    fn test_lineage_join() {
5744        let expr = parse("SELECT t.a, s.b FROM t JOIN s ON t.id = s.id");
5745
5746        let node_a = lineage("a", &expr, None, false).unwrap();
5747        let names_a = node_a.downstream_names();
5748        assert!(
5749            names_a.iter().any(|n| n == "t.a"),
5750            "Expected t.a, got: {:?}",
5751            names_a
5752        );
5753
5754        let node_b = lineage("b", &expr, None, false).unwrap();
5755        let names_b = node_b.downstream_names();
5756        assert!(
5757            names_b.iter().any(|n| n == "s.b"),
5758            "Expected s.b, got: {:?}",
5759            names_b
5760        );
5761    }
5762
5763    #[test]
5764    fn test_lineage_alias_leaf_has_resolved_source_name() {
5765        let expr = parse("SELECT t1.col1 FROM table1 t1 JOIN table2 t2 ON t1.id = t2.id");
5766        let node = lineage("col1", &expr, None, false).unwrap();
5767
5768        // Keep alias in the display lineage edge.
5769        let names = node.downstream_names();
5770        assert!(
5771            names.iter().any(|n| n == "t1.col1"),
5772            "Expected aliased column edge t1.col1, got: {:?}",
5773            names
5774        );
5775
5776        // Leaf should expose the resolved base table for consumers.
5777        let leaf = node
5778            .downstream
5779            .iter()
5780            .find(|n| n.name == "t1.col1")
5781            .expect("Expected t1.col1 leaf");
5782        assert_eq!(leaf.source_name, "table1");
5783        match &leaf.source {
5784            Expression::Table(table) => assert_eq!(table.name.name, "table1"),
5785            _ => panic!("Expected leaf source to be a table expression"),
5786        }
5787    }
5788
5789    #[test]
5790    fn test_lineage_derived_table() {
5791        let expr = parse("SELECT x.a FROM (SELECT a FROM t) AS x");
5792        let node = lineage("a", &expr, None, false).unwrap();
5793
5794        assert_eq!(node.name, "a");
5795        // Should trace through the derived table to t.a
5796        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5797        assert!(
5798            all_names.iter().any(|n| n == "t.a"),
5799            "Expected to trace through derived table to t.a, got: {:?}",
5800            all_names
5801        );
5802    }
5803
5804    #[test]
5805    fn test_lineage_cte() {
5806        let expr = parse("WITH cte AS (SELECT a FROM t) SELECT a FROM cte");
5807        let node = lineage("a", &expr, None, false).unwrap();
5808
5809        assert_eq!(node.name, "a");
5810        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5811        assert!(
5812            all_names.iter().any(|n| n == "t.a"),
5813            "Expected to trace through CTE to t.a, got: {:?}",
5814            all_names
5815        );
5816    }
5817
5818    #[test]
5819    fn test_lineage_union() {
5820        let expr = parse("SELECT a FROM t1 UNION SELECT a FROM t2");
5821        let node = lineage("a", &expr, None, false).unwrap();
5822
5823        assert_eq!(node.name, "a");
5824        // Should have 2 downstream branches
5825        assert_eq!(
5826            node.downstream.len(),
5827            2,
5828            "Expected 2 branches for UNION, got {}",
5829            node.downstream.len()
5830        );
5831    }
5832
5833    #[test]
5834    fn test_lineage_cte_union() {
5835        let expr = parse("WITH cte AS (SELECT a FROM t1 UNION SELECT a FROM t2) SELECT a FROM cte");
5836        let node = lineage("a", &expr, None, false).unwrap();
5837
5838        // Should trace through CTE into both UNION branches
5839        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5840        assert!(
5841            all_names.len() >= 3,
5842            "Expected at least 3 nodes for CTE with UNION, got: {:?}",
5843            all_names
5844        );
5845    }
5846
5847    #[test]
5848    fn test_lineage_star() {
5849        let expr = parse("SELECT * FROM t");
5850        let node = lineage("*", &expr, None, false).unwrap();
5851
5852        assert_eq!(node.name, "*");
5853        // Should have downstream for table t
5854        assert!(
5855            !node.downstream.is_empty(),
5856            "Star should produce downstream nodes"
5857        );
5858    }
5859
5860    #[test]
5861    fn test_lineage_subquery_in_select() {
5862        let expr = parse("SELECT (SELECT MAX(b) FROM s) AS x FROM t");
5863        let node = lineage("x", &expr, None, false).unwrap();
5864
5865        assert_eq!(node.name, "x");
5866        // Should have traced into the scalar subquery
5867        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5868        assert!(
5869            all_names.len() >= 2,
5870            "Expected tracing into scalar subquery, got: {:?}",
5871            all_names
5872        );
5873    }
5874
5875    #[test]
5876    fn test_lineage_multiple_columns() {
5877        let expr = parse("SELECT a, b FROM t");
5878
5879        let node_a = lineage("a", &expr, None, false).unwrap();
5880        let node_b = lineage("b", &expr, None, false).unwrap();
5881
5882        assert_eq!(node_a.name, "a");
5883        assert_eq!(node_b.name, "b");
5884
5885        // Each should trace independently
5886        let names_a = node_a.downstream_names();
5887        let names_b = node_b.downstream_names();
5888        assert!(names_a.iter().any(|n| n == "t.a"));
5889        assert!(names_b.iter().any(|n| n == "t.b"));
5890    }
5891
5892    #[test]
5893    fn test_get_source_tables() {
5894        let expr = parse("SELECT t.a, s.b FROM t JOIN s ON t.id = s.id");
5895        let node = lineage("a", &expr, None, false).unwrap();
5896
5897        let tables = get_source_tables(&node);
5898        assert!(
5899            tables.contains("t"),
5900            "Expected source table 't', got: {:?}",
5901            tables
5902        );
5903    }
5904
5905    #[test]
5906    fn test_lineage_column_not_found() {
5907        let expr = parse("SELECT a FROM t");
5908        let result = lineage("nonexistent", &expr, None, false);
5909        assert!(result.is_err());
5910    }
5911
5912    #[test]
5913    fn test_lineage_nested_cte() {
5914        let expr = parse(
5915            "WITH cte1 AS (SELECT a FROM t), \
5916             cte2 AS (SELECT a FROM cte1) \
5917             SELECT a FROM cte2",
5918        );
5919        let node = lineage("a", &expr, None, false).unwrap();
5920
5921        // Should trace through cte2 → cte1 → t
5922        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5923        assert!(
5924            all_names.len() >= 3,
5925            "Expected to trace through nested CTEs, got: {:?}",
5926            all_names
5927        );
5928    }
5929
5930    #[test]
5931    fn test_lineage_deeply_nested_cte_reaches_base_table() {
5932        let expr = parse(
5933            "WITH outer_cte AS (\
5934             WITH middle_cte AS (\
5935             WITH inner_cte AS (SELECT x AS col FROM base_table) \
5936             SELECT col FROM inner_cte\
5937             ) SELECT col FROM middle_cte\
5938             ) SELECT col FROM outer_cte",
5939        );
5940        let node = lineage("col", &expr, None, false).unwrap();
5941
5942        assert_lineage_contains(&node, "base_table.x");
5943        for cte_name in ["outer_cte", "middle_cte", "inner_cte"] {
5944            assert!(
5945                node.walk().any(|child| child.source_name == cte_name),
5946                "expected lineage to include CTE {cte_name}, got {:?}",
5947                lineage_names(&node)
5948            );
5949        }
5950    }
5951
5952    #[test]
5953    fn test_lineage_reused_nested_cte_traces_each_reference() {
5954        let expr = parse(
5955            "WITH shared AS (\
5956             WITH nested AS (SELECT x AS col FROM base_table) \
5957             SELECT col FROM nested\
5958             ) \
5959             SELECT s0.col + s1.col + s2.col AS total \
5960             FROM shared AS s0 \
5961             CROSS JOIN shared AS s1 \
5962             CROSS JOIN shared AS s2",
5963        );
5964        let node = lineage("total", &expr, None, false).unwrap();
5965
5966        let base_references = node
5967            .walk()
5968            .filter(|child| child.name == "base_table.x")
5969            .count();
5970        assert_eq!(
5971            base_references,
5972            3,
5973            "each shared CTE reference should reach base_table.x: {:?}",
5974            lineage_names(&node)
5975        );
5976    }
5977
5978    #[test]
5979    fn test_trim_selects_true() {
5980        let expr = parse("SELECT a, b, c FROM t");
5981        let node = lineage("a", &expr, None, true).unwrap();
5982
5983        // The source should be trimmed to only include 'a'
5984        if let Expression::Select(select) = &node.source {
5985            assert_eq!(
5986                select.expressions.len(),
5987                1,
5988                "Trimmed source should have 1 expression, got {}",
5989                select.expressions.len()
5990            );
5991        } else {
5992            panic!("Expected Select source");
5993        }
5994    }
5995
5996    #[test]
5997    fn test_trim_selects_false() {
5998        let expr = parse("SELECT a, b, c FROM t");
5999        let node = lineage("a", &expr, None, false).unwrap();
6000
6001        // The source should keep all columns
6002        if let Expression::Select(select) = &node.source {
6003            assert_eq!(
6004                select.expressions.len(),
6005                3,
6006                "Untrimmed source should have 3 expressions"
6007            );
6008        } else {
6009            panic!("Expected Select source");
6010        }
6011    }
6012
6013    #[test]
6014    fn test_lineage_expression_in_select() {
6015        let expr = parse("SELECT a + b AS c FROM t");
6016        let node = lineage("c", &expr, None, false).unwrap();
6017
6018        // Should trace to both a and b from t
6019        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6020        assert!(
6021            all_names.len() >= 3,
6022            "Expected to trace a + b to both columns, got: {:?}",
6023            all_names
6024        );
6025    }
6026
6027    #[test]
6028    fn test_set_operation_by_index() {
6029        let expr = parse("SELECT a FROM t1 UNION SELECT b FROM t2");
6030
6031        // Trace column "a" which is at index 0
6032        let node = lineage("a", &expr, None, false).unwrap();
6033
6034        // UNION branches should be traced by index
6035        assert_eq!(node.downstream.len(), 2);
6036    }
6037
6038    #[test]
6039    fn test_issue_384_output_columns_preserve_unknown_positions() {
6040        let expr = parse("SELECT a, 1, *, tail FROM unknown_source");
6041        let output = output_columns(&expr, None).expect("output columns");
6042
6043        assert!(!output.ordinal_complete);
6044        assert_eq!(
6045            output.columns,
6046            vec![
6047                OutputColumn::Named {
6048                    name: "a".to_string(),
6049                    ordinal: Some(0),
6050                },
6051                OutputColumn::Unnamed { ordinal: Some(1) },
6052                OutputColumn::Wildcard {
6053                    qualifier: None,
6054                    start_ordinal: Some(2),
6055                },
6056                OutputColumn::Named {
6057                    name: "tail".to_string(),
6058                    ordinal: None,
6059                },
6060            ]
6061        );
6062    }
6063
6064    #[test]
6065    fn test_issue_384_output_columns_use_leftmost_set_operation_branch() {
6066        let expr = parse(
6067            "SELECT * FROM unknown_source UNION ALL \
6068             SELECT known_first AS x, known_second AS y FROM known_source",
6069        );
6070        let output = output_columns(&expr, None).expect("output columns");
6071
6072        assert!(!output.ordinal_complete);
6073        assert_eq!(
6074            output.columns,
6075            vec![OutputColumn::Wildcard {
6076                qualifier: None,
6077                start_ordinal: Some(0),
6078            }]
6079        );
6080    }
6081
6082    #[test]
6083    fn test_issue_384_schema_expands_output_wildcard() {
6084        let expr = parse("SELECT * FROM unknown_source");
6085        let mut schema = MappingSchema::new();
6086        schema
6087            .add_table(
6088                "unknown_source",
6089                &[
6090                    ("first_col".to_string(), DataType::Text),
6091                    ("second_col".to_string(), DataType::Text),
6092                ],
6093                None,
6094            )
6095            .expect("schema setup");
6096
6097        let output = output_columns_with_schema(&expr, Some(&schema), None)
6098            .expect("schema-aware output columns");
6099        assert!(output.ordinal_complete);
6100        assert_eq!(
6101            output.columns,
6102            vec![
6103                OutputColumn::Named {
6104                    name: "first_col".to_string(),
6105                    ordinal: Some(0),
6106                },
6107                OutputColumn::Named {
6108                    name: "second_col".to_string(),
6109                    ordinal: Some(1),
6110                },
6111            ]
6112        );
6113    }
6114
6115    #[test]
6116    fn test_issue_411_duckdb_and_snowflake_union_by_name_layout_and_lineage() {
6117        let sql = "SELECT 1 AS left_value UNION ALL BY NAME SELECT 2 AS right_value";
6118
6119        for dialect in [DialectType::DuckDB, DialectType::Snowflake] {
6120            let expr = parse_dialect(sql, dialect);
6121            let output = output_columns(&expr, Some(dialect)).expect("output columns");
6122            assert_eq!(
6123                output.columns,
6124                vec![
6125                    OutputColumn::Named {
6126                        name: "left_value".to_string(),
6127                        ordinal: Some(0),
6128                    },
6129                    OutputColumn::Named {
6130                        name: "right_value".to_string(),
6131                        ordinal: Some(1),
6132                    },
6133                ],
6134                "unexpected output layout for {dialect:?}"
6135            );
6136
6137            let left = lineage_at(0, &expr, Some(dialect), false).expect("left lineage");
6138            assert_eq!(
6139                left.downstream
6140                    .iter()
6141                    .map(|node| node.name.as_str())
6142                    .collect::<Vec<_>>(),
6143                vec!["left_value"]
6144            );
6145            assert_eq!(
6146                left.downstream[0].set_branch,
6147                Some(SetBranch {
6148                    operator: SetOperator::Union,
6149                    ordinal: 0,
6150                    all: true,
6151                })
6152            );
6153
6154            let right = lineage_at(1, &expr, Some(dialect), false).expect("right lineage");
6155            assert_eq!(
6156                right
6157                    .downstream
6158                    .iter()
6159                    .map(|node| node.name.as_str())
6160                    .collect::<Vec<_>>(),
6161                vec!["right_value"]
6162            );
6163            assert_eq!(
6164                right.downstream[0].set_branch,
6165                Some(SetBranch {
6166                    operator: SetOperator::Union,
6167                    ordinal: 1,
6168                    all: true,
6169                })
6170            );
6171        }
6172    }
6173
6174    #[test]
6175    fn test_issue_411_bigquery_strict_by_name_reorders_lineage() {
6176        let dialect = DialectType::BigQuery;
6177        let expr = parse_dialect(
6178            "SELECT 1 AS a, 2 AS b UNION ALL BY NAME SELECT 3 AS b, 4 AS a",
6179            dialect,
6180        );
6181        let output = output_columns(&expr, Some(dialect)).expect("output columns");
6182        assert_eq!(
6183            output.columns,
6184            vec![
6185                OutputColumn::Named {
6186                    name: "a".to_string(),
6187                    ordinal: Some(0),
6188                },
6189                OutputColumn::Named {
6190                    name: "b".to_string(),
6191                    ordinal: Some(1),
6192                },
6193            ]
6194        );
6195
6196        for (ordinal, expected_name) in [(0, "a"), (1, "b")] {
6197            let node = lineage_at(ordinal, &expr, Some(dialect), false).expect("lineage");
6198            assert_eq!(
6199                node.downstream
6200                    .iter()
6201                    .map(|child| child.name.as_str())
6202                    .collect::<Vec<_>>(),
6203                vec![expected_name, expected_name]
6204            );
6205        }
6206    }
6207
6208    #[test]
6209    fn test_issue_411_bigquery_by_name_modes_and_explicit_order() {
6210        let dialect = DialectType::BigQuery;
6211        for (sql, expected) in [
6212            (
6213                "SELECT 1 AS a, 2 AS b INNER UNION ALL BY NAME SELECT 3 AS b, 4 AS c",
6214                vec!["b"],
6215            ),
6216            (
6217                "SELECT 1 AS a, 2 AS b LEFT OUTER UNION ALL BY NAME SELECT 3 AS b, 4 AS c",
6218                vec!["a", "b"],
6219            ),
6220            (
6221                "SELECT 1 AS a, 2 AS b FULL OUTER UNION ALL BY NAME SELECT 3 AS b, 4 AS c",
6222                vec!["a", "b", "c"],
6223            ),
6224            (
6225                "SELECT 1 AS a, 2 AS b FULL OUTER UNION ALL BY NAME ON (c, a) SELECT 3 AS b, 4 AS c",
6226                vec!["c", "a"],
6227            ),
6228            (
6229                "SELECT 1 AS a, 2 AS b INTERSECT DISTINCT BY NAME SELECT 3 AS b, 4 AS a",
6230                vec!["a", "b"],
6231            ),
6232            (
6233                "SELECT 1 AS a, 2 AS b EXCEPT DISTINCT BY NAME SELECT 3 AS b, 4 AS a",
6234                vec!["a", "b"],
6235            ),
6236        ] {
6237            let expr = parse_dialect(sql, dialect);
6238            let output = output_columns(&expr, Some(dialect)).expect("output columns");
6239            assert_eq!(
6240                output
6241                    .columns
6242                    .iter()
6243                    .filter_map(|column| match column {
6244                        OutputColumn::Named { name, .. } => Some(name.as_str()),
6245                        _ => None,
6246                    })
6247                    .collect::<Vec<_>>(),
6248                expected,
6249                "unexpected output layout for {sql}"
6250            );
6251        }
6252    }
6253
6254    #[test]
6255    fn test_issue_411_name_matching_uses_dialect_identifier_rules() {
6256        let sql = "SELECT 1 AS \"lower\" UNION ALL BY NAME SELECT 2 AS lower";
6257
6258        let duckdb = parse_dialect(sql, DialectType::DuckDB);
6259        assert_eq!(
6260            output_columns(&duckdb, Some(DialectType::DuckDB))
6261                .expect("DuckDB output")
6262                .columns
6263                .len(),
6264            1,
6265            "DuckDB quoted identifiers are case-insensitive"
6266        );
6267
6268        let snowflake = parse_dialect(sql, DialectType::Snowflake);
6269        assert_eq!(
6270            output_columns(&snowflake, Some(DialectType::Snowflake))
6271                .expect("Snowflake output")
6272                .columns
6273                .len(),
6274            2,
6275            "Snowflake quoted lowercase and unquoted uppercase identifiers differ"
6276        );
6277    }
6278
6279    #[test]
6280    fn test_issue_383_lineage_at_traces_resolvable_set_operation_branch() {
6281        let expr = parse(
6282            "SELECT * FROM unknown_source UNION ALL \
6283             SELECT known_first AS x, known_second AS y FROM known_source",
6284        );
6285        let node = lineage_at(1, &expr, None, false).expect("ordinal lineage");
6286
6287        assert_lineage_contains(&node, "known_source.known_second");
6288        assert_eq!(node.downstream.len(), 1);
6289        assert_eq!(
6290            node.downstream[0].set_branch,
6291            Some(SetBranch {
6292                operator: SetOperator::Union,
6293                ordinal: 1,
6294                all: true,
6295            })
6296        );
6297
6298        let expr = parse(
6299            "SELECT known_first, known_second FROM known_source EXCEPT \
6300             SELECT * FROM unknown_source",
6301        );
6302        let node = lineage_at(1, &expr, None, false).expect("left-only ordinal lineage");
6303
6304        assert_lineage_contains(&node, "known_source.known_second");
6305        assert_eq!(node.downstream.len(), 1);
6306        assert_eq!(
6307            node.downstream[0].set_branch,
6308            Some(SetBranch {
6309                operator: SetOperator::Except,
6310                ordinal: 0,
6311                all: false,
6312            })
6313        );
6314    }
6315
6316    #[test]
6317    fn test_issue_383_unresolved_wildcard_does_not_shift_ordinal() {
6318        let expr = parse(
6319            "SELECT *, tail FROM unknown_source UNION ALL \
6320             SELECT known_first, known_second FROM known_source",
6321        );
6322        let node = lineage_at(1, &expr, None, false).expect("partial ordinal lineage");
6323        let names = lineage_names(&node);
6324
6325        assert!(names.iter().any(|name| name == "known_source.known_second"));
6326        assert!(!names
6327            .iter()
6328            .any(|name| name.ends_with(".tail") || name == "tail"));
6329    }
6330
6331    #[test]
6332    fn test_issue_383_lineage_at_ignores_branch_output_names() {
6333        let expr = parse(
6334            "SELECT left_value AS left_name FROM left_source UNION ALL \
6335             SELECT right_value AS right_name FROM right_source",
6336        );
6337        let node = lineage_at(0, &expr, None, false).expect("ordinal lineage");
6338
6339        assert_lineage_contains(&node, "left_source.left_value");
6340        assert_lineage_contains(&node, "right_source.right_value");
6341    }
6342
6343    #[test]
6344    fn test_issue_383_lineage_at_supports_all_set_operations() {
6345        for (sql_operator, operator, all) in [
6346            ("UNION ALL", SetOperator::Union, true),
6347            ("INTERSECT", SetOperator::Intersect, false),
6348            ("EXCEPT", SetOperator::Except, false),
6349        ] {
6350            let expr = parse(&format!(
6351                "SELECT left_value AS left_name FROM left_source {sql_operator} \
6352                 SELECT right_value AS right_name FROM right_source"
6353            ));
6354            let node = lineage_at(0, &expr, None, false).expect("ordinal lineage");
6355            assert_eq!(
6356                node.downstream.len(),
6357                2,
6358                "expected both branches for {sql_operator}"
6359            );
6360            assert_eq!(
6361                node.downstream
6362                    .iter()
6363                    .map(|child| child.set_branch)
6364                    .collect::<Vec<_>>(),
6365                vec![
6366                    Some(SetBranch {
6367                        operator,
6368                        ordinal: 0,
6369                        all,
6370                    }),
6371                    Some(SetBranch {
6372                        operator,
6373                        ordinal: 1,
6374                        all,
6375                    }),
6376                ]
6377            );
6378        }
6379    }
6380
6381    #[test]
6382    fn test_issue_383_lineage_at_with_schema_expands_wildcard() {
6383        let expr = parse(
6384            "SELECT * FROM unknown_source UNION ALL \
6385             SELECT known_first, known_second FROM known_source",
6386        );
6387        let mut schema = MappingSchema::new();
6388        schema
6389            .add_table(
6390                "unknown_source",
6391                &[
6392                    ("first_col".to_string(), DataType::Text),
6393                    ("second_col".to_string(), DataType::Text),
6394                ],
6395                None,
6396            )
6397            .expect("schema setup");
6398        schema
6399            .add_table(
6400                "known_source",
6401                &[
6402                    ("known_first".to_string(), DataType::Text),
6403                    ("known_second".to_string(), DataType::Text),
6404                ],
6405                None,
6406            )
6407            .expect("schema setup");
6408
6409        let node = lineage_at_with_schema(1, &expr, Some(&schema), None, false)
6410            .expect("schema-aware ordinal lineage");
6411        assert_lineage_contains(&node, "unknown_source.second_col");
6412        assert_lineage_contains(&node, "known_source.known_second");
6413    }
6414
6415    #[test]
6416    fn test_issue_385_structured_lineage_resolution_errors() {
6417        let out_of_range = lineage_at(1, &parse("SELECT a FROM t"), None, false)
6418            .expect_err("ordinal should be out of range");
6419        assert!(matches!(
6420            out_of_range,
6421            Error::ColumnResolution {
6422                target: ColumnResolutionTarget::Ordinal { ordinal: 1 },
6423                reason: ColumnResolutionReason::NotFound,
6424            }
6425        ));
6426
6427        let indeterminate = lineage(
6428            "tail",
6429            &parse(
6430                "SELECT *, tail FROM unknown_source UNION ALL \
6431                 SELECT known_first, known_second FROM known_source",
6432            ),
6433            None,
6434            false,
6435        )
6436        .expect_err("tail ordinal should be indeterminate");
6437        assert!(matches!(
6438            indeterminate,
6439            Error::ColumnResolution {
6440                target: ColumnResolutionTarget::Name { ref name },
6441                reason: ColumnResolutionReason::Indeterminate,
6442            } if name == "tail"
6443        ));
6444
6445        let ambiguous = lineage(
6446            "duplicate",
6447            &parse("SELECT a AS duplicate, b AS duplicate FROM t"),
6448            None,
6449            false,
6450        )
6451        .expect_err("duplicate output name should be ambiguous");
6452        assert!(matches!(
6453            ambiguous,
6454            Error::ColumnResolution {
6455                target: ColumnResolutionTarget::Name { ref name },
6456                reason: ColumnResolutionReason::Ambiguous,
6457            } if name == "duplicate"
6458        ));
6459    }
6460
6461    // --- Tests for column lineage inside function calls (issue #18) ---
6462
6463    fn print_node(node: &LineageNode, indent: usize) {
6464        let pad = "  ".repeat(indent);
6465        println!(
6466            "{pad}name={:?} source_name={:?}",
6467            node.name, node.source_name
6468        );
6469        for child in &node.downstream {
6470            print_node(child, indent + 1);
6471        }
6472    }
6473
6474    #[test]
6475    fn test_issue18_repro() {
6476        // Exact scenario from the issue
6477        let query = "SELECT UPPER(name) as upper_name FROM users";
6478        println!("Query: {query}\n");
6479
6480        let dialect = crate::dialects::Dialect::get(DialectType::BigQuery);
6481        let exprs = dialect.parse(query).unwrap();
6482        let expr = &exprs[0];
6483
6484        let node = lineage("upper_name", expr, Some(DialectType::BigQuery), false).unwrap();
6485        println!("lineage(\"upper_name\"):");
6486        print_node(&node, 1);
6487
6488        let names = node.downstream_names();
6489        assert!(
6490            names.iter().any(|n| n == "users.name"),
6491            "Expected users.name in downstream, got: {:?}",
6492            names
6493        );
6494    }
6495
6496    #[test]
6497    fn test_lineage_bigquery_safe_namespace_issue207() {
6498        let query = r#"
6499WITH import_cte AS (
6500  SELECT timestamp, data, operation
6501  FROM `project`.`dataset`.`source_table`
6502),
6503transform_cte AS (
6504  SELECT
6505    timestamp,
6506    SAFE.PARSE_JSON(data) AS json_data
6507  FROM import_cte
6508)
6509SELECT json_data FROM transform_cte
6510"#;
6511        let expr = parse_one(query, DialectType::BigQuery).expect("parse");
6512        let node = lineage("json_data", &expr, Some(DialectType::BigQuery), false)
6513            .expect("lineage should resolve SAFE.PARSE_JSON arguments");
6514        let names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6515
6516        assert!(
6517            names.iter().any(|name| name == "source_table.data"),
6518            "expected source_table.data in lineage, got {names:?}"
6519        );
6520        assert!(
6521            !names
6522                .iter()
6523                .any(|name| name.eq_ignore_ascii_case("import_cte.safe")),
6524            "did not expect SAFE namespace receiver in lineage, got {names:?}"
6525        );
6526    }
6527
6528    #[test]
6529    fn test_lineage_bigquery_safe_namespace_method_call_guard() {
6530        let expr = parse("SELECT SAFE.PARSE_JSON(data) AS json_data FROM t");
6531        let node = lineage("json_data", &expr, Some(DialectType::BigQuery), false)
6532            .expect("lineage should resolve SAFE.PARSE_JSON arguments");
6533        let names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6534
6535        assert!(
6536            names.iter().any(|name| name == "t.data"),
6537            "expected t.data in lineage, got {names:?}"
6538        );
6539        assert!(
6540            !names.iter().any(|name| name.eq_ignore_ascii_case("t.safe")),
6541            "did not expect SAFE namespace receiver in lineage, got {names:?}"
6542        );
6543    }
6544
6545    #[test]
6546    fn test_lineage_method_call_receiver_control() {
6547        let expr = parse("SELECT obj.METHOD(arg) AS out FROM t");
6548        let node = lineage("out", &expr, None, false).expect("lineage");
6549        let names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6550
6551        assert!(
6552            names.iter().any(|name| name == "t.obj"),
6553            "expected ordinary method receiver to remain in lineage, got {names:?}"
6554        );
6555        assert!(
6556            names.iter().any(|name| name == "t.arg"),
6557            "expected method argument in lineage, got {names:?}"
6558        );
6559    }
6560
6561    #[test]
6562    fn test_lineage_upper_function() {
6563        let expr = parse("SELECT UPPER(name) AS upper_name FROM users");
6564        let node = lineage("upper_name", &expr, None, false).unwrap();
6565
6566        let names = node.downstream_names();
6567        assert!(
6568            names.iter().any(|n| n == "users.name"),
6569            "Expected users.name in downstream, got: {:?}",
6570            names
6571        );
6572    }
6573
6574    #[test]
6575    fn test_lineage_round_function() {
6576        let expr = parse("SELECT ROUND(price, 2) AS rounded FROM products");
6577        let node = lineage("rounded", &expr, None, false).unwrap();
6578
6579        let names = node.downstream_names();
6580        assert!(
6581            names.iter().any(|n| n == "products.price"),
6582            "Expected products.price in downstream, got: {:?}",
6583            names
6584        );
6585    }
6586
6587    #[test]
6588    fn test_lineage_coalesce_function() {
6589        let expr = parse("SELECT COALESCE(a, b) AS val FROM t");
6590        let node = lineage("val", &expr, None, false).unwrap();
6591
6592        let names = node.downstream_names();
6593        assert!(
6594            names.iter().any(|n| n == "t.a"),
6595            "Expected t.a in downstream, got: {:?}",
6596            names
6597        );
6598        assert!(
6599            names.iter().any(|n| n == "t.b"),
6600            "Expected t.b in downstream, got: {:?}",
6601            names
6602        );
6603    }
6604
6605    #[test]
6606    fn test_lineage_count_function() {
6607        let expr = parse("SELECT COUNT(id) AS cnt FROM t");
6608        let node = lineage("cnt", &expr, None, false).unwrap();
6609
6610        let names = node.downstream_names();
6611        assert!(
6612            names.iter().any(|n| n == "t.id"),
6613            "Expected t.id in downstream, got: {:?}",
6614            names
6615        );
6616    }
6617
6618    #[test]
6619    fn test_lineage_sum_function() {
6620        let expr = parse("SELECT SUM(amount) AS total FROM t");
6621        let node = lineage("total", &expr, None, false).unwrap();
6622
6623        let names = node.downstream_names();
6624        assert!(
6625            names.iter().any(|n| n == "t.amount"),
6626            "Expected t.amount in downstream, got: {:?}",
6627            names
6628        );
6629    }
6630
6631    #[test]
6632    fn test_lineage_case_with_nested_functions() {
6633        let expr =
6634            parse("SELECT CASE WHEN x > 0 THEN UPPER(name) ELSE LOWER(name) END AS result FROM t");
6635        let node = lineage("result", &expr, None, false).unwrap();
6636
6637        let names = node.downstream_names();
6638        assert!(
6639            names.iter().any(|n| n == "t.x"),
6640            "Expected t.x in downstream, got: {:?}",
6641            names
6642        );
6643        assert!(
6644            names.iter().any(|n| n == "t.name"),
6645            "Expected t.name in downstream, got: {:?}",
6646            names
6647        );
6648    }
6649
6650    #[test]
6651    fn test_lineage_substring_function() {
6652        let expr = parse("SELECT SUBSTRING(name, 1, 3) AS short FROM t");
6653        let node = lineage("short", &expr, None, false).unwrap();
6654
6655        let names = node.downstream_names();
6656        assert!(
6657            names.iter().any(|n| n == "t.name"),
6658            "Expected t.name in downstream, got: {:?}",
6659            names
6660        );
6661    }
6662
6663    // --- CTE + SELECT * tests (ported from sqlglot test_lineage.py) ---
6664
6665    #[test]
6666    fn test_lineage_cte_select_star() {
6667        // Ported from sqlglot: test_lineage_source_with_star
6668        // WITH y AS (SELECT * FROM x) SELECT a FROM y
6669        // After star expansion: SELECT y.a AS a FROM y
6670        let expr = parse("WITH y AS (SELECT * FROM x) SELECT a FROM y");
6671        let node = lineage("a", &expr, None, false).unwrap();
6672
6673        assert_eq!(node.name, "a");
6674        // Should successfully resolve column 'a' through the CTE
6675        // (previously failed with "Cannot find column 'a' in query")
6676        assert!(
6677            !node.downstream.is_empty(),
6678            "Expected downstream nodes tracing through CTE, got none"
6679        );
6680    }
6681
6682    #[test]
6683    fn test_lineage_schema_less_cte_star_passthrough_resolves_base_column() {
6684        let expr = parse("WITH c AS (SELECT * FROM t) SELECT c.x FROM c");
6685        let node = lineage("x", &expr, None, false).unwrap();
6686
6687        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6688        assert!(
6689            all_names.iter().any(|name| name == "t.x"),
6690            "Expected schema-less CTE star passthrough to reach t.x, got: {:?}",
6691            all_names
6692        );
6693
6694        let cte_node = node
6695            .walk()
6696            .find(|child| child.source_kind == SourceKind::Cte && child.source_name == "c")
6697            .expect("expected CTE hop with source_name c");
6698        assert_eq!(cte_node.source_kind, SourceKind::Cte);
6699        assert_eq!(cte_node.source_name, "c");
6700    }
6701
6702    #[test]
6703    fn test_lineage_schema_less_cte_star_passthrough_with_aggregation() {
6704        let expr = parse(
6705            "WITH c AS (SELECT * FROM t) \
6706             SELECT SUM(c.x) AS s FROM c GROUP BY 1",
6707        );
6708        let node = lineage("s", &expr, None, false).unwrap();
6709
6710        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6711        assert!(
6712            all_names.iter().any(|name| name == "t.x"),
6713            "Expected aggregate over CTE star passthrough to reach t.x, got: {:?}",
6714            all_names
6715        );
6716    }
6717
6718    #[test]
6719    fn test_lineage_schema_less_cte_star_passthrough_with_join_and_alias() {
6720        let expr = parse(
6721            "WITH a AS (SELECT * FROM t1), b AS (SELECT * FROM t2) \
6722             SELECT SUM(b.x) AS s FROM a LEFT JOIN b ON b.id = a.id GROUP BY a.k",
6723        );
6724        let node = lineage("s", &expr, None, false).unwrap();
6725
6726        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6727        assert!(
6728            all_names.iter().any(|name| name == "t2.x"),
6729            "Expected joined CTE star passthrough to reach t2.x, got: {:?}",
6730            all_names
6731        );
6732    }
6733
6734    #[test]
6735    fn test_lineage_schema_less_chained_cte_star_passthrough() {
6736        let expr = parse(
6737            "WITH c1 AS (SELECT * FROM t), \
6738             c2 AS (SELECT * FROM c1), \
6739             c3 AS (SELECT * FROM c2) \
6740             SELECT c3.x FROM c3",
6741        );
6742        let node = lineage("x", &expr, None, false).unwrap();
6743
6744        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6745        assert!(
6746            all_names.iter().any(|name| name == "t.x"),
6747            "Expected chained CTE star passthrough to reach t.x, got: {:?}",
6748            all_names
6749        );
6750    }
6751
6752    #[test]
6753    fn test_lineage_schema_less_unqualified_star_with_multiple_sources_does_not_guess() {
6754        let expr = parse("SELECT * FROM t1 JOIN t2 ON t1.id = t2.id");
6755        let result = lineage("x", &expr, None, false);
6756
6757        assert!(
6758            result.is_err(),
6759            "Unqualified star over multiple sources should remain ambiguous, got: {:?}",
6760            result
6761        );
6762    }
6763
6764    #[test]
6765    fn test_lineage_schema_less_star_passthrough_applies_modifiers() {
6766        let leaves = |dialect, sql, column| {
6767            let expr = parse_dialect(sql, dialect);
6768            lineage(column, &expr, Some(dialect), false).map(|node| {
6769                node.walk()
6770                    .filter(|node| node.downstream.is_empty())
6771                    .map(|node| node.name.clone())
6772                    .collect::<Vec<_>>()
6773            })
6774        };
6775
6776        let sql = "SELECT * EXCLUDE (b) FROM (SELECT * FROM t) s";
6777        assert!(leaves(DialectType::Snowflake, sql, "b").is_err());
6778        assert_eq!(
6779            leaves(DialectType::Snowflake, sql, "a").unwrap(),
6780            vec!["t.a"]
6781        );
6782
6783        let sql = "SELECT * RENAME (a AS d) FROM (SELECT * FROM t) s";
6784        assert_eq!(
6785            leaves(DialectType::Snowflake, sql, "d").unwrap(),
6786            vec!["t.a"]
6787        );
6788        assert!(leaves(DialectType::Snowflake, sql, "a").is_err());
6789
6790        let sql = "SELECT * REPLACE (a * 2 AS a) FROM (SELECT * FROM t) s";
6791        let expr = parse_dialect(sql, DialectType::BigQuery);
6792        let node = lineage("a", &expr, Some(DialectType::BigQuery), false).unwrap();
6793        assert!(matches!(node.expression, Expression::Alias(_)));
6794        assert_eq!(
6795            leaves(DialectType::BigQuery, sql, "a").unwrap(),
6796            vec!["t.a"]
6797        );
6798    }
6799
6800    #[test]
6801    fn test_lineage_cte_select_star_renamed_column() {
6802        // dbt standard pattern: CTE with column rename + outer SELECT *
6803        // This is the primary use case for dbt projects (jaffle-shop etc.)
6804        let expr =
6805            parse("WITH renamed AS (SELECT id AS customer_id FROM source) SELECT * FROM renamed");
6806        let node = lineage("customer_id", &expr, None, false).unwrap();
6807
6808        assert_eq!(node.name, "customer_id");
6809        // Should trace customer_id → renamed CTE → source.id
6810        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6811        assert!(
6812            all_names.len() >= 2,
6813            "Expected at least 2 nodes (customer_id → source), got: {:?}",
6814            all_names
6815        );
6816    }
6817
6818    #[test]
6819    fn test_lineage_cte_select_star_multiple_columns() {
6820        // CTE exposes multiple columns, outer SELECT * should resolve each
6821        let expr = parse("WITH cte AS (SELECT a, b, c FROM t) SELECT * FROM cte");
6822
6823        for col in &["a", "b", "c"] {
6824            let node = lineage(col, &expr, None, false).unwrap();
6825            assert_eq!(node.name, *col);
6826            // Verify lineage resolves without error (star expanded to explicit columns)
6827            let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6828            assert!(
6829                all_names.len() >= 2,
6830                "Expected at least 2 nodes for column {}, got: {:?}",
6831                col,
6832                all_names
6833            );
6834        }
6835    }
6836
6837    #[test]
6838    fn test_lineage_nested_cte_select_star() {
6839        // Nested CTE star expansion: cte2 references cte1 via SELECT *
6840        let expr = parse(
6841            "WITH cte1 AS (SELECT a FROM t), \
6842             cte2 AS (SELECT * FROM cte1) \
6843             SELECT * FROM cte2",
6844        );
6845        let node = lineage("a", &expr, None, false).unwrap();
6846
6847        assert_eq!(node.name, "a");
6848        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6849        assert!(
6850            all_names.len() >= 3,
6851            "Expected at least 3 nodes (a → cte2 → cte1 → t.a), got: {:?}",
6852            all_names
6853        );
6854    }
6855
6856    #[test]
6857    fn test_lineage_three_level_nested_cte_star() {
6858        // Three-level nested CTE: cte3 → cte2 → cte1 → t
6859        let expr = parse(
6860            "WITH cte1 AS (SELECT x FROM t), \
6861             cte2 AS (SELECT * FROM cte1), \
6862             cte3 AS (SELECT * FROM cte2) \
6863             SELECT * FROM cte3",
6864        );
6865        let node = lineage("x", &expr, None, false).unwrap();
6866
6867        assert_eq!(node.name, "x");
6868        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6869        assert!(
6870            all_names.len() >= 4,
6871            "Expected at least 4 nodes through 3-level CTE chain, got: {:?}",
6872            all_names
6873        );
6874    }
6875
6876    #[test]
6877    fn test_lineage_cte_union_star() {
6878        // CTE with UNION body, outer SELECT * should resolve from left branch
6879        let expr = parse(
6880            "WITH cte AS (SELECT a, b FROM t1 UNION ALL SELECT a, b FROM t2) \
6881             SELECT * FROM cte",
6882        );
6883        let node = lineage("a", &expr, None, false).unwrap();
6884
6885        assert_eq!(node.name, "a");
6886        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6887        assert!(
6888            all_names.len() >= 2,
6889            "Expected at least 2 nodes for CTE union star, got: {:?}",
6890            all_names
6891        );
6892    }
6893
6894    #[test]
6895    fn test_issue_368_expand_cte_stars_rewrites_every_union_arm() {
6896        let mut expr = parse_one(ISSUE_368_SQL, DialectType::BigQuery).unwrap();
6897
6898        expand_cte_stars(&mut expr, None);
6899
6900        assert_eq!(
6901            crate::generate(&expr, DialectType::BigQuery).unwrap(),
6902            "WITH base AS (SELECT 1 AS col_a), literal_branch AS (SELECT 2 AS col_a), \
6903             unioned AS (SELECT base.col_a FROM base UNION ALL \
6904             SELECT literal_branch.col_a FROM literal_branch) \
6905             SELECT col_a FROM unioned"
6906        );
6907    }
6908
6909    #[test]
6910    fn test_issue_368_lineage_resolves_non_leftmost_union_star() {
6911        let expr = parse_one(ISSUE_368_SQL, DialectType::BigQuery).unwrap();
6912
6913        let node = lineage("col_a", &expr, Some(DialectType::BigQuery), false).unwrap();
6914        let names = lineage_names(&node);
6915
6916        assert!(
6917            node.walk().any(|child| {
6918                child.name == "col_a"
6919                    && child.source_name == "literal_branch"
6920                    && child.source_kind == SourceKind::Cte
6921            }),
6922            "expected the right UNION branch to resolve to literal_branch.col_a, got {node:#?}"
6923        );
6924        assert!(
6925            !names
6926                .iter()
6927                .any(|name| name == "*" || name == "literal_branch.*"),
6928            "did not expect an unresolved right-branch star, got {names:?}"
6929        );
6930    }
6931
6932    #[test]
6933    fn test_expand_cte_stars_rewrites_all_set_operation_kinds() {
6934        for operator in ["UNION ALL", "INTERSECT", "EXCEPT"] {
6935            let mut expr = parse(&format!(
6936                "WITH left_cte AS (SELECT 1 AS col_a), \
6937                 right_cte AS (SELECT 2 AS col_a), \
6938                 combined AS (SELECT * FROM left_cte {operator} SELECT * FROM right_cte) \
6939                 SELECT col_a FROM combined"
6940            ));
6941
6942            expand_cte_stars(&mut expr, None);
6943
6944            let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6945            assert!(
6946                sql.contains("SELECT left_cte.col_a FROM left_cte"),
6947                "expected left arm expansion for {operator}, got {sql}"
6948            );
6949            assert!(
6950                sql.contains("SELECT right_cte.col_a FROM right_cte"),
6951                "expected right arm expansion for {operator}, got {sql}"
6952            );
6953        }
6954    }
6955
6956    #[test]
6957    fn test_expand_cte_stars_rewrites_nested_parenthesized_set_operations() {
6958        let mut expr = parse(
6959            "WITH a AS (SELECT 1 AS x), \
6960             b AS (SELECT 2 AS x), \
6961             c AS (SELECT 3 AS x), \
6962             combined AS ((SELECT * FROM a UNION ALL SELECT * FROM b) \
6963             UNION ALL SELECT * FROM c) \
6964             SELECT x FROM combined",
6965        );
6966
6967        expand_cte_stars(&mut expr, None);
6968
6969        let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6970        for source in ["a", "b", "c"] {
6971            assert!(
6972                sql.contains(&format!("SELECT {source}.x FROM {source}")),
6973                "expected nested arm {source} to be expanded, got {sql}"
6974            );
6975        }
6976    }
6977
6978    #[test]
6979    fn test_expand_cte_stars_rewrites_root_set_operation() {
6980        let mut expr = parse(
6981            "WITH a AS (SELECT 1 AS x), b AS (SELECT 2 AS x) \
6982             SELECT * FROM a UNION ALL SELECT * FROM b",
6983        );
6984
6985        expand_cte_stars(&mut expr, None);
6986
6987        let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6988        assert!(
6989            sql.contains("SELECT a.x FROM a UNION ALL SELECT b.x FROM b"),
6990            "expected both root UNION arms to be expanded, got {sql}"
6991        );
6992    }
6993
6994    #[test]
6995    fn test_expand_cte_stars_preserves_leftmost_output_names() {
6996        let mut expr = parse(
6997            "WITH a AS (SELECT 1 AS left_name), \
6998             b AS (SELECT 2 AS right_name), \
6999             combined AS (SELECT * FROM a UNION ALL SELECT * FROM b) \
7000             SELECT * FROM combined",
7001        );
7002
7003        expand_cte_stars(&mut expr, None);
7004
7005        let sql = crate::generate(&expr, DialectType::Generic).unwrap();
7006        assert!(
7007            sql.ends_with("SELECT combined.left_name FROM combined"),
7008            "expected the set operation output name to come from the left arm, got {sql}"
7009        );
7010        assert!(
7011            sql.contains("SELECT b.right_name FROM b"),
7012            "expected the differently named right arm to still be expanded, got {sql}"
7013        );
7014    }
7015
7016    #[test]
7017    fn test_expand_cte_stars_applies_except_replace_rename() {
7018        let mut expr = parse_dialect(
7019            "WITH a AS (SELECT 1 AS x, 2 AS y, 3 AS z), \
7020             b AS (SELECT * EXCLUDE (y) RENAME (x AS w) FROM a) \
7021             SELECT * FROM b",
7022            DialectType::Snowflake,
7023        );
7024
7025        expand_cte_stars(&mut expr, None);
7026
7027        assert_eq!(
7028            crate::generate(&expr, DialectType::Snowflake).unwrap(),
7029            "WITH a AS (SELECT 1 AS x, 2 AS y, 3 AS z), \
7030             b AS (SELECT a.x AS w, a.z FROM a) \
7031             SELECT b.w, b.z FROM b"
7032        );
7033
7034        let mut expr = parse_dialect(
7035            "WITH a AS (SELECT 1 AS x, 2 AS y) \
7036             SELECT * REPLACE (y + 1 AS y) FROM a",
7037            DialectType::BigQuery,
7038        );
7039
7040        expand_cte_stars(&mut expr, None);
7041
7042        assert_eq!(
7043            crate::generate(&expr, DialectType::BigQuery).unwrap(),
7044            "WITH a AS (SELECT 1 AS x, 2 AS y) SELECT a.x, y + 1 AS y FROM a"
7045        );
7046    }
7047
7048    #[test]
7049    fn test_expand_cte_stars_rewrites_body_with_explicit_cte_columns() {
7050        let mut expr = parse(
7051            "WITH a AS (SELECT 1 AS x), \
7052             b AS (SELECT 2 AS x), \
7053             combined(output_name) AS (SELECT * FROM a UNION ALL SELECT * FROM b) \
7054             SELECT * FROM combined",
7055        );
7056
7057        expand_cte_stars(&mut expr, None);
7058
7059        let sql = crate::generate(&expr, DialectType::Generic).unwrap();
7060        assert!(
7061            sql.contains("SELECT a.x FROM a UNION ALL SELECT b.x FROM b"),
7062            "expected explicit aliases not to suppress body expansion, got {sql}"
7063        );
7064        assert!(
7065            sql.ends_with("SELECT combined.output_name FROM combined"),
7066            "expected the explicit CTE output name to override the body name, got {sql}"
7067        );
7068    }
7069
7070    #[test]
7071    fn test_expand_cte_stars_keeps_recursive_self_reference_conservative() {
7072        let mut expr = parse(
7073            "WITH RECURSIVE r(x) AS (\
7074             SELECT 1 AS x UNION ALL SELECT * FROM r\
7075             ) SELECT * FROM r",
7076        );
7077
7078        expand_cte_stars(&mut expr, None);
7079
7080        let sql = crate::generate(&expr, DialectType::Generic).unwrap();
7081        assert!(
7082            sql.contains("UNION ALL SELECT * FROM r"),
7083            "expected the recursive body star to remain untouched, got {sql}"
7084        );
7085        assert!(
7086            sql.ends_with("SELECT r.x FROM r"),
7087            "expected the explicit recursive CTE column to expand the outer star, got {sql}"
7088        );
7089    }
7090
7091    #[test]
7092    fn test_expand_cte_stars_preserves_genuinely_unresolved_branch_star() {
7093        let mut expr = parse(
7094            "WITH known AS (SELECT 1 AS x), \
7095             combined AS (SELECT * FROM known UNION ALL SELECT * FROM missing) \
7096             SELECT * FROM combined",
7097        );
7098
7099        expand_cte_stars(&mut expr, None);
7100
7101        let sql = crate::generate(&expr, DialectType::Generic).unwrap();
7102        assert!(
7103            sql.contains("SELECT known.x FROM known UNION ALL SELECT * FROM missing"),
7104            "expected only the resolvable branch to expand, got {sql}"
7105        );
7106        assert!(
7107            sql.ends_with("SELECT combined.x FROM combined"),
7108            "expected the leftmost output name to remain usable, got {sql}"
7109        );
7110    }
7111
7112    #[test]
7113    fn test_lineage_cte_star_unknown_table() {
7114        // When CTE references an unknown table, star expansion is skipped gracefully
7115        // and lineage falls back to normal resolution (which may fail)
7116        let expr = parse(
7117            "WITH cte AS (SELECT * FROM unknown_table) \
7118             SELECT * FROM cte",
7119        );
7120        // This should not panic — it may succeed or fail depending on resolution,
7121        // but should not crash
7122        let _result = lineage("x", &expr, None, false);
7123    }
7124
7125    #[test]
7126    fn test_lineage_cte_explicit_columns() {
7127        // CTE with explicit column list: cte(x, y) AS (SELECT a, b FROM t)
7128        let expr = parse(
7129            "WITH cte(x, y) AS (SELECT a, b FROM t) \
7130             SELECT * FROM cte",
7131        );
7132        let node = lineage("x", &expr, None, false).unwrap();
7133        assert_eq!(node.name, "x");
7134    }
7135
7136    #[test]
7137    fn test_lineage_cte_qualified_star() {
7138        // Qualified star: SELECT cte.* FROM cte
7139        let expr = parse(
7140            "WITH cte AS (SELECT a, b FROM t) \
7141             SELECT cte.* FROM cte",
7142        );
7143        for col in &["a", "b"] {
7144            let node = lineage(col, &expr, None, false).unwrap();
7145            assert_eq!(node.name, *col);
7146            let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7147            assert!(
7148                all_names.len() >= 2,
7149                "Expected at least 2 nodes for qualified star column {}, got: {:?}",
7150                col,
7151                all_names
7152            );
7153        }
7154    }
7155
7156    #[test]
7157    fn test_lineage_subquery_select_star() {
7158        // Ported from sqlglot: test_select_star
7159        // SELECT x FROM (SELECT * FROM table_a)
7160        let expr = parse("SELECT x FROM (SELECT * FROM table_a)");
7161        let node = lineage("x", &expr, None, false).unwrap();
7162
7163        assert_eq!(node.name, "x");
7164        assert!(
7165            !node.downstream.is_empty(),
7166            "Expected downstream nodes for subquery with SELECT *, got none"
7167        );
7168    }
7169
7170    #[test]
7171    fn test_lineage_cte_star_with_schema_external_table() {
7172        // CTE references an external table via SELECT * — schema enables expansion
7173        let sql = r#"WITH orders AS (SELECT * FROM stg_orders)
7174SELECT * FROM orders"#;
7175        let expr = parse(sql);
7176
7177        let mut schema = MappingSchema::new();
7178        let cols = vec![
7179            ("order_id".to_string(), DataType::Unknown),
7180            ("customer_id".to_string(), DataType::Unknown),
7181            ("amount".to_string(), DataType::Unknown),
7182        ];
7183        schema.add_table("stg_orders", &cols, None).unwrap();
7184
7185        let node =
7186            lineage_with_schema("order_id", &expr, Some(&schema as &dyn Schema), None, false)
7187                .unwrap();
7188        assert_eq!(node.name, "order_id");
7189    }
7190
7191    #[test]
7192    fn test_lineage_cte_star_with_schema_three_part_name() {
7193        // CTE references an external table with fully-qualified 3-part name
7194        let sql = r#"WITH orders AS (SELECT * FROM "db"."schema"."stg_orders")
7195SELECT * FROM orders"#;
7196        let expr = parse(sql);
7197
7198        let mut schema = MappingSchema::new();
7199        let cols = vec![
7200            ("order_id".to_string(), DataType::Unknown),
7201            ("customer_id".to_string(), DataType::Unknown),
7202        ];
7203        schema
7204            .add_table("db.schema.stg_orders", &cols, None)
7205            .unwrap();
7206
7207        let node = lineage_with_schema(
7208            "customer_id",
7209            &expr,
7210            Some(&schema as &dyn Schema),
7211            None,
7212            false,
7213        )
7214        .unwrap();
7215        assert_eq!(node.name, "customer_id");
7216    }
7217
7218    #[test]
7219    fn test_lineage_cte_star_with_schema_nested() {
7220        // Nested CTEs: outer CTE references inner CTE with SELECT *,
7221        // inner CTE references external table with SELECT *
7222        let sql = r#"WITH
7223            raw AS (SELECT * FROM external_table),
7224            enriched AS (SELECT * FROM raw)
7225        SELECT * FROM enriched"#;
7226        let expr = parse(sql);
7227
7228        let mut schema = MappingSchema::new();
7229        let cols = vec![
7230            ("id".to_string(), DataType::Unknown),
7231            ("name".to_string(), DataType::Unknown),
7232        ];
7233        schema.add_table("external_table", &cols, None).unwrap();
7234
7235        let node =
7236            lineage_with_schema("name", &expr, Some(&schema as &dyn Schema), None, false).unwrap();
7237        assert_eq!(node.name, "name");
7238    }
7239
7240    #[test]
7241    fn test_lineage_cte_qualified_star_with_schema() {
7242        // CTE uses qualified star (orders.*) from a CTE whose columns
7243        // come from an external table via SELECT *
7244        let sql = r#"WITH
7245            orders AS (SELECT * FROM stg_orders),
7246            enriched AS (
7247                SELECT orders.*, 'extra' AS extra
7248                FROM orders
7249            )
7250        SELECT * FROM enriched"#;
7251        let expr = parse(sql);
7252
7253        let mut schema = MappingSchema::new();
7254        let cols = vec![
7255            ("order_id".to_string(), DataType::Unknown),
7256            ("total".to_string(), DataType::Unknown),
7257        ];
7258        schema.add_table("stg_orders", &cols, None).unwrap();
7259
7260        let node =
7261            lineage_with_schema("order_id", &expr, Some(&schema as &dyn Schema), None, false)
7262                .unwrap();
7263        assert_eq!(node.name, "order_id");
7264
7265        // Also verify the extra column works
7266        let extra =
7267            lineage_with_schema("extra", &expr, Some(&schema as &dyn Schema), None, false).unwrap();
7268        assert_eq!(extra.name, "extra");
7269    }
7270
7271    #[test]
7272    fn test_lineage_cte_star_without_schema_still_works() {
7273        // Without schema, CTE-to-CTE star expansion still works
7274        let sql = r#"WITH
7275            cte1 AS (SELECT id, name FROM raw_table),
7276            cte2 AS (SELECT * FROM cte1)
7277        SELECT * FROM cte2"#;
7278        let expr = parse(sql);
7279
7280        // No schema — should still resolve through CTE chain
7281        let node = lineage("id", &expr, None, false).unwrap();
7282        assert_eq!(node.name, "id");
7283    }
7284
7285    #[test]
7286    fn test_lineage_nested_cte_star_with_join_and_schema() {
7287        // Reproduces dbt pattern: CTE chain with qualified star and JOIN
7288        // base_orders -> with_payments (JOIN) -> final -> outer SELECT
7289        let sql = r#"WITH
7290base_orders AS (
7291    SELECT * FROM stg_orders
7292),
7293with_payments AS (
7294    SELECT
7295        base_orders.*,
7296        p.amount
7297    FROM base_orders
7298    LEFT JOIN stg_payments p ON base_orders.order_id = p.order_id
7299),
7300final_cte AS (
7301    SELECT * FROM with_payments
7302)
7303SELECT * FROM final_cte"#;
7304        let expr = parse(sql);
7305
7306        let mut schema = MappingSchema::new();
7307        let order_cols = vec![
7308            (
7309                "order_id".to_string(),
7310                crate::expressions::DataType::Unknown,
7311            ),
7312            (
7313                "customer_id".to_string(),
7314                crate::expressions::DataType::Unknown,
7315            ),
7316            ("status".to_string(), crate::expressions::DataType::Unknown),
7317        ];
7318        let pay_cols = vec![
7319            (
7320                "payment_id".to_string(),
7321                crate::expressions::DataType::Unknown,
7322            ),
7323            (
7324                "order_id".to_string(),
7325                crate::expressions::DataType::Unknown,
7326            ),
7327            ("amount".to_string(), crate::expressions::DataType::Unknown),
7328        ];
7329        schema.add_table("stg_orders", &order_cols, None).unwrap();
7330        schema.add_table("stg_payments", &pay_cols, None).unwrap();
7331
7332        // order_id should trace back to stg_orders
7333        let node =
7334            lineage_with_schema("order_id", &expr, Some(&schema as &dyn Schema), None, false)
7335                .unwrap();
7336        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7337
7338        // The leaf should be "stg_orders.order_id" (not just "order_id")
7339        let has_table_qualified = all_names
7340            .iter()
7341            .any(|n| n.contains('.') && n.contains("order_id"));
7342        assert!(
7343            has_table_qualified,
7344            "Expected table-qualified leaf like 'stg_orders.order_id', got: {:?}",
7345            all_names
7346        );
7347
7348        // amount should trace back to stg_payments
7349        let node = lineage_with_schema("amount", &expr, Some(&schema as &dyn Schema), None, false)
7350            .unwrap();
7351        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7352
7353        let has_table_qualified = all_names
7354            .iter()
7355            .any(|n| n.contains('.') && n.contains("amount"));
7356        assert!(
7357            has_table_qualified,
7358            "Expected table-qualified leaf like 'stg_payments.amount', got: {:?}",
7359            all_names
7360        );
7361    }
7362
7363    #[test]
7364    fn test_lineage_cte_alias_resolution() {
7365        // FROM cte_name AS alias pattern: alias should resolve through CTE to source table
7366        let sql = r#"WITH import_stg_items AS (
7367    SELECT item_id, name, status FROM stg_items
7368)
7369SELECT base.item_id, base.status
7370FROM import_stg_items AS base"#;
7371        let expr = parse(sql);
7372
7373        let node = lineage("item_id", &expr, None, false).unwrap();
7374        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7375        // Should trace through alias "base" → CTE "import_stg_items" → "stg_items.item_id"
7376        assert!(
7377            all_names.iter().any(|n| n == "stg_items.item_id"),
7378            "Expected leaf 'stg_items.item_id', got: {:?}",
7379            all_names
7380        );
7381    }
7382
7383    #[test]
7384    fn test_lineage_cte_alias_with_schema_and_star() {
7385        // CTE alias + SELECT * expansion: FROM cte AS alias with star in CTE body
7386        let sql = r#"WITH import_stg AS (
7387    SELECT * FROM stg_items
7388)
7389SELECT base.item_id, base.status
7390FROM import_stg AS base"#;
7391        let expr = parse(sql);
7392
7393        let mut schema = MappingSchema::new();
7394        schema
7395            .add_table(
7396                "stg_items",
7397                &[
7398                    ("item_id".to_string(), DataType::Unknown),
7399                    ("name".to_string(), DataType::Unknown),
7400                    ("status".to_string(), DataType::Unknown),
7401                ],
7402                None,
7403            )
7404            .unwrap();
7405
7406        let node = lineage_with_schema("item_id", &expr, Some(&schema as &dyn Schema), None, false)
7407            .unwrap();
7408        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7409        assert!(
7410            all_names.iter().any(|n| n == "stg_items.item_id"),
7411            "Expected leaf 'stg_items.item_id', got: {:?}",
7412            all_names
7413        );
7414    }
7415
7416    #[test]
7417    fn test_lineage_cte_alias_with_join() {
7418        // Multiple CTE aliases in a JOIN: each should resolve independently
7419        let sql = r#"WITH
7420    import_users AS (SELECT id, name FROM users),
7421    import_orders AS (SELECT id, user_id, amount FROM orders)
7422SELECT u.name, o.amount
7423FROM import_users AS u
7424LEFT JOIN import_orders AS o ON u.id = o.user_id"#;
7425        let expr = parse(sql);
7426
7427        let node = lineage("name", &expr, None, false).unwrap();
7428        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7429        assert!(
7430            all_names.iter().any(|n| n == "users.name"),
7431            "Expected leaf 'users.name', got: {:?}",
7432            all_names
7433        );
7434
7435        let node = lineage("amount", &expr, None, false).unwrap();
7436        let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
7437        assert!(
7438            all_names.iter().any(|n| n == "orders.amount"),
7439            "Expected leaf 'orders.amount', got: {:?}",
7440            all_names
7441        );
7442    }
7443
7444    // -----------------------------------------------------------------------
7445    // Quoted CTE name tests — verifying SQL identifier case semantics
7446    // -----------------------------------------------------------------------
7447
7448    #[test]
7449    fn test_lineage_unquoted_cte_case_insensitive() {
7450        // Unquoted CTE names are case-insensitive (both normalized to lowercase).
7451        // MyCte and MYCTE should match.
7452        let expr = parse("WITH MyCte AS (SELECT id AS col FROM source) SELECT * FROM MYCTE");
7453        let node = lineage("col", &expr, None, false).unwrap();
7454        assert_eq!(node.name, "col");
7455        assert!(
7456            !node.downstream.is_empty(),
7457            "Unquoted CTE should resolve case-insensitively"
7458        );
7459    }
7460
7461    #[test]
7462    fn test_lineage_quoted_cte_case_preserved() {
7463        // Quoted CTE name preserves case. "MyCte" referenced as "MyCte" should match.
7464        let expr = parse(r#"WITH "MyCte" AS (SELECT id AS col FROM source) SELECT * FROM "MyCte""#);
7465        let node = lineage("col", &expr, None, false).unwrap();
7466        assert_eq!(node.name, "col");
7467        assert!(
7468            !node.downstream.is_empty(),
7469            "Quoted CTE with matching case should resolve"
7470        );
7471    }
7472
7473    #[test]
7474    fn test_lineage_quoted_cte_case_mismatch_no_expansion() {
7475        // Quoted CTE "MyCte" referenced as "mycte" — case mismatch.
7476        // sqlglot treats this as a table reference, not a CTE match.
7477        // Star expansion should NOT resolve through the CTE.
7478        let expr = parse(r#"WITH "MyCte" AS (SELECT id AS col FROM source) SELECT * FROM "mycte""#);
7479        // lineage("col", ...) should fail because "mycte" is treated as an external
7480        // table (not matching CTE "MyCte"), and SELECT * cannot be expanded.
7481        let result = lineage("col", &expr, None, false);
7482        assert!(
7483            result.is_err(),
7484            "Quoted CTE with case mismatch should not expand star: {:?}",
7485            result
7486        );
7487    }
7488
7489    #[test]
7490    fn test_lineage_mixed_quoted_unquoted_cte() {
7491        // Mix of unquoted and quoted CTEs in a nested chain.
7492        let expr = parse(
7493            r#"WITH unquoted AS (SELECT 1 AS a FROM t), "Quoted" AS (SELECT a FROM unquoted) SELECT * FROM "Quoted""#,
7494        );
7495        let node = lineage("a", &expr, None, false).unwrap();
7496        assert_eq!(node.name, "a");
7497        assert!(
7498            !node.downstream.is_empty(),
7499            "Mixed quoted/unquoted CTE chain should resolve"
7500        );
7501    }
7502
7503    // -----------------------------------------------------------------------
7504    // Known bugs: quoted CTE case sensitivity in scope/lineage tracing paths
7505    // -----------------------------------------------------------------------
7506    //
7507    // expand_cte_stars correctly handles quoted vs unquoted CTE names via
7508    // normalize_cte_name(). However, the scope system (scope.rs add_table_to_scope)
7509    // and the lineage tracing path (to_node_inner) use eq_ignore_ascii_case or
7510    // direct string comparison for CTE name matching, ignoring the quoted status.
7511    //
7512    // sqlglot's normalize_identifiers treats quoted identifiers as case-sensitive
7513    // and unquoted as case-insensitive. The scope system should do the same.
7514    //
7515    // Fixing these requires changes across scope.rs and lineage.rs CTE resolution,
7516    // which is broader than the star expansion scope of this PR.
7517
7518    #[test]
7519    fn test_lineage_quoted_cte_case_mismatch_non_star_known_bug() {
7520        // Known bug: scope.rs add_table_to_scope uses eq_ignore_ascii_case for
7521        // all identifiers including quoted ones, so quoted CTE "MyCte" referenced
7522        // as "mycte" incorrectly resolves to the CTE.
7523        //
7524        // Per SQL semantics (and sqlglot behavior), quoted identifiers are
7525        // case-sensitive: "mycte" should NOT match CTE "MyCte".
7526        //
7527        // This test asserts the CURRENT BUGGY behavior. When the bug is fixed,
7528        // this test should fail — update the assertion to match correct behavior:
7529        //   child.source_name should be "" (table ref), not "MyCte" (CTE ref).
7530        let expr = parse(r#"WITH "MyCte" AS (SELECT 1 AS col) SELECT col FROM "mycte""#);
7531        let node = lineage("col", &expr, None, false).unwrap();
7532        assert!(!node.downstream.is_empty());
7533        let child = &node.downstream[0];
7534        // BUG: "mycte" incorrectly resolves to CTE "MyCte"
7535        assert_eq!(
7536            child.source_name, "MyCte",
7537            "Known bug: quoted CTE case mismatch should NOT resolve, but currently does. \
7538             If this fails, the bug may be fixed — update to assert source_name != \"MyCte\""
7539        );
7540    }
7541
7542    #[test]
7543    fn test_lineage_quoted_cte_case_mismatch_qualified_col_known_bug() {
7544        // Known bug: same as above but with qualified column reference ("mycte".col).
7545        // scope.rs resolves "mycte" to CTE "MyCte" case-insensitively even for
7546        // quoted identifiers, so "mycte".col incorrectly traces through CTE "MyCte".
7547        //
7548        // This test asserts the CURRENT BUGGY behavior. When the bug is fixed,
7549        // this test should fail — update to assert source_name != "MyCte".
7550        let expr = parse(r#"WITH "MyCte" AS (SELECT 1 AS col) SELECT "mycte".col FROM "mycte""#);
7551        let node = lineage("col", &expr, None, false).unwrap();
7552        assert!(!node.downstream.is_empty());
7553        let child = &node.downstream[0];
7554        // BUG: "mycte".col incorrectly resolves through CTE "MyCte"
7555        assert_eq!(
7556            child.source_name, "MyCte",
7557            "Known bug: quoted CTE case mismatch should NOT resolve, but currently does. \
7558             If this fails, the bug may be fixed — update to assert source_name != \"MyCte\""
7559        );
7560    }
7561
7562    #[test]
7563    fn test_lineage_recursive_cte_terminates_at_base_case() {
7564        let expr = parse_dialect(
7565            "WITH RECURSIVE nums AS (\
7566             SELECT 1 AS n \
7567             UNION ALL \
7568             SELECT n + 1 FROM nums WHERE n < 5\
7569             ) SELECT n FROM nums",
7570            DialectType::DuckDB,
7571        );
7572        let node = lineage("n", &expr, Some(DialectType::DuckDB), false).unwrap();
7573        let names = lineage_names(&node);
7574
7575        assert!(
7576            names.len() <= 12,
7577            "recursive CTE lineage should not unroll repeatedly, got {names:?}"
7578        );
7579        assert!(
7580            node.walk()
7581                .any(|child| child.source_kind == SourceKind::Cte && child.source_name == "nums"),
7582            "expected recursive source to be marked as a CTE, got {names:?}"
7583        );
7584    }
7585
7586    #[test]
7587    fn test_lineage_window_partition_and_order_columns() {
7588        let expr = parse(
7589            "WITH c AS (SELECT user_id, ts FROM events) \
7590             SELECT ROW_NUMBER() OVER (PARTITION BY c.user_id ORDER BY c.ts) AS out FROM c",
7591        );
7592        let node = lineage("out", &expr, None, false).unwrap();
7593
7594        assert_lineage_contains(&node, "events.user_id");
7595        assert_lineage_contains(&node, "events.ts");
7596    }
7597
7598    #[test]
7599    fn test_lineage_window_aggregate_order_column() {
7600        let expr = parse(
7601            "WITH c AS (SELECT amount, d FROM txns) \
7602             SELECT SUM(c.amount) OVER (ORDER BY c.d) AS running FROM c",
7603        );
7604        let node = lineage("running", &expr, None, false).unwrap();
7605
7606        assert_lineage_contains(&node, "txns.amount");
7607        assert_lineage_contains(&node, "txns.d");
7608    }
7609
7610    #[test]
7611    fn test_lineage_named_window_columns() {
7612        let expr = parse(
7613            "SELECT ROW_NUMBER() OVER w AS out \
7614             FROM events \
7615             WINDOW w AS (PARTITION BY user_id ORDER BY ts)",
7616        );
7617        let node = lineage("out", &expr, None, false).unwrap();
7618
7619        assert_lineage_contains(&node, "events.user_id");
7620        assert_lineage_contains(&node, "events.ts");
7621    }
7622
7623    #[test]
7624    fn test_lineage_within_group_order_column() {
7625        let expr =
7626            parse("SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY amount) AS p FROM txns");
7627        let node = lineage("p", &expr, None, false).unwrap();
7628
7629        assert_lineage_contains(&node, "txns.amount");
7630    }
7631
7632    #[test]
7633    fn test_lineage_query_wrappers_resolve_inner_select() {
7634        for sql in [
7635            "CREATE TABLE tgt AS SELECT x FROM src",
7636            "CREATE VIEW v AS SELECT x FROM src",
7637            "INSERT INTO tgt SELECT x FROM src",
7638        ] {
7639            let expr = parse(sql);
7640            let node = lineage("x", &expr, None, false).unwrap();
7641            assert_lineage_contains(&node, "src.x");
7642        }
7643    }
7644
7645    #[test]
7646    fn test_lineage_scalar_subquery_through_cte_reaches_base_table() {
7647        let expr = parse(
7648            "WITH c AS (SELECT x FROM t) \
7649             SELECT (SELECT SUM(x) FROM c) AS s FROM c LIMIT 1",
7650        );
7651        let node = lineage("s", &expr, None, false).unwrap();
7652
7653        assert_lineage_contains(&node, "t.x");
7654        assert!(
7655            node.walk()
7656                .any(|child| child.source_kind == SourceKind::Cte && child.source_name == "c"),
7657            "expected scalar subquery CTE hop in lineage, got {:?}",
7658            lineage_names(&node)
7659        );
7660    }
7661
7662    #[test]
7663    fn test_lineage_scalar_subqueries_inside_expression_wrappers() {
7664        for sql in [
7665            "WITH c AS (SELECT a, b FROM t) \
7666             SELECT CASE WHEN c.a > 0 THEN c.b ELSE (SELECT MAX(z) FROM o) END AS r FROM c",
7667            "WITH c AS (SELECT a FROM t) \
7668             SELECT COALESCE(c.a, (SELECT MAX(z) FROM o)) AS r FROM c",
7669            "WITH c AS (SELECT a FROM t) \
7670             SELECT CAST((SELECT MAX(z) FROM o) AS INT) + c.a AS r FROM c",
7671            "WITH c AS (SELECT a FROM t) \
7672             SELECT CASE WHEN c.a BETWEEN 0 AND (SELECT MAX(z) FROM o) THEN c.a END AS r FROM c",
7673        ] {
7674            let expr = parse_dialect(sql, DialectType::DuckDB);
7675            let node = lineage("r", &expr, Some(DialectType::DuckDB), false)
7676                .unwrap_or_else(|error| panic!("lineage failed for {sql}: {error}"));
7677
7678            assert_lineage_contains(&node, "o.z");
7679            assert_lineage_contains(&node, "t.a");
7680        }
7681    }
7682
7683    #[test]
7684    fn test_lineage_nested_set_operation_inside_derived_table() {
7685        let expr = parse_dialect(
7686            "SELECT v FROM ((SELECT v FROM t1 UNION ALL SELECT v FROM t2) \
7687             UNION ALL SELECT v FROM t3) u",
7688            DialectType::DuckDB,
7689        );
7690        let node = lineage("v", &expr, Some(DialectType::DuckDB), false).unwrap();
7691
7692        assert_lineage_contains(&node, "t1.v");
7693        assert_lineage_contains(&node, "t2.v");
7694        assert_lineage_contains(&node, "t3.v");
7695    }
7696
7697    #[test]
7698    fn test_lineage_select_alias_reference_resolves_to_alias_source() {
7699        let expr = parse_dialect(
7700            "WITH c AS (SELECT x FROM t) SELECT c.x AS a, a + 1 AS b FROM c",
7701            DialectType::DuckDB,
7702        );
7703        let node = lineage("b", &expr, Some(DialectType::DuckDB), false).unwrap();
7704
7705        assert_lineage_contains(&node, "t.x");
7706    }
7707
7708    #[test]
7709    fn test_lineage_pivot_output_resolves_aggregation_input() {
7710        let expr = parse_dialect(
7711            "SELECT * FROM (SELECT region, q, amt FROM sales) \
7712             PIVOT(SUM(amt) FOR q IN ('Q1' AS q1))",
7713            DialectType::DuckDB,
7714        );
7715        let node = lineage("q1", &expr, Some(DialectType::DuckDB), false).unwrap();
7716
7717        assert_lineage_contains(&node, "sales.amt");
7718    }
7719
7720    #[test]
7721    fn test_lineage_pivot_multi_aggregate_and_alias_columns() {
7722        let multi = parse_dialect(
7723            "SELECT * FROM (SELECT category, value, price FROM t) \
7724             PIVOT(SUM(value) AS value_sum, MAX(price) FOR category IN ('a' AS cat_a, 'b'))",
7725            DialectType::DuckDB,
7726        );
7727        let value_sum =
7728            lineage("cat_a_value_sum", &multi, Some(DialectType::DuckDB), false).unwrap();
7729        assert_lineage_contains(&value_sum, "t.value");
7730
7731        let max_price =
7732            lineage("cat_a_max(price)", &multi, Some(DialectType::DuckDB), false).unwrap();
7733        assert_lineage_contains(&max_price, "t.price");
7734
7735        let aliased = parse_dialect(
7736            "SELECT * FROM (SELECT region, q, amt FROM sales) \
7737             PIVOT(SUM(amt) FOR q IN ('Q1')) AS p(region2, p1)",
7738            DialectType::DuckDB,
7739        );
7740        let region = lineage("region2", &aliased, Some(DialectType::DuckDB), false).unwrap();
7741        assert_lineage_contains(&region, "sales.region");
7742
7743        let pivot_value = lineage("p1", &aliased, Some(DialectType::DuckDB), false).unwrap();
7744        assert_lineage_contains(&pivot_value, "sales.amt");
7745    }
7746
7747    #[test]
7748    fn test_lineage_pivot_through_cte_resolves_aggregation_input() {
7749        let expr = parse_dialect(
7750            "WITH src AS (SELECT region, q, amt FROM sales) \
7751             SELECT q1 FROM src PIVOT(SUM(amt) FOR q IN ('Q1' AS q1))",
7752            DialectType::DuckDB,
7753        );
7754        let node = lineage("q1", &expr, Some(DialectType::DuckDB), false).unwrap();
7755
7756        assert_lineage_contains(&node, "sales.amt");
7757    }
7758
7759    #[test]
7760    fn test_lineage_pivot_cte_input_resolution_470() {
7761        for source in ["src", "src AS i"] {
7762            for values in ["'books' AS books", "ANY ORDER BY category"] {
7763                let sql = format!("WITH src AS (SELECT customer_id, category, amount FROM staged_orders) SELECT customer_id FROM {source} PIVOT(MAX(amount) FOR category IN ({values}))");
7764                let expr = parse_dialect(&sql, DialectType::Snowflake);
7765                let node =
7766                    lineage("customer_id", &expr, Some(DialectType::Snowflake), false).unwrap();
7767                assert_lineage_contains(&node, "staged_orders.customer_id");
7768            }
7769        }
7770        for (column, relation, expected) in [
7771            (
7772                "books",
7773                "src PIVOT(MAX(amount) FOR category IN ('books' AS books))",
7774                "staged_orders.amount",
7775            ),
7776            (
7777                "value",
7778                "src UNPIVOT(value FOR measure IN (customer_id, amount))",
7779                "staged_orders.amount",
7780            ),
7781            (
7782                "category",
7783                "public.src PIVOT(MAX(amount) FOR customer_id IN (1))",
7784                "src.category",
7785            ),
7786        ] {
7787            let expr = parse_dialect(&format!("WITH src AS (SELECT customer_id, category, amount FROM staged_orders) SELECT {column} FROM {relation}"), DialectType::Snowflake);
7788            let node = lineage(column, &expr, Some(DialectType::Snowflake), false).unwrap();
7789            assert_lineage_contains(&node, expected);
7790            if relation.starts_with("public.") {
7791                assert!(!lineage_names(&node)
7792                    .iter()
7793                    .any(|name| name.starts_with("staged_orders.")));
7794            }
7795        }
7796    }
7797
7798    #[test]
7799    fn test_lineage_unpivot_value_resolves_input_columns() {
7800        let expr = parse_dialect(
7801            "SELECT name, val FROM t UNPIVOT(val FOR col IN (a, b, c))",
7802            DialectType::DuckDB,
7803        );
7804        let node = lineage("val", &expr, Some(DialectType::DuckDB), false).unwrap();
7805
7806        assert_lineage_contains(&node, "t.a");
7807        assert_lineage_contains(&node, "t.b");
7808        assert_lineage_contains(&node, "t.c");
7809    }
7810
7811    #[test]
7812    fn test_lineage_unpivot_multi_value_columns_resolve_positionally() {
7813        let expr = parse_dialect(
7814            "SELECT first_half_sales, second_half_sales, semester \
7815             FROM produce \
7816             UNPIVOT((first_half_sales, second_half_sales) \
7817             FOR semester IN ((q1, q2) AS 'semester_1', (q3, q4) AS 'semester_2'))",
7818            DialectType::BigQuery,
7819        );
7820
7821        let first = lineage(
7822            "first_half_sales",
7823            &expr,
7824            Some(DialectType::BigQuery),
7825            false,
7826        )
7827        .unwrap();
7828        assert_lineage_contains(&first, "produce.q1");
7829        assert_lineage_contains(&first, "produce.q3");
7830
7831        let second = lineage(
7832            "second_half_sales",
7833            &expr,
7834            Some(DialectType::BigQuery),
7835            false,
7836        )
7837        .unwrap();
7838        assert_lineage_contains(&second, "produce.q2");
7839        assert_lineage_contains(&second, "produce.q4");
7840    }
7841
7842    #[test]
7843    fn test_lineage_pivot_output_reads_aggregation_and_field() {
7844        let expr = parse_dialect(
7845            "SELECT order_id, card FROM (SELECT order_id, method, amount FROM payments) \
7846             PIVOT (SUM(amount) FOR method IN ('card', 'coupon')) AS p",
7847            DialectType::Snowflake,
7848        );
7849        let node = lineage("card", &expr, Some(DialectType::Snowflake), false).unwrap();
7850        assert_lineage_contains(&node, "payments.amount");
7851        assert_lineage_contains(&node, "payments.method");
7852        assert!(node
7853            .walk()
7854            .any(|step| crate::traversal::contains_aggregate(&step.expression)));
7855
7856        let node = lineage("order_id", &expr, Some(DialectType::Snowflake), false).unwrap();
7857        assert_eq!(
7858            lineage_names(&node)
7859                .iter()
7860                .filter(|name| name.starts_with("payments."))
7861                .collect::<Vec<_>>(),
7862            ["payments.order_id"]
7863        );
7864    }
7865
7866    #[test]
7867    fn test_lineage_unpivot_passthrough_resolves_input_column() {
7868        let expr = parse_dialect(
7869            "SELECT customer_id FROM (SELECT customer_id, first_name, last_name FROM customers) \
7870             UNPIVOT (name_value FOR name_kind IN (first_name, last_name))",
7871            DialectType::Snowflake,
7872        );
7873        let node = lineage("customer_id", &expr, Some(DialectType::Snowflake), false).unwrap();
7874        assert_lineage_contains(&node, "customers.customer_id");
7875    }
7876
7877    #[test]
7878    fn test_lineage_top_level_union_over_ctes_reaches_base_tables() {
7879        let expr = parse(
7880            "WITH a AS (SELECT x FROM t1), b AS (SELECT x FROM t2) \
7881             SELECT x FROM a UNION SELECT x FROM b",
7882        );
7883        let node = lineage("x", &expr, None, false).unwrap();
7884
7885        assert_lineage_contains(&node, "t1.x");
7886        assert_lineage_contains(&node, "t2.x");
7887        for cte_name in ["a", "b"] {
7888            assert!(
7889                node.walk().any(|child| child.source_name == cte_name),
7890                "expected set-operation lineage to retain CTE source {cte_name}: {:?}",
7891                lineage_names(&node)
7892            );
7893        }
7894    }
7895
7896    #[test]
7897    fn test_lineage_star_excludes_semi_join_rhs_source() {
7898        let expr = parse_dialect(
7899            "SELECT * FROM orders LEFT SEMI JOIN customers ON orders.customer_id = customers.id",
7900            DialectType::DuckDB,
7901        );
7902        let node = lineage("customer_id", &expr, Some(DialectType::DuckDB), false).unwrap();
7903
7904        assert_lineage_contains(&node, "orders.customer_id");
7905    }
7906
7907    // --- Comment handling tests (ported from sqlglot test_lineage.py) ---
7908
7909    /// sqlglot: test_node_name_doesnt_contain_comment
7910    /// Comments in column expressions should not affect lineage resolution.
7911    /// NOTE: This test uses SELECT * from a derived table, which is a separate
7912    /// known limitation in polyglot-sql (star expansion in subqueries).
7913    #[test]
7914    #[ignore = "requires derived table star expansion (separate issue)"]
7915    fn test_node_name_doesnt_contain_comment() {
7916        let expr = parse("SELECT * FROM (SELECT x /* c */ FROM t1) AS t2");
7917        let node = lineage("x", &expr, None, false).unwrap();
7918
7919        assert_eq!(node.name, "x");
7920        assert!(!node.downstream.is_empty());
7921    }
7922
7923    /// A line comment between SELECT and the first column wraps the column
7924    /// in an Annotated node. Lineage must unwrap it to find the column name.
7925    /// Verify that commented and uncommented queries produce identical lineage.
7926    #[test]
7927    fn test_comment_before_first_column_in_cte() {
7928        let sql_with_comment = "with t as (select 1 as a) select\n  -- comment\n  a from t";
7929        let sql_without_comment = "with t as (select 1 as a) select a from t";
7930
7931        // Without comment — baseline
7932        let expr_ok = parse(sql_without_comment);
7933        let node_ok = lineage("a", &expr_ok, None, false).expect("without comment should succeed");
7934
7935        // With comment — should produce identical lineage
7936        let expr_comment = parse(sql_with_comment);
7937        let node_comment = lineage("a", &expr_comment, None, false)
7938            .expect("with comment before first column should succeed");
7939
7940        assert_eq!(node_ok.name, node_comment.name, "node names should match");
7941        assert_eq!(
7942            node_ok.downstream_names(),
7943            node_comment.downstream_names(),
7944            "downstream lineage should be identical with or without comment"
7945        );
7946    }
7947
7948    /// Block comment between SELECT and first column.
7949    #[test]
7950    fn test_block_comment_before_first_column() {
7951        let sql = "with t as (select 1 as a) select /* section */ a from t";
7952        let expr = parse(sql);
7953        let node = lineage("a", &expr, None, false)
7954            .expect("block comment before first column should succeed");
7955        assert_eq!(node.name, "a");
7956        assert!(
7957            !node.downstream.is_empty(),
7958            "should have downstream lineage"
7959        );
7960    }
7961
7962    /// Comment before first column should not affect second column resolution.
7963    #[test]
7964    fn test_comment_before_first_column_second_col_ok() {
7965        let sql = "with t as (select 1 as a, 2 as b) select\n  -- comment\n  a, b from t";
7966        let expr = parse(sql);
7967
7968        let node_a =
7969            lineage("a", &expr, None, false).expect("column a with comment should succeed");
7970        assert_eq!(node_a.name, "a");
7971
7972        let node_b =
7973            lineage("b", &expr, None, false).expect("column b with comment should succeed");
7974        assert_eq!(node_b.name, "b");
7975    }
7976
7977    /// Aliased column with preceding comment.
7978    #[test]
7979    fn test_comment_before_aliased_column() {
7980        let sql = "with t as (select 1 as x) select\n  -- renamed\n  x as y from t";
7981        let expr = parse(sql);
7982        let node =
7983            lineage("y", &expr, None, false).expect("aliased column with comment should succeed");
7984        assert_eq!(node.name, "y");
7985        assert!(
7986            !node.downstream.is_empty(),
7987            "aliased column should have downstream lineage"
7988        );
7989    }
7990}