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