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