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