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