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