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