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