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