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