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