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