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