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_lateral_view_columns_resolve_to_virtual_sources() {
5071 let cases = [
5072 (
5073 DialectType::Spark,
5074 "SELECT x AS out FROM t LATERAL VIEW EXPLODE(items) u AS x",
5075 ),
5076 (
5077 DialectType::Hive,
5078 "SELECT x AS out FROM t LATERAL VIEW EXPLODE(items) u AS x",
5079 ),
5080 ];
5081
5082 for (dialect, sql) in cases {
5083 let expr = parse_one(sql, dialect).unwrap_or_else(|e| panic!("parse {dialect:?}: {e}"));
5084 let node = lineage("out", &expr, Some(dialect), false)
5085 .unwrap_or_else(|e| panic!("lineage {dialect:?}: {e}"));
5086 let virtual_child = node
5087 .downstream
5088 .first()
5089 .unwrap_or_else(|| panic!("expected virtual child for {dialect:?}"));
5090
5091 assert_eq!(virtual_child.name, "_0.x");
5092 assert_eq!(virtual_child.source_name, "_0");
5093 assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5094 assert_eq!(virtual_child.source_alias.as_deref(), Some("u"));
5095
5096 let real_child = virtual_child
5097 .downstream
5098 .first()
5099 .unwrap_or_else(|| panic!("expected table dependency for {dialect:?}"));
5100 assert_eq!(real_child.name, "t.items");
5101 assert_eq!(real_child.source_kind, SourceKind::Table);
5102 }
5103 }
5104
5105 #[test]
5106 fn test_lineage_snowflake_lateral_flatten_is_virtual_source() {
5107 let expr = parse_one(
5108 "SELECT f.value AS value FROM raw_events, LATERAL FLATTEN(INPUT => payload:items) AS f",
5109 DialectType::Snowflake,
5110 )
5111 .expect("parse");
5112
5113 let node = lineage("value", &expr, Some(DialectType::Snowflake), false).expect("lineage");
5114 let virtual_child = node.downstream.first().expect("virtual flatten source");
5115 assert_eq!(virtual_child.name, "_0.value");
5116 assert_eq!(virtual_child.source_name, "_0");
5117 assert_eq!(virtual_child.source_kind, SourceKind::Virtual);
5118 assert_eq!(virtual_child.source_alias.as_deref(), Some("f"));
5119
5120 let real_child = virtual_child
5121 .downstream
5122 .first()
5123 .expect("FLATTEN input should depend on raw_events.payload");
5124 assert_eq!(real_child.name, "raw_events.payload");
5125 assert_eq!(real_child.source_kind, SourceKind::Table);
5126 }
5127
5128 #[test]
5129 fn test_lineage_with_schema_snowflake_datediff_date_part_issue_61() {
5130 let expr = parse_one(
5131 "SELECT DATEDIFF(day, date_utc, CURRENT_DATE()) AS recency FROM fact.some_daily_metrics",
5132 DialectType::Snowflake,
5133 )
5134 .expect("parse");
5135
5136 let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5137 schema
5138 .add_table(
5139 "fact.some_daily_metrics",
5140 &[("date_utc".to_string(), DataType::Date)],
5141 None,
5142 )
5143 .expect("schema setup");
5144
5145 let node = lineage_with_schema(
5146 "recency",
5147 &expr,
5148 Some(&schema),
5149 Some(DialectType::Snowflake),
5150 false,
5151 )
5152 .expect("lineage_with_schema should not treat date part as a column");
5153
5154 let names = node.downstream_names();
5155 assert!(
5156 names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5157 "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5158 names
5159 );
5160 assert!(
5161 !names.iter().any(|n| n.ends_with(".day") || n == "day"),
5162 "Did not expect date part to appear as lineage column, got: {:?}",
5163 names
5164 );
5165 }
5166
5167 #[test]
5168 fn test_snowflake_datediff_parses_to_typed_ast() {
5169 let expr = parse_one(
5170 "SELECT DATEDIFF(day, date_utc, CURRENT_DATE()) AS recency FROM fact.some_daily_metrics",
5171 DialectType::Snowflake,
5172 )
5173 .expect("parse");
5174
5175 match expr {
5176 Expression::Select(select) => match &select.expressions[0] {
5177 Expression::Alias(alias) => match &alias.this {
5178 Expression::DateDiff(f) => {
5179 assert_eq!(f.unit, Some(crate::expressions::IntervalUnit::Day));
5180 }
5181 other => panic!("expected DateDiff, got {other:?}"),
5182 },
5183 other => panic!("expected Alias, got {other:?}"),
5184 },
5185 other => panic!("expected Select, got {other:?}"),
5186 }
5187 }
5188
5189 #[test]
5190 fn test_lineage_with_schema_snowflake_dateadd_date_part_issue_followup() {
5191 let expr = parse_one(
5192 "SELECT DATEADD(day, 1, date_utc) AS next_day FROM fact.some_daily_metrics",
5193 DialectType::Snowflake,
5194 )
5195 .expect("parse");
5196
5197 let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5198 schema
5199 .add_table(
5200 "fact.some_daily_metrics",
5201 &[("date_utc".to_string(), DataType::Date)],
5202 None,
5203 )
5204 .expect("schema setup");
5205
5206 let node = lineage_with_schema(
5207 "next_day",
5208 &expr,
5209 Some(&schema),
5210 Some(DialectType::Snowflake),
5211 false,
5212 )
5213 .expect("lineage_with_schema should not treat DATEADD date part as a column");
5214
5215 let names = node.downstream_names();
5216 assert!(
5217 names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5218 "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5219 names
5220 );
5221 assert!(
5222 !names.iter().any(|n| n.ends_with(".day") || n == "day"),
5223 "Did not expect date part to appear as lineage column, got: {:?}",
5224 names
5225 );
5226 }
5227
5228 #[test]
5229 fn test_lineage_with_schema_snowflake_date_part_identifier_issue_followup() {
5230 let expr = parse_one(
5231 "SELECT DATE_PART(day, date_utc) AS day_part FROM fact.some_daily_metrics",
5232 DialectType::Snowflake,
5233 )
5234 .expect("parse");
5235
5236 let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5237 schema
5238 .add_table(
5239 "fact.some_daily_metrics",
5240 &[("date_utc".to_string(), DataType::Date)],
5241 None,
5242 )
5243 .expect("schema setup");
5244
5245 let node = lineage_with_schema(
5246 "day_part",
5247 &expr,
5248 Some(&schema),
5249 Some(DialectType::Snowflake),
5250 false,
5251 )
5252 .expect("lineage_with_schema should not treat DATE_PART identifier as a column");
5253
5254 let names = node.downstream_names();
5255 assert!(
5256 names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5257 "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5258 names
5259 );
5260 assert!(
5261 !names.iter().any(|n| n.ends_with(".day") || n == "day"),
5262 "Did not expect date part to appear as lineage column, got: {:?}",
5263 names
5264 );
5265 }
5266
5267 #[test]
5268 fn test_lineage_with_schema_snowflake_date_part_string_literal_control() {
5269 let expr = parse_one(
5270 "SELECT DATE_PART('day', date_utc) AS day_part FROM fact.some_daily_metrics",
5271 DialectType::Snowflake,
5272 )
5273 .expect("parse");
5274
5275 let mut schema = MappingSchema::with_dialect(DialectType::Snowflake);
5276 schema
5277 .add_table(
5278 "fact.some_daily_metrics",
5279 &[("date_utc".to_string(), DataType::Date)],
5280 None,
5281 )
5282 .expect("schema setup");
5283
5284 let node = lineage_with_schema(
5285 "day_part",
5286 &expr,
5287 Some(&schema),
5288 Some(DialectType::Snowflake),
5289 false,
5290 )
5291 .expect("quoted DATE_PART should continue to work");
5292
5293 let names = node.downstream_names();
5294 assert!(
5295 names.iter().any(|n| n == "some_daily_metrics.date_utc"),
5296 "Expected some_daily_metrics.date_utc in downstream, got: {:?}",
5297 names
5298 );
5299 }
5300
5301 #[test]
5302 fn test_snowflake_dateadd_date_part_identifier_stays_generic_function() {
5303 let expr = parse_one(
5304 "SELECT DATEADD(day, 1, date_utc) AS next_day FROM fact.some_daily_metrics",
5305 DialectType::Snowflake,
5306 )
5307 .expect("parse");
5308
5309 match expr {
5310 Expression::Select(select) => match &select.expressions[0] {
5311 Expression::Alias(alias) => match &alias.this {
5312 Expression::Function(f) => {
5313 assert_eq!(f.name.to_uppercase(), "DATEADD");
5314 assert!(matches!(&f.args[0], Expression::Var(v) if v.this == "day"));
5315 }
5316 other => panic!("expected generic DATEADD function, got {other:?}"),
5317 },
5318 other => panic!("expected Alias, got {other:?}"),
5319 },
5320 other => panic!("expected Select, got {other:?}"),
5321 }
5322 }
5323
5324 #[test]
5325 fn test_snowflake_date_part_identifier_stays_generic_function_with_var_arg() {
5326 let expr = parse_one(
5327 "SELECT DATE_PART(day, date_utc) AS day_part FROM fact.some_daily_metrics",
5328 DialectType::Snowflake,
5329 )
5330 .expect("parse");
5331
5332 match expr {
5333 Expression::Select(select) => match &select.expressions[0] {
5334 Expression::Alias(alias) => match &alias.this {
5335 Expression::Function(f) => {
5336 assert_eq!(f.name.to_uppercase(), "DATE_PART");
5337 assert!(matches!(&f.args[0], Expression::Var(v) if v.this == "day"));
5338 }
5339 other => panic!("expected generic DATE_PART function, got {other:?}"),
5340 },
5341 other => panic!("expected Alias, got {other:?}"),
5342 },
5343 other => panic!("expected Select, got {other:?}"),
5344 }
5345 }
5346
5347 #[test]
5348 fn test_snowflake_date_part_string_literal_stays_generic_function() {
5349 let expr = parse_one(
5350 "SELECT DATE_PART('day', date_utc) AS day_part FROM fact.some_daily_metrics",
5351 DialectType::Snowflake,
5352 )
5353 .expect("parse");
5354
5355 match expr {
5356 Expression::Select(select) => match &select.expressions[0] {
5357 Expression::Alias(alias) => match &alias.this {
5358 Expression::Function(f) => {
5359 assert_eq!(f.name.to_uppercase(), "DATE_PART");
5360 }
5361 other => panic!("expected generic DATE_PART function, got {other:?}"),
5362 },
5363 other => panic!("expected Alias, got {other:?}"),
5364 },
5365 other => panic!("expected Select, got {other:?}"),
5366 }
5367 }
5368
5369 #[test]
5370 fn test_lineage_join() {
5371 let expr = parse("SELECT t.a, s.b FROM t JOIN s ON t.id = s.id");
5372
5373 let node_a = lineage("a", &expr, None, false).unwrap();
5374 let names_a = node_a.downstream_names();
5375 assert!(
5376 names_a.iter().any(|n| n == "t.a"),
5377 "Expected t.a, got: {:?}",
5378 names_a
5379 );
5380
5381 let node_b = lineage("b", &expr, None, false).unwrap();
5382 let names_b = node_b.downstream_names();
5383 assert!(
5384 names_b.iter().any(|n| n == "s.b"),
5385 "Expected s.b, got: {:?}",
5386 names_b
5387 );
5388 }
5389
5390 #[test]
5391 fn test_lineage_alias_leaf_has_resolved_source_name() {
5392 let expr = parse("SELECT t1.col1 FROM table1 t1 JOIN table2 t2 ON t1.id = t2.id");
5393 let node = lineage("col1", &expr, None, false).unwrap();
5394
5395 let names = node.downstream_names();
5397 assert!(
5398 names.iter().any(|n| n == "t1.col1"),
5399 "Expected aliased column edge t1.col1, got: {:?}",
5400 names
5401 );
5402
5403 let leaf = node
5405 .downstream
5406 .iter()
5407 .find(|n| n.name == "t1.col1")
5408 .expect("Expected t1.col1 leaf");
5409 assert_eq!(leaf.source_name, "table1");
5410 match &leaf.source {
5411 Expression::Table(table) => assert_eq!(table.name.name, "table1"),
5412 _ => panic!("Expected leaf source to be a table expression"),
5413 }
5414 }
5415
5416 #[test]
5417 fn test_lineage_derived_table() {
5418 let expr = parse("SELECT x.a FROM (SELECT a FROM t) AS x");
5419 let node = lineage("a", &expr, None, false).unwrap();
5420
5421 assert_eq!(node.name, "a");
5422 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5424 assert!(
5425 all_names.iter().any(|n| n == "t.a"),
5426 "Expected to trace through derived table to t.a, got: {:?}",
5427 all_names
5428 );
5429 }
5430
5431 #[test]
5432 fn test_lineage_cte() {
5433 let expr = parse("WITH cte AS (SELECT a FROM t) SELECT a FROM cte");
5434 let node = lineage("a", &expr, None, false).unwrap();
5435
5436 assert_eq!(node.name, "a");
5437 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5438 assert!(
5439 all_names.iter().any(|n| n == "t.a"),
5440 "Expected to trace through CTE to t.a, got: {:?}",
5441 all_names
5442 );
5443 }
5444
5445 #[test]
5446 fn test_lineage_union() {
5447 let expr = parse("SELECT a FROM t1 UNION SELECT a FROM t2");
5448 let node = lineage("a", &expr, None, false).unwrap();
5449
5450 assert_eq!(node.name, "a");
5451 assert_eq!(
5453 node.downstream.len(),
5454 2,
5455 "Expected 2 branches for UNION, got {}",
5456 node.downstream.len()
5457 );
5458 }
5459
5460 #[test]
5461 fn test_lineage_cte_union() {
5462 let expr = parse("WITH cte AS (SELECT a FROM t1 UNION SELECT a FROM t2) SELECT a FROM cte");
5463 let node = lineage("a", &expr, None, false).unwrap();
5464
5465 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5467 assert!(
5468 all_names.len() >= 3,
5469 "Expected at least 3 nodes for CTE with UNION, got: {:?}",
5470 all_names
5471 );
5472 }
5473
5474 #[test]
5475 fn test_lineage_star() {
5476 let expr = parse("SELECT * FROM t");
5477 let node = lineage("*", &expr, None, false).unwrap();
5478
5479 assert_eq!(node.name, "*");
5480 assert!(
5482 !node.downstream.is_empty(),
5483 "Star should produce downstream nodes"
5484 );
5485 }
5486
5487 #[test]
5488 fn test_lineage_subquery_in_select() {
5489 let expr = parse("SELECT (SELECT MAX(b) FROM s) AS x FROM t");
5490 let node = lineage("x", &expr, None, false).unwrap();
5491
5492 assert_eq!(node.name, "x");
5493 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5495 assert!(
5496 all_names.len() >= 2,
5497 "Expected tracing into scalar subquery, got: {:?}",
5498 all_names
5499 );
5500 }
5501
5502 #[test]
5503 fn test_lineage_multiple_columns() {
5504 let expr = parse("SELECT a, b FROM t");
5505
5506 let node_a = lineage("a", &expr, None, false).unwrap();
5507 let node_b = lineage("b", &expr, None, false).unwrap();
5508
5509 assert_eq!(node_a.name, "a");
5510 assert_eq!(node_b.name, "b");
5511
5512 let names_a = node_a.downstream_names();
5514 let names_b = node_b.downstream_names();
5515 assert!(names_a.iter().any(|n| n == "t.a"));
5516 assert!(names_b.iter().any(|n| n == "t.b"));
5517 }
5518
5519 #[test]
5520 fn test_get_source_tables() {
5521 let expr = parse("SELECT t.a, s.b FROM t JOIN s ON t.id = s.id");
5522 let node = lineage("a", &expr, None, false).unwrap();
5523
5524 let tables = get_source_tables(&node);
5525 assert!(
5526 tables.contains("t"),
5527 "Expected source table 't', got: {:?}",
5528 tables
5529 );
5530 }
5531
5532 #[test]
5533 fn test_lineage_column_not_found() {
5534 let expr = parse("SELECT a FROM t");
5535 let result = lineage("nonexistent", &expr, None, false);
5536 assert!(result.is_err());
5537 }
5538
5539 #[test]
5540 fn test_lineage_nested_cte() {
5541 let expr = parse(
5542 "WITH cte1 AS (SELECT a FROM t), \
5543 cte2 AS (SELECT a FROM cte1) \
5544 SELECT a FROM cte2",
5545 );
5546 let node = lineage("a", &expr, None, false).unwrap();
5547
5548 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5550 assert!(
5551 all_names.len() >= 3,
5552 "Expected to trace through nested CTEs, got: {:?}",
5553 all_names
5554 );
5555 }
5556
5557 #[test]
5558 fn test_lineage_deeply_nested_cte_reaches_base_table() {
5559 let expr = parse(
5560 "WITH outer_cte AS (\
5561 WITH middle_cte AS (\
5562 WITH inner_cte AS (SELECT x AS col FROM base_table) \
5563 SELECT col FROM inner_cte\
5564 ) SELECT col FROM middle_cte\
5565 ) SELECT col FROM outer_cte",
5566 );
5567 let node = lineage("col", &expr, None, false).unwrap();
5568
5569 assert_lineage_contains(&node, "base_table.x");
5570 for cte_name in ["outer_cte", "middle_cte", "inner_cte"] {
5571 assert!(
5572 node.walk().any(|child| child.source_name == cte_name),
5573 "expected lineage to include CTE {cte_name}, got {:?}",
5574 lineage_names(&node)
5575 );
5576 }
5577 }
5578
5579 #[test]
5580 fn test_lineage_reused_nested_cte_traces_each_reference() {
5581 let expr = parse(
5582 "WITH shared AS (\
5583 WITH nested AS (SELECT x AS col FROM base_table) \
5584 SELECT col FROM nested\
5585 ) \
5586 SELECT s0.col + s1.col + s2.col AS total \
5587 FROM shared AS s0 \
5588 CROSS JOIN shared AS s1 \
5589 CROSS JOIN shared AS s2",
5590 );
5591 let node = lineage("total", &expr, None, false).unwrap();
5592
5593 let base_references = node
5594 .walk()
5595 .filter(|child| child.name == "base_table.x")
5596 .count();
5597 assert_eq!(
5598 base_references,
5599 3,
5600 "each shared CTE reference should reach base_table.x: {:?}",
5601 lineage_names(&node)
5602 );
5603 }
5604
5605 #[test]
5606 fn test_trim_selects_true() {
5607 let expr = parse("SELECT a, b, c FROM t");
5608 let node = lineage("a", &expr, None, true).unwrap();
5609
5610 if let Expression::Select(select) = &node.source {
5612 assert_eq!(
5613 select.expressions.len(),
5614 1,
5615 "Trimmed source should have 1 expression, got {}",
5616 select.expressions.len()
5617 );
5618 } else {
5619 panic!("Expected Select source");
5620 }
5621 }
5622
5623 #[test]
5624 fn test_trim_selects_false() {
5625 let expr = parse("SELECT a, b, c FROM t");
5626 let node = lineage("a", &expr, None, false).unwrap();
5627
5628 if let Expression::Select(select) = &node.source {
5630 assert_eq!(
5631 select.expressions.len(),
5632 3,
5633 "Untrimmed source should have 3 expressions"
5634 );
5635 } else {
5636 panic!("Expected Select source");
5637 }
5638 }
5639
5640 #[test]
5641 fn test_lineage_expression_in_select() {
5642 let expr = parse("SELECT a + b AS c FROM t");
5643 let node = lineage("c", &expr, None, false).unwrap();
5644
5645 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
5647 assert!(
5648 all_names.len() >= 3,
5649 "Expected to trace a + b to both columns, got: {:?}",
5650 all_names
5651 );
5652 }
5653
5654 #[test]
5655 fn test_set_operation_by_index() {
5656 let expr = parse("SELECT a FROM t1 UNION SELECT b FROM t2");
5657
5658 let node = lineage("a", &expr, None, false).unwrap();
5660
5661 assert_eq!(node.downstream.len(), 2);
5663 }
5664
5665 #[test]
5666 fn test_issue_384_output_columns_preserve_unknown_positions() {
5667 let expr = parse("SELECT a, 1, *, tail FROM unknown_source");
5668 let output = output_columns(&expr, None).expect("output columns");
5669
5670 assert!(!output.ordinal_complete);
5671 assert_eq!(
5672 output.columns,
5673 vec![
5674 OutputColumn::Named {
5675 name: "a".to_string(),
5676 ordinal: Some(0),
5677 },
5678 OutputColumn::Unnamed { ordinal: Some(1) },
5679 OutputColumn::Wildcard {
5680 qualifier: None,
5681 start_ordinal: Some(2),
5682 },
5683 OutputColumn::Named {
5684 name: "tail".to_string(),
5685 ordinal: None,
5686 },
5687 ]
5688 );
5689 }
5690
5691 #[test]
5692 fn test_issue_384_output_columns_use_leftmost_set_operation_branch() {
5693 let expr = parse(
5694 "SELECT * FROM unknown_source UNION ALL \
5695 SELECT known_first AS x, known_second AS y FROM known_source",
5696 );
5697 let output = output_columns(&expr, None).expect("output columns");
5698
5699 assert!(!output.ordinal_complete);
5700 assert_eq!(
5701 output.columns,
5702 vec![OutputColumn::Wildcard {
5703 qualifier: None,
5704 start_ordinal: Some(0),
5705 }]
5706 );
5707 }
5708
5709 #[test]
5710 fn test_issue_384_schema_expands_output_wildcard() {
5711 let expr = parse("SELECT * FROM unknown_source");
5712 let mut schema = MappingSchema::new();
5713 schema
5714 .add_table(
5715 "unknown_source",
5716 &[
5717 ("first_col".to_string(), DataType::Text),
5718 ("second_col".to_string(), DataType::Text),
5719 ],
5720 None,
5721 )
5722 .expect("schema setup");
5723
5724 let output = output_columns_with_schema(&expr, Some(&schema), None)
5725 .expect("schema-aware output columns");
5726 assert!(output.ordinal_complete);
5727 assert_eq!(
5728 output.columns,
5729 vec![
5730 OutputColumn::Named {
5731 name: "first_col".to_string(),
5732 ordinal: Some(0),
5733 },
5734 OutputColumn::Named {
5735 name: "second_col".to_string(),
5736 ordinal: Some(1),
5737 },
5738 ]
5739 );
5740 }
5741
5742 #[test]
5743 fn test_issue_411_duckdb_and_snowflake_union_by_name_layout_and_lineage() {
5744 let sql = "SELECT 1 AS left_value UNION ALL BY NAME SELECT 2 AS right_value";
5745
5746 for dialect in [DialectType::DuckDB, DialectType::Snowflake] {
5747 let expr = parse_dialect(sql, dialect);
5748 let output = output_columns(&expr, Some(dialect)).expect("output columns");
5749 assert_eq!(
5750 output.columns,
5751 vec![
5752 OutputColumn::Named {
5753 name: "left_value".to_string(),
5754 ordinal: Some(0),
5755 },
5756 OutputColumn::Named {
5757 name: "right_value".to_string(),
5758 ordinal: Some(1),
5759 },
5760 ],
5761 "unexpected output layout for {dialect:?}"
5762 );
5763
5764 let left = lineage_at(0, &expr, Some(dialect), false).expect("left lineage");
5765 assert_eq!(
5766 left.downstream
5767 .iter()
5768 .map(|node| node.name.as_str())
5769 .collect::<Vec<_>>(),
5770 vec!["left_value"]
5771 );
5772 assert_eq!(
5773 left.downstream[0].set_branch,
5774 Some(SetBranch {
5775 operator: SetOperator::Union,
5776 ordinal: 0,
5777 all: true,
5778 })
5779 );
5780
5781 let right = lineage_at(1, &expr, Some(dialect), false).expect("right lineage");
5782 assert_eq!(
5783 right
5784 .downstream
5785 .iter()
5786 .map(|node| node.name.as_str())
5787 .collect::<Vec<_>>(),
5788 vec!["right_value"]
5789 );
5790 assert_eq!(
5791 right.downstream[0].set_branch,
5792 Some(SetBranch {
5793 operator: SetOperator::Union,
5794 ordinal: 1,
5795 all: true,
5796 })
5797 );
5798 }
5799 }
5800
5801 #[test]
5802 fn test_issue_411_bigquery_strict_by_name_reorders_lineage() {
5803 let dialect = DialectType::BigQuery;
5804 let expr = parse_dialect(
5805 "SELECT 1 AS a, 2 AS b UNION ALL BY NAME SELECT 3 AS b, 4 AS a",
5806 dialect,
5807 );
5808 let output = output_columns(&expr, Some(dialect)).expect("output columns");
5809 assert_eq!(
5810 output.columns,
5811 vec![
5812 OutputColumn::Named {
5813 name: "a".to_string(),
5814 ordinal: Some(0),
5815 },
5816 OutputColumn::Named {
5817 name: "b".to_string(),
5818 ordinal: Some(1),
5819 },
5820 ]
5821 );
5822
5823 for (ordinal, expected_name) in [(0, "a"), (1, "b")] {
5824 let node = lineage_at(ordinal, &expr, Some(dialect), false).expect("lineage");
5825 assert_eq!(
5826 node.downstream
5827 .iter()
5828 .map(|child| child.name.as_str())
5829 .collect::<Vec<_>>(),
5830 vec![expected_name, expected_name]
5831 );
5832 }
5833 }
5834
5835 #[test]
5836 fn test_issue_411_bigquery_by_name_modes_and_explicit_order() {
5837 let dialect = DialectType::BigQuery;
5838 for (sql, expected) in [
5839 (
5840 "SELECT 1 AS a, 2 AS b INNER UNION ALL BY NAME SELECT 3 AS b, 4 AS c",
5841 vec!["b"],
5842 ),
5843 (
5844 "SELECT 1 AS a, 2 AS b LEFT OUTER UNION ALL BY NAME SELECT 3 AS b, 4 AS c",
5845 vec!["a", "b"],
5846 ),
5847 (
5848 "SELECT 1 AS a, 2 AS b FULL OUTER UNION ALL BY NAME SELECT 3 AS b, 4 AS c",
5849 vec!["a", "b", "c"],
5850 ),
5851 (
5852 "SELECT 1 AS a, 2 AS b FULL OUTER UNION ALL BY NAME ON (c, a) SELECT 3 AS b, 4 AS c",
5853 vec!["c", "a"],
5854 ),
5855 (
5856 "SELECT 1 AS a, 2 AS b INTERSECT DISTINCT BY NAME SELECT 3 AS b, 4 AS a",
5857 vec!["a", "b"],
5858 ),
5859 (
5860 "SELECT 1 AS a, 2 AS b EXCEPT DISTINCT BY NAME SELECT 3 AS b, 4 AS a",
5861 vec!["a", "b"],
5862 ),
5863 ] {
5864 let expr = parse_dialect(sql, dialect);
5865 let output = output_columns(&expr, Some(dialect)).expect("output columns");
5866 assert_eq!(
5867 output
5868 .columns
5869 .iter()
5870 .filter_map(|column| match column {
5871 OutputColumn::Named { name, .. } => Some(name.as_str()),
5872 _ => None,
5873 })
5874 .collect::<Vec<_>>(),
5875 expected,
5876 "unexpected output layout for {sql}"
5877 );
5878 }
5879 }
5880
5881 #[test]
5882 fn test_issue_411_name_matching_uses_dialect_identifier_rules() {
5883 let sql = "SELECT 1 AS \"lower\" UNION ALL BY NAME SELECT 2 AS lower";
5884
5885 let duckdb = parse_dialect(sql, DialectType::DuckDB);
5886 assert_eq!(
5887 output_columns(&duckdb, Some(DialectType::DuckDB))
5888 .expect("DuckDB output")
5889 .columns
5890 .len(),
5891 1,
5892 "DuckDB quoted identifiers are case-insensitive"
5893 );
5894
5895 let snowflake = parse_dialect(sql, DialectType::Snowflake);
5896 assert_eq!(
5897 output_columns(&snowflake, Some(DialectType::Snowflake))
5898 .expect("Snowflake output")
5899 .columns
5900 .len(),
5901 2,
5902 "Snowflake quoted lowercase and unquoted uppercase identifiers differ"
5903 );
5904 }
5905
5906 #[test]
5907 fn test_issue_383_lineage_at_traces_resolvable_set_operation_branch() {
5908 let expr = parse(
5909 "SELECT * FROM unknown_source UNION ALL \
5910 SELECT known_first AS x, known_second AS y FROM known_source",
5911 );
5912 let node = lineage_at(1, &expr, None, false).expect("ordinal lineage");
5913
5914 assert_lineage_contains(&node, "known_source.known_second");
5915 assert_eq!(node.downstream.len(), 1);
5916 assert_eq!(
5917 node.downstream[0].set_branch,
5918 Some(SetBranch {
5919 operator: SetOperator::Union,
5920 ordinal: 1,
5921 all: true,
5922 })
5923 );
5924
5925 let expr = parse(
5926 "SELECT known_first, known_second FROM known_source EXCEPT \
5927 SELECT * FROM unknown_source",
5928 );
5929 let node = lineage_at(1, &expr, None, false).expect("left-only ordinal lineage");
5930
5931 assert_lineage_contains(&node, "known_source.known_second");
5932 assert_eq!(node.downstream.len(), 1);
5933 assert_eq!(
5934 node.downstream[0].set_branch,
5935 Some(SetBranch {
5936 operator: SetOperator::Except,
5937 ordinal: 0,
5938 all: false,
5939 })
5940 );
5941 }
5942
5943 #[test]
5944 fn test_issue_383_unresolved_wildcard_does_not_shift_ordinal() {
5945 let expr = parse(
5946 "SELECT *, tail FROM unknown_source UNION ALL \
5947 SELECT known_first, known_second FROM known_source",
5948 );
5949 let node = lineage_at(1, &expr, None, false).expect("partial ordinal lineage");
5950 let names = lineage_names(&node);
5951
5952 assert!(names.iter().any(|name| name == "known_source.known_second"));
5953 assert!(!names
5954 .iter()
5955 .any(|name| name.ends_with(".tail") || name == "tail"));
5956 }
5957
5958 #[test]
5959 fn test_issue_383_lineage_at_ignores_branch_output_names() {
5960 let expr = parse(
5961 "SELECT left_value AS left_name FROM left_source UNION ALL \
5962 SELECT right_value AS right_name FROM right_source",
5963 );
5964 let node = lineage_at(0, &expr, None, false).expect("ordinal lineage");
5965
5966 assert_lineage_contains(&node, "left_source.left_value");
5967 assert_lineage_contains(&node, "right_source.right_value");
5968 }
5969
5970 #[test]
5971 fn test_issue_383_lineage_at_supports_all_set_operations() {
5972 for (sql_operator, operator, all) in [
5973 ("UNION ALL", SetOperator::Union, true),
5974 ("INTERSECT", SetOperator::Intersect, false),
5975 ("EXCEPT", SetOperator::Except, false),
5976 ] {
5977 let expr = parse(&format!(
5978 "SELECT left_value AS left_name FROM left_source {sql_operator} \
5979 SELECT right_value AS right_name FROM right_source"
5980 ));
5981 let node = lineage_at(0, &expr, None, false).expect("ordinal lineage");
5982 assert_eq!(
5983 node.downstream.len(),
5984 2,
5985 "expected both branches for {sql_operator}"
5986 );
5987 assert_eq!(
5988 node.downstream
5989 .iter()
5990 .map(|child| child.set_branch)
5991 .collect::<Vec<_>>(),
5992 vec![
5993 Some(SetBranch {
5994 operator,
5995 ordinal: 0,
5996 all,
5997 }),
5998 Some(SetBranch {
5999 operator,
6000 ordinal: 1,
6001 all,
6002 }),
6003 ]
6004 );
6005 }
6006 }
6007
6008 #[test]
6009 fn test_issue_383_lineage_at_with_schema_expands_wildcard() {
6010 let expr = parse(
6011 "SELECT * FROM unknown_source UNION ALL \
6012 SELECT known_first, known_second FROM known_source",
6013 );
6014 let mut schema = MappingSchema::new();
6015 schema
6016 .add_table(
6017 "unknown_source",
6018 &[
6019 ("first_col".to_string(), DataType::Text),
6020 ("second_col".to_string(), DataType::Text),
6021 ],
6022 None,
6023 )
6024 .expect("schema setup");
6025 schema
6026 .add_table(
6027 "known_source",
6028 &[
6029 ("known_first".to_string(), DataType::Text),
6030 ("known_second".to_string(), DataType::Text),
6031 ],
6032 None,
6033 )
6034 .expect("schema setup");
6035
6036 let node = lineage_at_with_schema(1, &expr, Some(&schema), None, false)
6037 .expect("schema-aware ordinal lineage");
6038 assert_lineage_contains(&node, "unknown_source.second_col");
6039 assert_lineage_contains(&node, "known_source.known_second");
6040 }
6041
6042 #[test]
6043 fn test_issue_385_structured_lineage_resolution_errors() {
6044 let out_of_range = lineage_at(1, &parse("SELECT a FROM t"), None, false)
6045 .expect_err("ordinal should be out of range");
6046 assert!(matches!(
6047 out_of_range,
6048 Error::ColumnResolution {
6049 target: ColumnResolutionTarget::Ordinal { ordinal: 1 },
6050 reason: ColumnResolutionReason::NotFound,
6051 }
6052 ));
6053
6054 let indeterminate = lineage(
6055 "tail",
6056 &parse(
6057 "SELECT *, tail FROM unknown_source UNION ALL \
6058 SELECT known_first, known_second FROM known_source",
6059 ),
6060 None,
6061 false,
6062 )
6063 .expect_err("tail ordinal should be indeterminate");
6064 assert!(matches!(
6065 indeterminate,
6066 Error::ColumnResolution {
6067 target: ColumnResolutionTarget::Name { ref name },
6068 reason: ColumnResolutionReason::Indeterminate,
6069 } if name == "tail"
6070 ));
6071
6072 let ambiguous = lineage(
6073 "duplicate",
6074 &parse("SELECT a AS duplicate, b AS duplicate FROM t"),
6075 None,
6076 false,
6077 )
6078 .expect_err("duplicate output name should be ambiguous");
6079 assert!(matches!(
6080 ambiguous,
6081 Error::ColumnResolution {
6082 target: ColumnResolutionTarget::Name { ref name },
6083 reason: ColumnResolutionReason::Ambiguous,
6084 } if name == "duplicate"
6085 ));
6086 }
6087
6088 fn print_node(node: &LineageNode, indent: usize) {
6091 let pad = " ".repeat(indent);
6092 println!(
6093 "{pad}name={:?} source_name={:?}",
6094 node.name, node.source_name
6095 );
6096 for child in &node.downstream {
6097 print_node(child, indent + 1);
6098 }
6099 }
6100
6101 #[test]
6102 fn test_issue18_repro() {
6103 let query = "SELECT UPPER(name) as upper_name FROM users";
6105 println!("Query: {query}\n");
6106
6107 let dialect = crate::dialects::Dialect::get(DialectType::BigQuery);
6108 let exprs = dialect.parse(query).unwrap();
6109 let expr = &exprs[0];
6110
6111 let node = lineage("upper_name", expr, Some(DialectType::BigQuery), false).unwrap();
6112 println!("lineage(\"upper_name\"):");
6113 print_node(&node, 1);
6114
6115 let names = node.downstream_names();
6116 assert!(
6117 names.iter().any(|n| n == "users.name"),
6118 "Expected users.name in downstream, got: {:?}",
6119 names
6120 );
6121 }
6122
6123 #[test]
6124 fn test_lineage_bigquery_safe_namespace_issue207() {
6125 let query = r#"
6126WITH import_cte AS (
6127 SELECT timestamp, data, operation
6128 FROM `project`.`dataset`.`source_table`
6129),
6130transform_cte AS (
6131 SELECT
6132 timestamp,
6133 SAFE.PARSE_JSON(data) AS json_data
6134 FROM import_cte
6135)
6136SELECT json_data FROM transform_cte
6137"#;
6138 let expr = parse_one(query, DialectType::BigQuery).expect("parse");
6139 let node = lineage("json_data", &expr, Some(DialectType::BigQuery), false)
6140 .expect("lineage should resolve SAFE.PARSE_JSON arguments");
6141 let names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6142
6143 assert!(
6144 names.iter().any(|name| name == "source_table.data"),
6145 "expected source_table.data in lineage, got {names:?}"
6146 );
6147 assert!(
6148 !names
6149 .iter()
6150 .any(|name| name.eq_ignore_ascii_case("import_cte.safe")),
6151 "did not expect SAFE namespace receiver in lineage, got {names:?}"
6152 );
6153 }
6154
6155 #[test]
6156 fn test_lineage_bigquery_safe_namespace_method_call_guard() {
6157 let expr = parse("SELECT SAFE.PARSE_JSON(data) AS json_data FROM t");
6158 let node = lineage("json_data", &expr, Some(DialectType::BigQuery), false)
6159 .expect("lineage should resolve SAFE.PARSE_JSON arguments");
6160 let names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6161
6162 assert!(
6163 names.iter().any(|name| name == "t.data"),
6164 "expected t.data in lineage, got {names:?}"
6165 );
6166 assert!(
6167 !names.iter().any(|name| name.eq_ignore_ascii_case("t.safe")),
6168 "did not expect SAFE namespace receiver in lineage, got {names:?}"
6169 );
6170 }
6171
6172 #[test]
6173 fn test_lineage_method_call_receiver_control() {
6174 let expr = parse("SELECT obj.METHOD(arg) AS out FROM t");
6175 let node = lineage("out", &expr, None, false).expect("lineage");
6176 let names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6177
6178 assert!(
6179 names.iter().any(|name| name == "t.obj"),
6180 "expected ordinary method receiver to remain in lineage, got {names:?}"
6181 );
6182 assert!(
6183 names.iter().any(|name| name == "t.arg"),
6184 "expected method argument in lineage, got {names:?}"
6185 );
6186 }
6187
6188 #[test]
6189 fn test_lineage_upper_function() {
6190 let expr = parse("SELECT UPPER(name) AS upper_name FROM users");
6191 let node = lineage("upper_name", &expr, None, false).unwrap();
6192
6193 let names = node.downstream_names();
6194 assert!(
6195 names.iter().any(|n| n == "users.name"),
6196 "Expected users.name in downstream, got: {:?}",
6197 names
6198 );
6199 }
6200
6201 #[test]
6202 fn test_lineage_round_function() {
6203 let expr = parse("SELECT ROUND(price, 2) AS rounded FROM products");
6204 let node = lineage("rounded", &expr, None, false).unwrap();
6205
6206 let names = node.downstream_names();
6207 assert!(
6208 names.iter().any(|n| n == "products.price"),
6209 "Expected products.price in downstream, got: {:?}",
6210 names
6211 );
6212 }
6213
6214 #[test]
6215 fn test_lineage_coalesce_function() {
6216 let expr = parse("SELECT COALESCE(a, b) AS val FROM t");
6217 let node = lineage("val", &expr, None, false).unwrap();
6218
6219 let names = node.downstream_names();
6220 assert!(
6221 names.iter().any(|n| n == "t.a"),
6222 "Expected t.a in downstream, got: {:?}",
6223 names
6224 );
6225 assert!(
6226 names.iter().any(|n| n == "t.b"),
6227 "Expected t.b in downstream, got: {:?}",
6228 names
6229 );
6230 }
6231
6232 #[test]
6233 fn test_lineage_count_function() {
6234 let expr = parse("SELECT COUNT(id) AS cnt FROM t");
6235 let node = lineage("cnt", &expr, None, false).unwrap();
6236
6237 let names = node.downstream_names();
6238 assert!(
6239 names.iter().any(|n| n == "t.id"),
6240 "Expected t.id in downstream, got: {:?}",
6241 names
6242 );
6243 }
6244
6245 #[test]
6246 fn test_lineage_sum_function() {
6247 let expr = parse("SELECT SUM(amount) AS total FROM t");
6248 let node = lineage("total", &expr, None, false).unwrap();
6249
6250 let names = node.downstream_names();
6251 assert!(
6252 names.iter().any(|n| n == "t.amount"),
6253 "Expected t.amount in downstream, got: {:?}",
6254 names
6255 );
6256 }
6257
6258 #[test]
6259 fn test_lineage_case_with_nested_functions() {
6260 let expr =
6261 parse("SELECT CASE WHEN x > 0 THEN UPPER(name) ELSE LOWER(name) END AS result FROM t");
6262 let node = lineage("result", &expr, None, false).unwrap();
6263
6264 let names = node.downstream_names();
6265 assert!(
6266 names.iter().any(|n| n == "t.x"),
6267 "Expected t.x in downstream, got: {:?}",
6268 names
6269 );
6270 assert!(
6271 names.iter().any(|n| n == "t.name"),
6272 "Expected t.name in downstream, got: {:?}",
6273 names
6274 );
6275 }
6276
6277 #[test]
6278 fn test_lineage_substring_function() {
6279 let expr = parse("SELECT SUBSTRING(name, 1, 3) AS short FROM t");
6280 let node = lineage("short", &expr, None, false).unwrap();
6281
6282 let names = node.downstream_names();
6283 assert!(
6284 names.iter().any(|n| n == "t.name"),
6285 "Expected t.name in downstream, got: {:?}",
6286 names
6287 );
6288 }
6289
6290 #[test]
6293 fn test_lineage_cte_select_star() {
6294 let expr = parse("WITH y AS (SELECT * FROM x) SELECT a FROM y");
6298 let node = lineage("a", &expr, None, false).unwrap();
6299
6300 assert_eq!(node.name, "a");
6301 assert!(
6304 !node.downstream.is_empty(),
6305 "Expected downstream nodes tracing through CTE, got none"
6306 );
6307 }
6308
6309 #[test]
6310 fn test_lineage_schema_less_cte_star_passthrough_resolves_base_column() {
6311 let expr = parse("WITH c AS (SELECT * FROM t) SELECT c.x FROM c");
6312 let node = lineage("x", &expr, None, false).unwrap();
6313
6314 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6315 assert!(
6316 all_names.iter().any(|name| name == "t.x"),
6317 "Expected schema-less CTE star passthrough to reach t.x, got: {:?}",
6318 all_names
6319 );
6320
6321 let cte_node = node
6322 .walk()
6323 .find(|child| child.source_kind == SourceKind::Cte && child.source_name == "c")
6324 .expect("expected CTE hop with source_name c");
6325 assert_eq!(cte_node.source_kind, SourceKind::Cte);
6326 assert_eq!(cte_node.source_name, "c");
6327 }
6328
6329 #[test]
6330 fn test_lineage_schema_less_cte_star_passthrough_with_aggregation() {
6331 let expr = parse(
6332 "WITH c AS (SELECT * FROM t) \
6333 SELECT SUM(c.x) AS s FROM c GROUP BY 1",
6334 );
6335 let node = lineage("s", &expr, None, false).unwrap();
6336
6337 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6338 assert!(
6339 all_names.iter().any(|name| name == "t.x"),
6340 "Expected aggregate over CTE star passthrough to reach t.x, got: {:?}",
6341 all_names
6342 );
6343 }
6344
6345 #[test]
6346 fn test_lineage_schema_less_cte_star_passthrough_with_join_and_alias() {
6347 let expr = parse(
6348 "WITH a AS (SELECT * FROM t1), b AS (SELECT * FROM t2) \
6349 SELECT SUM(b.x) AS s FROM a LEFT JOIN b ON b.id = a.id GROUP BY a.k",
6350 );
6351 let node = lineage("s", &expr, None, false).unwrap();
6352
6353 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6354 assert!(
6355 all_names.iter().any(|name| name == "t2.x"),
6356 "Expected joined CTE star passthrough to reach t2.x, got: {:?}",
6357 all_names
6358 );
6359 }
6360
6361 #[test]
6362 fn test_lineage_schema_less_chained_cte_star_passthrough() {
6363 let expr = parse(
6364 "WITH c1 AS (SELECT * FROM t), \
6365 c2 AS (SELECT * FROM c1), \
6366 c3 AS (SELECT * FROM c2) \
6367 SELECT c3.x FROM c3",
6368 );
6369 let node = lineage("x", &expr, None, false).unwrap();
6370
6371 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6372 assert!(
6373 all_names.iter().any(|name| name == "t.x"),
6374 "Expected chained CTE star passthrough to reach t.x, got: {:?}",
6375 all_names
6376 );
6377 }
6378
6379 #[test]
6380 fn test_lineage_schema_less_unqualified_star_with_multiple_sources_does_not_guess() {
6381 let expr = parse("SELECT * FROM t1 JOIN t2 ON t1.id = t2.id");
6382 let result = lineage("x", &expr, None, false);
6383
6384 assert!(
6385 result.is_err(),
6386 "Unqualified star over multiple sources should remain ambiguous, got: {:?}",
6387 result
6388 );
6389 }
6390
6391 #[test]
6392 fn test_lineage_cte_select_star_renamed_column() {
6393 let expr =
6396 parse("WITH renamed AS (SELECT id AS customer_id FROM source) SELECT * FROM renamed");
6397 let node = lineage("customer_id", &expr, None, false).unwrap();
6398
6399 assert_eq!(node.name, "customer_id");
6400 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6402 assert!(
6403 all_names.len() >= 2,
6404 "Expected at least 2 nodes (customer_id → source), got: {:?}",
6405 all_names
6406 );
6407 }
6408
6409 #[test]
6410 fn test_lineage_cte_select_star_multiple_columns() {
6411 let expr = parse("WITH cte AS (SELECT a, b, c FROM t) SELECT * FROM cte");
6413
6414 for col in &["a", "b", "c"] {
6415 let node = lineage(col, &expr, None, false).unwrap();
6416 assert_eq!(node.name, *col);
6417 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6419 assert!(
6420 all_names.len() >= 2,
6421 "Expected at least 2 nodes for column {}, got: {:?}",
6422 col,
6423 all_names
6424 );
6425 }
6426 }
6427
6428 #[test]
6429 fn test_lineage_nested_cte_select_star() {
6430 let expr = parse(
6432 "WITH cte1 AS (SELECT a FROM t), \
6433 cte2 AS (SELECT * FROM cte1) \
6434 SELECT * FROM cte2",
6435 );
6436 let node = lineage("a", &expr, None, false).unwrap();
6437
6438 assert_eq!(node.name, "a");
6439 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6440 assert!(
6441 all_names.len() >= 3,
6442 "Expected at least 3 nodes (a → cte2 → cte1 → t.a), got: {:?}",
6443 all_names
6444 );
6445 }
6446
6447 #[test]
6448 fn test_lineage_three_level_nested_cte_star() {
6449 let expr = parse(
6451 "WITH cte1 AS (SELECT x FROM t), \
6452 cte2 AS (SELECT * FROM cte1), \
6453 cte3 AS (SELECT * FROM cte2) \
6454 SELECT * FROM cte3",
6455 );
6456 let node = lineage("x", &expr, None, false).unwrap();
6457
6458 assert_eq!(node.name, "x");
6459 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6460 assert!(
6461 all_names.len() >= 4,
6462 "Expected at least 4 nodes through 3-level CTE chain, got: {:?}",
6463 all_names
6464 );
6465 }
6466
6467 #[test]
6468 fn test_lineage_cte_union_star() {
6469 let expr = parse(
6471 "WITH cte AS (SELECT a, b FROM t1 UNION ALL SELECT a, b FROM t2) \
6472 SELECT * FROM cte",
6473 );
6474 let node = lineage("a", &expr, None, false).unwrap();
6475
6476 assert_eq!(node.name, "a");
6477 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6478 assert!(
6479 all_names.len() >= 2,
6480 "Expected at least 2 nodes for CTE union star, got: {:?}",
6481 all_names
6482 );
6483 }
6484
6485 #[test]
6486 fn test_issue_368_expand_cte_stars_rewrites_every_union_arm() {
6487 let mut expr = parse_one(ISSUE_368_SQL, DialectType::BigQuery).unwrap();
6488
6489 expand_cte_stars(&mut expr, None);
6490
6491 assert_eq!(
6492 crate::generate(&expr, DialectType::BigQuery).unwrap(),
6493 "WITH base AS (SELECT 1 AS col_a), literal_branch AS (SELECT 2 AS col_a), \
6494 unioned AS (SELECT base.col_a FROM base UNION ALL \
6495 SELECT literal_branch.col_a FROM literal_branch) \
6496 SELECT col_a FROM unioned"
6497 );
6498 }
6499
6500 #[test]
6501 fn test_issue_368_lineage_resolves_non_leftmost_union_star() {
6502 let expr = parse_one(ISSUE_368_SQL, DialectType::BigQuery).unwrap();
6503
6504 let node = lineage("col_a", &expr, Some(DialectType::BigQuery), false).unwrap();
6505 let names = lineage_names(&node);
6506
6507 assert!(
6508 node.walk().any(|child| {
6509 child.name == "col_a"
6510 && child.source_name == "literal_branch"
6511 && child.source_kind == SourceKind::Cte
6512 }),
6513 "expected the right UNION branch to resolve to literal_branch.col_a, got {node:#?}"
6514 );
6515 assert!(
6516 !names
6517 .iter()
6518 .any(|name| name == "*" || name == "literal_branch.*"),
6519 "did not expect an unresolved right-branch star, got {names:?}"
6520 );
6521 }
6522
6523 #[test]
6524 fn test_expand_cte_stars_rewrites_all_set_operation_kinds() {
6525 for operator in ["UNION ALL", "INTERSECT", "EXCEPT"] {
6526 let mut expr = parse(&format!(
6527 "WITH left_cte AS (SELECT 1 AS col_a), \
6528 right_cte AS (SELECT 2 AS col_a), \
6529 combined AS (SELECT * FROM left_cte {operator} SELECT * FROM right_cte) \
6530 SELECT col_a FROM combined"
6531 ));
6532
6533 expand_cte_stars(&mut expr, None);
6534
6535 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6536 assert!(
6537 sql.contains("SELECT left_cte.col_a FROM left_cte"),
6538 "expected left arm expansion for {operator}, got {sql}"
6539 );
6540 assert!(
6541 sql.contains("SELECT right_cte.col_a FROM right_cte"),
6542 "expected right arm expansion for {operator}, got {sql}"
6543 );
6544 }
6545 }
6546
6547 #[test]
6548 fn test_expand_cte_stars_rewrites_nested_parenthesized_set_operations() {
6549 let mut expr = parse(
6550 "WITH a AS (SELECT 1 AS x), \
6551 b AS (SELECT 2 AS x), \
6552 c AS (SELECT 3 AS x), \
6553 combined AS ((SELECT * FROM a UNION ALL SELECT * FROM b) \
6554 UNION ALL SELECT * FROM c) \
6555 SELECT x FROM combined",
6556 );
6557
6558 expand_cte_stars(&mut expr, None);
6559
6560 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6561 for source in ["a", "b", "c"] {
6562 assert!(
6563 sql.contains(&format!("SELECT {source}.x FROM {source}")),
6564 "expected nested arm {source} to be expanded, got {sql}"
6565 );
6566 }
6567 }
6568
6569 #[test]
6570 fn test_expand_cte_stars_rewrites_root_set_operation() {
6571 let mut expr = parse(
6572 "WITH a AS (SELECT 1 AS x), b AS (SELECT 2 AS x) \
6573 SELECT * FROM a UNION ALL SELECT * FROM b",
6574 );
6575
6576 expand_cte_stars(&mut expr, None);
6577
6578 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6579 assert!(
6580 sql.contains("SELECT a.x FROM a UNION ALL SELECT b.x FROM b"),
6581 "expected both root UNION arms to be expanded, got {sql}"
6582 );
6583 }
6584
6585 #[test]
6586 fn test_expand_cte_stars_preserves_leftmost_output_names() {
6587 let mut expr = parse(
6588 "WITH a AS (SELECT 1 AS left_name), \
6589 b AS (SELECT 2 AS right_name), \
6590 combined AS (SELECT * FROM a UNION ALL SELECT * FROM b) \
6591 SELECT * FROM combined",
6592 );
6593
6594 expand_cte_stars(&mut expr, None);
6595
6596 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6597 assert!(
6598 sql.ends_with("SELECT combined.left_name FROM combined"),
6599 "expected the set operation output name to come from the left arm, got {sql}"
6600 );
6601 assert!(
6602 sql.contains("SELECT b.right_name FROM b"),
6603 "expected the differently named right arm to still be expanded, got {sql}"
6604 );
6605 }
6606
6607 #[test]
6608 fn test_expand_cte_stars_rewrites_body_with_explicit_cte_columns() {
6609 let mut expr = parse(
6610 "WITH a AS (SELECT 1 AS x), \
6611 b AS (SELECT 2 AS x), \
6612 combined(output_name) AS (SELECT * FROM a UNION ALL SELECT * FROM b) \
6613 SELECT * FROM combined",
6614 );
6615
6616 expand_cte_stars(&mut expr, None);
6617
6618 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6619 assert!(
6620 sql.contains("SELECT a.x FROM a UNION ALL SELECT b.x FROM b"),
6621 "expected explicit aliases not to suppress body expansion, got {sql}"
6622 );
6623 assert!(
6624 sql.ends_with("SELECT combined.output_name FROM combined"),
6625 "expected the explicit CTE output name to override the body name, got {sql}"
6626 );
6627 }
6628
6629 #[test]
6630 fn test_expand_cte_stars_keeps_recursive_self_reference_conservative() {
6631 let mut expr = parse(
6632 "WITH RECURSIVE r(x) AS (\
6633 SELECT 1 AS x UNION ALL SELECT * FROM r\
6634 ) SELECT * FROM r",
6635 );
6636
6637 expand_cte_stars(&mut expr, None);
6638
6639 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6640 assert!(
6641 sql.contains("UNION ALL SELECT * FROM r"),
6642 "expected the recursive body star to remain untouched, got {sql}"
6643 );
6644 assert!(
6645 sql.ends_with("SELECT r.x FROM r"),
6646 "expected the explicit recursive CTE column to expand the outer star, got {sql}"
6647 );
6648 }
6649
6650 #[test]
6651 fn test_expand_cte_stars_preserves_genuinely_unresolved_branch_star() {
6652 let mut expr = parse(
6653 "WITH known AS (SELECT 1 AS x), \
6654 combined AS (SELECT * FROM known UNION ALL SELECT * FROM missing) \
6655 SELECT * FROM combined",
6656 );
6657
6658 expand_cte_stars(&mut expr, None);
6659
6660 let sql = crate::generate(&expr, DialectType::Generic).unwrap();
6661 assert!(
6662 sql.contains("SELECT known.x FROM known UNION ALL SELECT * FROM missing"),
6663 "expected only the resolvable branch to expand, got {sql}"
6664 );
6665 assert!(
6666 sql.ends_with("SELECT combined.x FROM combined"),
6667 "expected the leftmost output name to remain usable, got {sql}"
6668 );
6669 }
6670
6671 #[test]
6672 fn test_lineage_cte_star_unknown_table() {
6673 let expr = parse(
6676 "WITH cte AS (SELECT * FROM unknown_table) \
6677 SELECT * FROM cte",
6678 );
6679 let _result = lineage("x", &expr, None, false);
6682 }
6683
6684 #[test]
6685 fn test_lineage_cte_explicit_columns() {
6686 let expr = parse(
6688 "WITH cte(x, y) AS (SELECT a, b FROM t) \
6689 SELECT * FROM cte",
6690 );
6691 let node = lineage("x", &expr, None, false).unwrap();
6692 assert_eq!(node.name, "x");
6693 }
6694
6695 #[test]
6696 fn test_lineage_cte_qualified_star() {
6697 let expr = parse(
6699 "WITH cte AS (SELECT a, b FROM t) \
6700 SELECT cte.* FROM cte",
6701 );
6702 for col in &["a", "b"] {
6703 let node = lineage(col, &expr, None, false).unwrap();
6704 assert_eq!(node.name, *col);
6705 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6706 assert!(
6707 all_names.len() >= 2,
6708 "Expected at least 2 nodes for qualified star column {}, got: {:?}",
6709 col,
6710 all_names
6711 );
6712 }
6713 }
6714
6715 #[test]
6716 fn test_lineage_subquery_select_star() {
6717 let expr = parse("SELECT x FROM (SELECT * FROM table_a)");
6720 let node = lineage("x", &expr, None, false).unwrap();
6721
6722 assert_eq!(node.name, "x");
6723 assert!(
6724 !node.downstream.is_empty(),
6725 "Expected downstream nodes for subquery with SELECT *, got none"
6726 );
6727 }
6728
6729 #[test]
6730 fn test_lineage_cte_star_with_schema_external_table() {
6731 let sql = r#"WITH orders AS (SELECT * FROM stg_orders)
6733SELECT * FROM orders"#;
6734 let expr = parse(sql);
6735
6736 let mut schema = MappingSchema::new();
6737 let cols = vec![
6738 ("order_id".to_string(), DataType::Unknown),
6739 ("customer_id".to_string(), DataType::Unknown),
6740 ("amount".to_string(), DataType::Unknown),
6741 ];
6742 schema.add_table("stg_orders", &cols, None).unwrap();
6743
6744 let node =
6745 lineage_with_schema("order_id", &expr, Some(&schema as &dyn Schema), None, false)
6746 .unwrap();
6747 assert_eq!(node.name, "order_id");
6748 }
6749
6750 #[test]
6751 fn test_lineage_cte_star_with_schema_three_part_name() {
6752 let sql = r#"WITH orders AS (SELECT * FROM "db"."schema"."stg_orders")
6754SELECT * FROM orders"#;
6755 let expr = parse(sql);
6756
6757 let mut schema = MappingSchema::new();
6758 let cols = vec![
6759 ("order_id".to_string(), DataType::Unknown),
6760 ("customer_id".to_string(), DataType::Unknown),
6761 ];
6762 schema
6763 .add_table("db.schema.stg_orders", &cols, None)
6764 .unwrap();
6765
6766 let node = lineage_with_schema(
6767 "customer_id",
6768 &expr,
6769 Some(&schema as &dyn Schema),
6770 None,
6771 false,
6772 )
6773 .unwrap();
6774 assert_eq!(node.name, "customer_id");
6775 }
6776
6777 #[test]
6778 fn test_lineage_cte_star_with_schema_nested() {
6779 let sql = r#"WITH
6782 raw AS (SELECT * FROM external_table),
6783 enriched AS (SELECT * FROM raw)
6784 SELECT * FROM enriched"#;
6785 let expr = parse(sql);
6786
6787 let mut schema = MappingSchema::new();
6788 let cols = vec![
6789 ("id".to_string(), DataType::Unknown),
6790 ("name".to_string(), DataType::Unknown),
6791 ];
6792 schema.add_table("external_table", &cols, None).unwrap();
6793
6794 let node =
6795 lineage_with_schema("name", &expr, Some(&schema as &dyn Schema), None, false).unwrap();
6796 assert_eq!(node.name, "name");
6797 }
6798
6799 #[test]
6800 fn test_lineage_cte_qualified_star_with_schema() {
6801 let sql = r#"WITH
6804 orders AS (SELECT * FROM stg_orders),
6805 enriched AS (
6806 SELECT orders.*, 'extra' AS extra
6807 FROM orders
6808 )
6809 SELECT * FROM enriched"#;
6810 let expr = parse(sql);
6811
6812 let mut schema = MappingSchema::new();
6813 let cols = vec![
6814 ("order_id".to_string(), DataType::Unknown),
6815 ("total".to_string(), DataType::Unknown),
6816 ];
6817 schema.add_table("stg_orders", &cols, None).unwrap();
6818
6819 let node =
6820 lineage_with_schema("order_id", &expr, Some(&schema as &dyn Schema), None, false)
6821 .unwrap();
6822 assert_eq!(node.name, "order_id");
6823
6824 let extra =
6826 lineage_with_schema("extra", &expr, Some(&schema as &dyn Schema), None, false).unwrap();
6827 assert_eq!(extra.name, "extra");
6828 }
6829
6830 #[test]
6831 fn test_lineage_cte_star_without_schema_still_works() {
6832 let sql = r#"WITH
6834 cte1 AS (SELECT id, name FROM raw_table),
6835 cte2 AS (SELECT * FROM cte1)
6836 SELECT * FROM cte2"#;
6837 let expr = parse(sql);
6838
6839 let node = lineage("id", &expr, None, false).unwrap();
6841 assert_eq!(node.name, "id");
6842 }
6843
6844 #[test]
6845 fn test_lineage_nested_cte_star_with_join_and_schema() {
6846 let sql = r#"WITH
6849base_orders AS (
6850 SELECT * FROM stg_orders
6851),
6852with_payments AS (
6853 SELECT
6854 base_orders.*,
6855 p.amount
6856 FROM base_orders
6857 LEFT JOIN stg_payments p ON base_orders.order_id = p.order_id
6858),
6859final_cte AS (
6860 SELECT * FROM with_payments
6861)
6862SELECT * FROM final_cte"#;
6863 let expr = parse(sql);
6864
6865 let mut schema = MappingSchema::new();
6866 let order_cols = vec![
6867 (
6868 "order_id".to_string(),
6869 crate::expressions::DataType::Unknown,
6870 ),
6871 (
6872 "customer_id".to_string(),
6873 crate::expressions::DataType::Unknown,
6874 ),
6875 ("status".to_string(), crate::expressions::DataType::Unknown),
6876 ];
6877 let pay_cols = vec![
6878 (
6879 "payment_id".to_string(),
6880 crate::expressions::DataType::Unknown,
6881 ),
6882 (
6883 "order_id".to_string(),
6884 crate::expressions::DataType::Unknown,
6885 ),
6886 ("amount".to_string(), crate::expressions::DataType::Unknown),
6887 ];
6888 schema.add_table("stg_orders", &order_cols, None).unwrap();
6889 schema.add_table("stg_payments", &pay_cols, None).unwrap();
6890
6891 let node =
6893 lineage_with_schema("order_id", &expr, Some(&schema as &dyn Schema), None, false)
6894 .unwrap();
6895 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6896
6897 let has_table_qualified = all_names
6899 .iter()
6900 .any(|n| n.contains('.') && n.contains("order_id"));
6901 assert!(
6902 has_table_qualified,
6903 "Expected table-qualified leaf like 'stg_orders.order_id', got: {:?}",
6904 all_names
6905 );
6906
6907 let node = lineage_with_schema("amount", &expr, Some(&schema as &dyn Schema), None, false)
6909 .unwrap();
6910 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6911
6912 let has_table_qualified = all_names
6913 .iter()
6914 .any(|n| n.contains('.') && n.contains("amount"));
6915 assert!(
6916 has_table_qualified,
6917 "Expected table-qualified leaf like 'stg_payments.amount', got: {:?}",
6918 all_names
6919 );
6920 }
6921
6922 #[test]
6923 fn test_lineage_cte_alias_resolution() {
6924 let sql = r#"WITH import_stg_items AS (
6926 SELECT item_id, name, status FROM stg_items
6927)
6928SELECT base.item_id, base.status
6929FROM import_stg_items AS base"#;
6930 let expr = parse(sql);
6931
6932 let node = lineage("item_id", &expr, None, false).unwrap();
6933 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6934 assert!(
6936 all_names.iter().any(|n| n == "stg_items.item_id"),
6937 "Expected leaf 'stg_items.item_id', got: {:?}",
6938 all_names
6939 );
6940 }
6941
6942 #[test]
6943 fn test_lineage_cte_alias_with_schema_and_star() {
6944 let sql = r#"WITH import_stg AS (
6946 SELECT * FROM stg_items
6947)
6948SELECT base.item_id, base.status
6949FROM import_stg AS base"#;
6950 let expr = parse(sql);
6951
6952 let mut schema = MappingSchema::new();
6953 schema
6954 .add_table(
6955 "stg_items",
6956 &[
6957 ("item_id".to_string(), DataType::Unknown),
6958 ("name".to_string(), DataType::Unknown),
6959 ("status".to_string(), DataType::Unknown),
6960 ],
6961 None,
6962 )
6963 .unwrap();
6964
6965 let node = lineage_with_schema("item_id", &expr, Some(&schema as &dyn Schema), None, false)
6966 .unwrap();
6967 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6968 assert!(
6969 all_names.iter().any(|n| n == "stg_items.item_id"),
6970 "Expected leaf 'stg_items.item_id', got: {:?}",
6971 all_names
6972 );
6973 }
6974
6975 #[test]
6976 fn test_lineage_cte_alias_with_join() {
6977 let sql = r#"WITH
6979 import_users AS (SELECT id, name FROM users),
6980 import_orders AS (SELECT id, user_id, amount FROM orders)
6981SELECT u.name, o.amount
6982FROM import_users AS u
6983LEFT JOIN import_orders AS o ON u.id = o.user_id"#;
6984 let expr = parse(sql);
6985
6986 let node = lineage("name", &expr, None, false).unwrap();
6987 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6988 assert!(
6989 all_names.iter().any(|n| n == "users.name"),
6990 "Expected leaf 'users.name', got: {:?}",
6991 all_names
6992 );
6993
6994 let node = lineage("amount", &expr, None, false).unwrap();
6995 let all_names: Vec<_> = node.walk().map(|n| n.name.clone()).collect();
6996 assert!(
6997 all_names.iter().any(|n| n == "orders.amount"),
6998 "Expected leaf 'orders.amount', got: {:?}",
6999 all_names
7000 );
7001 }
7002
7003 #[test]
7008 fn test_lineage_unquoted_cte_case_insensitive() {
7009 let expr = parse("WITH MyCte AS (SELECT id AS col FROM source) SELECT * FROM MYCTE");
7012 let node = lineage("col", &expr, None, false).unwrap();
7013 assert_eq!(node.name, "col");
7014 assert!(
7015 !node.downstream.is_empty(),
7016 "Unquoted CTE should resolve case-insensitively"
7017 );
7018 }
7019
7020 #[test]
7021 fn test_lineage_quoted_cte_case_preserved() {
7022 let expr = parse(r#"WITH "MyCte" AS (SELECT id AS col FROM source) SELECT * FROM "MyCte""#);
7024 let node = lineage("col", &expr, None, false).unwrap();
7025 assert_eq!(node.name, "col");
7026 assert!(
7027 !node.downstream.is_empty(),
7028 "Quoted CTE with matching case should resolve"
7029 );
7030 }
7031
7032 #[test]
7033 fn test_lineage_quoted_cte_case_mismatch_no_expansion() {
7034 let expr = parse(r#"WITH "MyCte" AS (SELECT id AS col FROM source) SELECT * FROM "mycte""#);
7038 let result = lineage("col", &expr, None, false);
7041 assert!(
7042 result.is_err(),
7043 "Quoted CTE with case mismatch should not expand star: {:?}",
7044 result
7045 );
7046 }
7047
7048 #[test]
7049 fn test_lineage_mixed_quoted_unquoted_cte() {
7050 let expr = parse(
7052 r#"WITH unquoted AS (SELECT 1 AS a FROM t), "Quoted" AS (SELECT a FROM unquoted) SELECT * FROM "Quoted""#,
7053 );
7054 let node = lineage("a", &expr, None, false).unwrap();
7055 assert_eq!(node.name, "a");
7056 assert!(
7057 !node.downstream.is_empty(),
7058 "Mixed quoted/unquoted CTE chain should resolve"
7059 );
7060 }
7061
7062 #[test]
7078 fn test_lineage_quoted_cte_case_mismatch_non_star_known_bug() {
7079 let expr = parse(r#"WITH "MyCte" AS (SELECT 1 AS col) SELECT col FROM "mycte""#);
7090 let node = lineage("col", &expr, None, false).unwrap();
7091 assert!(!node.downstream.is_empty());
7092 let child = &node.downstream[0];
7093 assert_eq!(
7095 child.source_name, "MyCte",
7096 "Known bug: quoted CTE case mismatch should NOT resolve, but currently does. \
7097 If this fails, the bug may be fixed — update to assert source_name != \"MyCte\""
7098 );
7099 }
7100
7101 #[test]
7102 fn test_lineage_quoted_cte_case_mismatch_qualified_col_known_bug() {
7103 let expr = parse(r#"WITH "MyCte" AS (SELECT 1 AS col) SELECT "mycte".col FROM "mycte""#);
7110 let node = lineage("col", &expr, None, false).unwrap();
7111 assert!(!node.downstream.is_empty());
7112 let child = &node.downstream[0];
7113 assert_eq!(
7115 child.source_name, "MyCte",
7116 "Known bug: quoted CTE case mismatch should NOT resolve, but currently does. \
7117 If this fails, the bug may be fixed — update to assert source_name != \"MyCte\""
7118 );
7119 }
7120
7121 #[test]
7122 fn test_lineage_recursive_cte_terminates_at_base_case() {
7123 let expr = parse_dialect(
7124 "WITH RECURSIVE nums AS (\
7125 SELECT 1 AS n \
7126 UNION ALL \
7127 SELECT n + 1 FROM nums WHERE n < 5\
7128 ) SELECT n FROM nums",
7129 DialectType::DuckDB,
7130 );
7131 let node = lineage("n", &expr, Some(DialectType::DuckDB), false).unwrap();
7132 let names = lineage_names(&node);
7133
7134 assert!(
7135 names.len() <= 12,
7136 "recursive CTE lineage should not unroll repeatedly, got {names:?}"
7137 );
7138 assert!(
7139 node.walk()
7140 .any(|child| child.source_kind == SourceKind::Cte && child.source_name == "nums"),
7141 "expected recursive source to be marked as a CTE, got {names:?}"
7142 );
7143 }
7144
7145 #[test]
7146 fn test_lineage_window_partition_and_order_columns() {
7147 let expr = parse(
7148 "WITH c AS (SELECT user_id, ts FROM events) \
7149 SELECT ROW_NUMBER() OVER (PARTITION BY c.user_id ORDER BY c.ts) AS out FROM c",
7150 );
7151 let node = lineage("out", &expr, None, false).unwrap();
7152
7153 assert_lineage_contains(&node, "events.user_id");
7154 assert_lineage_contains(&node, "events.ts");
7155 }
7156
7157 #[test]
7158 fn test_lineage_window_aggregate_order_column() {
7159 let expr = parse(
7160 "WITH c AS (SELECT amount, d FROM txns) \
7161 SELECT SUM(c.amount) OVER (ORDER BY c.d) AS running FROM c",
7162 );
7163 let node = lineage("running", &expr, None, false).unwrap();
7164
7165 assert_lineage_contains(&node, "txns.amount");
7166 assert_lineage_contains(&node, "txns.d");
7167 }
7168
7169 #[test]
7170 fn test_lineage_named_window_columns() {
7171 let expr = parse(
7172 "SELECT ROW_NUMBER() OVER w AS out \
7173 FROM events \
7174 WINDOW w AS (PARTITION BY user_id ORDER BY ts)",
7175 );
7176 let node = lineage("out", &expr, None, false).unwrap();
7177
7178 assert_lineage_contains(&node, "events.user_id");
7179 assert_lineage_contains(&node, "events.ts");
7180 }
7181
7182 #[test]
7183 fn test_lineage_within_group_order_column() {
7184 let expr =
7185 parse("SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY amount) AS p FROM txns");
7186 let node = lineage("p", &expr, None, false).unwrap();
7187
7188 assert_lineage_contains(&node, "txns.amount");
7189 }
7190
7191 #[test]
7192 fn test_lineage_query_wrappers_resolve_inner_select() {
7193 for sql in [
7194 "CREATE TABLE tgt AS SELECT x FROM src",
7195 "CREATE VIEW v AS SELECT x FROM src",
7196 "INSERT INTO tgt SELECT x FROM src",
7197 ] {
7198 let expr = parse(sql);
7199 let node = lineage("x", &expr, None, false).unwrap();
7200 assert_lineage_contains(&node, "src.x");
7201 }
7202 }
7203
7204 #[test]
7205 fn test_lineage_scalar_subquery_through_cte_reaches_base_table() {
7206 let expr = parse(
7207 "WITH c AS (SELECT x FROM t) \
7208 SELECT (SELECT SUM(x) FROM c) AS s FROM c LIMIT 1",
7209 );
7210 let node = lineage("s", &expr, None, false).unwrap();
7211
7212 assert_lineage_contains(&node, "t.x");
7213 assert!(
7214 node.walk()
7215 .any(|child| child.source_kind == SourceKind::Cte && child.source_name == "c"),
7216 "expected scalar subquery CTE hop in lineage, got {:?}",
7217 lineage_names(&node)
7218 );
7219 }
7220
7221 #[test]
7222 fn test_lineage_scalar_subqueries_inside_expression_wrappers() {
7223 for sql in [
7224 "WITH c AS (SELECT a, b FROM t) \
7225 SELECT CASE WHEN c.a > 0 THEN c.b ELSE (SELECT MAX(z) FROM o) END AS r FROM c",
7226 "WITH c AS (SELECT a FROM t) \
7227 SELECT COALESCE(c.a, (SELECT MAX(z) FROM o)) AS r FROM c",
7228 "WITH c AS (SELECT a FROM t) \
7229 SELECT CAST((SELECT MAX(z) FROM o) AS INT) + c.a AS r FROM c",
7230 "WITH c AS (SELECT a FROM t) \
7231 SELECT CASE WHEN c.a BETWEEN 0 AND (SELECT MAX(z) FROM o) THEN c.a END AS r FROM c",
7232 ] {
7233 let expr = parse_dialect(sql, DialectType::DuckDB);
7234 let node = lineage("r", &expr, Some(DialectType::DuckDB), false)
7235 .unwrap_or_else(|error| panic!("lineage failed for {sql}: {error}"));
7236
7237 assert_lineage_contains(&node, "o.z");
7238 assert_lineage_contains(&node, "t.a");
7239 }
7240 }
7241
7242 #[test]
7243 fn test_lineage_nested_set_operation_inside_derived_table() {
7244 let expr = parse_dialect(
7245 "SELECT v FROM ((SELECT v FROM t1 UNION ALL SELECT v FROM t2) \
7246 UNION ALL SELECT v FROM t3) u",
7247 DialectType::DuckDB,
7248 );
7249 let node = lineage("v", &expr, Some(DialectType::DuckDB), false).unwrap();
7250
7251 assert_lineage_contains(&node, "t1.v");
7252 assert_lineage_contains(&node, "t2.v");
7253 assert_lineage_contains(&node, "t3.v");
7254 }
7255
7256 #[test]
7257 fn test_lineage_select_alias_reference_resolves_to_alias_source() {
7258 let expr = parse_dialect(
7259 "WITH c AS (SELECT x FROM t) SELECT c.x AS a, a + 1 AS b FROM c",
7260 DialectType::DuckDB,
7261 );
7262 let node = lineage("b", &expr, Some(DialectType::DuckDB), false).unwrap();
7263
7264 assert_lineage_contains(&node, "t.x");
7265 }
7266
7267 #[test]
7268 fn test_lineage_pivot_output_resolves_aggregation_input() {
7269 let expr = parse_dialect(
7270 "SELECT * FROM (SELECT region, q, amt FROM sales) \
7271 PIVOT(SUM(amt) FOR q IN ('Q1' AS q1))",
7272 DialectType::DuckDB,
7273 );
7274 let node = lineage("q1", &expr, Some(DialectType::DuckDB), false).unwrap();
7275
7276 assert_lineage_contains(&node, "sales.amt");
7277 }
7278
7279 #[test]
7280 fn test_lineage_pivot_multi_aggregate_and_alias_columns() {
7281 let multi = parse_dialect(
7282 "SELECT * FROM (SELECT category, value, price FROM t) \
7283 PIVOT(SUM(value) AS value_sum, MAX(price) FOR category IN ('a' AS cat_a, 'b'))",
7284 DialectType::DuckDB,
7285 );
7286 let value_sum =
7287 lineage("cat_a_value_sum", &multi, Some(DialectType::DuckDB), false).unwrap();
7288 assert_lineage_contains(&value_sum, "t.value");
7289
7290 let max_price =
7291 lineage("cat_a_max(price)", &multi, Some(DialectType::DuckDB), false).unwrap();
7292 assert_lineage_contains(&max_price, "t.price");
7293
7294 let aliased = parse_dialect(
7295 "SELECT * FROM (SELECT region, q, amt FROM sales) \
7296 PIVOT(SUM(amt) FOR q IN ('Q1')) AS p(region2, p1)",
7297 DialectType::DuckDB,
7298 );
7299 let region = lineage("region2", &aliased, Some(DialectType::DuckDB), false).unwrap();
7300 assert_lineage_contains(®ion, "sales.region");
7301
7302 let pivot_value = lineage("p1", &aliased, Some(DialectType::DuckDB), false).unwrap();
7303 assert_lineage_contains(&pivot_value, "sales.amt");
7304 }
7305
7306 #[test]
7307 fn test_lineage_pivot_through_cte_resolves_aggregation_input() {
7308 let expr = parse_dialect(
7309 "WITH src AS (SELECT region, q, amt FROM sales) \
7310 SELECT q1 FROM src PIVOT(SUM(amt) FOR q IN ('Q1' AS q1))",
7311 DialectType::DuckDB,
7312 );
7313 let node = lineage("q1", &expr, Some(DialectType::DuckDB), false).unwrap();
7314
7315 assert_lineage_contains(&node, "sales.amt");
7316 }
7317
7318 #[test]
7319 fn test_lineage_unpivot_value_resolves_input_columns() {
7320 let expr = parse_dialect(
7321 "SELECT name, val FROM t UNPIVOT(val FOR col IN (a, b, c))",
7322 DialectType::DuckDB,
7323 );
7324 let node = lineage("val", &expr, Some(DialectType::DuckDB), false).unwrap();
7325
7326 assert_lineage_contains(&node, "t.a");
7327 assert_lineage_contains(&node, "t.b");
7328 assert_lineage_contains(&node, "t.c");
7329 }
7330
7331 #[test]
7332 fn test_lineage_unpivot_multi_value_columns_resolve_positionally() {
7333 let expr = parse_dialect(
7334 "SELECT first_half_sales, second_half_sales, semester \
7335 FROM produce \
7336 UNPIVOT((first_half_sales, second_half_sales) \
7337 FOR semester IN ((q1, q2) AS 'semester_1', (q3, q4) AS 'semester_2'))",
7338 DialectType::BigQuery,
7339 );
7340
7341 let first = lineage(
7342 "first_half_sales",
7343 &expr,
7344 Some(DialectType::BigQuery),
7345 false,
7346 )
7347 .unwrap();
7348 assert_lineage_contains(&first, "produce.q1");
7349 assert_lineage_contains(&first, "produce.q3");
7350
7351 let second = lineage(
7352 "second_half_sales",
7353 &expr,
7354 Some(DialectType::BigQuery),
7355 false,
7356 )
7357 .unwrap();
7358 assert_lineage_contains(&second, "produce.q2");
7359 assert_lineage_contains(&second, "produce.q4");
7360 }
7361
7362 #[test]
7363 fn test_lineage_top_level_union_over_ctes_reaches_base_tables() {
7364 let expr = parse(
7365 "WITH a AS (SELECT x FROM t1), b AS (SELECT x FROM t2) \
7366 SELECT x FROM a UNION SELECT x FROM b",
7367 );
7368 let node = lineage("x", &expr, None, false).unwrap();
7369
7370 assert_lineage_contains(&node, "t1.x");
7371 assert_lineage_contains(&node, "t2.x");
7372 for cte_name in ["a", "b"] {
7373 assert!(
7374 node.walk().any(|child| child.source_name == cte_name),
7375 "expected set-operation lineage to retain CTE source {cte_name}: {:?}",
7376 lineage_names(&node)
7377 );
7378 }
7379 }
7380
7381 #[test]
7382 fn test_lineage_star_excludes_semi_join_rhs_source() {
7383 let expr = parse_dialect(
7384 "SELECT * FROM orders LEFT SEMI JOIN customers ON orders.customer_id = customers.id",
7385 DialectType::DuckDB,
7386 );
7387 let node = lineage("customer_id", &expr, Some(DialectType::DuckDB), false).unwrap();
7388
7389 assert_lineage_contains(&node, "orders.customer_id");
7390 }
7391
7392 #[test]
7399 #[ignore = "requires derived table star expansion (separate issue)"]
7400 fn test_node_name_doesnt_contain_comment() {
7401 let expr = parse("SELECT * FROM (SELECT x /* c */ FROM t1) AS t2");
7402 let node = lineage("x", &expr, None, false).unwrap();
7403
7404 assert_eq!(node.name, "x");
7405 assert!(!node.downstream.is_empty());
7406 }
7407
7408 #[test]
7412 fn test_comment_before_first_column_in_cte() {
7413 let sql_with_comment = "with t as (select 1 as a) select\n -- comment\n a from t";
7414 let sql_without_comment = "with t as (select 1 as a) select a from t";
7415
7416 let expr_ok = parse(sql_without_comment);
7418 let node_ok = lineage("a", &expr_ok, None, false).expect("without comment should succeed");
7419
7420 let expr_comment = parse(sql_with_comment);
7422 let node_comment = lineage("a", &expr_comment, None, false)
7423 .expect("with comment before first column should succeed");
7424
7425 assert_eq!(node_ok.name, node_comment.name, "node names should match");
7426 assert_eq!(
7427 node_ok.downstream_names(),
7428 node_comment.downstream_names(),
7429 "downstream lineage should be identical with or without comment"
7430 );
7431 }
7432
7433 #[test]
7435 fn test_block_comment_before_first_column() {
7436 let sql = "with t as (select 1 as a) select /* section */ a from t";
7437 let expr = parse(sql);
7438 let node = lineage("a", &expr, None, false)
7439 .expect("block comment before first column should succeed");
7440 assert_eq!(node.name, "a");
7441 assert!(
7442 !node.downstream.is_empty(),
7443 "should have downstream lineage"
7444 );
7445 }
7446
7447 #[test]
7449 fn test_comment_before_first_column_second_col_ok() {
7450 let sql = "with t as (select 1 as a, 2 as b) select\n -- comment\n a, b from t";
7451 let expr = parse(sql);
7452
7453 let node_a =
7454 lineage("a", &expr, None, false).expect("column a with comment should succeed");
7455 assert_eq!(node_a.name, "a");
7456
7457 let node_b =
7458 lineage("b", &expr, None, false).expect("column b with comment should succeed");
7459 assert_eq!(node_b.name, "b");
7460 }
7461
7462 #[test]
7464 fn test_comment_before_aliased_column() {
7465 let sql = "with t as (select 1 as x) select\n -- renamed\n x as y from t";
7466 let expr = parse(sql);
7467 let node =
7468 lineage("y", &expr, None, false).expect("aliased column with comment should succeed");
7469 assert_eq!(node.name, "y");
7470 assert!(
7471 !node.downstream.is_empty(),
7472 "aliased column should have downstream lineage"
7473 );
7474 }
7475}