1use crate::ast_transforms::get_output_column_names;
9use crate::dialects::{Dialect, DialectType};
10use crate::expressions::{DataType, Expression, JoinKind, TableRef, With};
11use crate::lineage::{lineage_by_index_from_expression, LineageNode};
12use crate::optimizer::annotate_types::annotate_types;
13use crate::optimizer::qualify_columns::{qualify_columns, QualifyColumnsOptions};
14use crate::schema::{MappingSchema, Schema};
15use crate::scope::{build_scope, Scope, SourceInfo, SourceKind};
16use crate::traversal::{contains_aggregate, ExpressionWalk};
17use crate::validation::{mapping_schema_from_validation_schema_with_dialect, ValidationSchema};
18use crate::{parse_one, Error, Result};
19use serde::{Deserialize, Serialize};
20use std::collections::{HashMap, HashSet};
21
22#[derive(Debug, Clone, Serialize, Deserialize, Default)]
24#[serde(rename_all = "camelCase", default)]
25pub struct AnalyzeQueryOptions {
26 pub dialect: DialectType,
28 pub schema: Option<ValidationSchema>,
30}
31
32#[derive(Debug, Clone, Serialize, Deserialize)]
34#[serde(rename_all = "camelCase")]
35pub struct QueryAnalysis {
36 pub shape: QueryShape,
37 pub ctes: Vec<String>,
38 pub cte_facts: Vec<CteFact>,
39 pub projections: Vec<ProjectionFact>,
40 pub relations: Vec<RelationFact>,
41 pub base_tables: Vec<RelationFact>,
42 pub star_projections: Vec<StarProjectionFact>,
43 pub set_operations: Vec<SetOperationFact>,
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(rename_all = "snake_case")]
49pub enum QueryShape {
50 Select,
51 SetOperation,
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize)]
56#[serde(rename_all = "camelCase")]
57pub struct ProjectionFact {
58 pub index: usize,
59 pub name: Option<String>,
60 pub is_star: bool,
61 pub star_table: Option<String>,
62 pub transform_kind: TransformKind,
63 #[serde(skip_serializing_if = "Option::is_none")]
64 pub transform_function: Option<TransformFunctionFact>,
65 pub cast_type: Option<String>,
66 pub type_hint: Option<String>,
67 pub nullability: ProjectionNullability,
68 pub upstream: Vec<ColumnReferenceFact>,
69}
70
71#[derive(Debug, Clone, Serialize, Deserialize)]
73#[serde(rename_all = "camelCase")]
74pub struct TransformFunctionFact {
75 pub name: String,
76 pub literal_args: Vec<String>,
77 pub column_args: Vec<ColumnReferenceFact>,
78}
79
80#[derive(Debug, Clone, Serialize, Deserialize)]
82#[serde(rename_all = "camelCase")]
83pub struct CteFact {
84 pub name: String,
85 pub columns: Vec<String>,
86 pub body_sql: String,
87 pub output_columns: Vec<String>,
88}
89
90#[derive(Debug, Clone, Serialize, Deserialize)]
92#[serde(rename_all = "camelCase")]
93pub struct StarProjectionFact {
94 pub index: usize,
95 pub table: Option<String>,
96 pub expanded_columns: Vec<String>,
97}
98
99#[derive(Debug, Clone, Serialize, Deserialize)]
101#[serde(rename_all = "camelCase")]
102pub struct ColumnReferenceFact {
103 pub source_name: Option<String>,
104 pub source_alias: Option<String>,
105 pub source_kind: SourceKind,
106 pub table: Option<String>,
107 pub column: String,
108 pub unqualified: bool,
109 pub confidence: ReferenceConfidence,
110}
111
112#[derive(Debug, Clone, Serialize, Deserialize)]
114#[serde(rename_all = "camelCase")]
115pub struct RelationFact {
116 pub name: String,
117 pub alias: Option<String>,
118 pub kind: SourceKind,
119 pub columns: Vec<String>,
120 pub catalog: Option<String>,
121 pub schema: Option<String>,
122 pub table: Option<String>,
123}
124
125#[derive(Debug, Clone, Serialize, Deserialize)]
127#[serde(rename_all = "camelCase")]
128pub struct SetOperationFact {
129 pub kind: String,
130 pub all: bool,
131 pub distinct: bool,
132 pub output_columns: Vec<String>,
133 pub branches: Vec<SetOperationBranchFact>,
134}
135
136#[derive(Debug, Clone, Serialize, Deserialize)]
138#[serde(rename_all = "camelCase")]
139pub struct SetOperationBranchFact {
140 pub index: usize,
141 pub projections: Vec<ProjectionFact>,
142}
143
144#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
146#[serde(rename_all = "snake_case")]
147pub enum TransformKind {
148 Direct,
149 Cast,
150 Aggregation,
151 Constant,
152 Expression,
153 Star,
154}
155
156#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
158#[serde(rename_all = "snake_case")]
159pub enum ReferenceConfidence {
160 Resolved,
161 Ambiguous,
162 Unknown,
163}
164
165#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
167#[serde(rename_all = "snake_case")]
168pub enum ProjectionNullability {
169 NonNull,
170 Nullable,
171 Unknown,
172}
173
174pub fn analyze_query(sql: &str, options: AnalyzeQueryOptions) -> Result<QueryAnalysis> {
176 let mut expression = parse_one(sql, options.dialect)?;
177 expression = effective_query(expression);
178 ensure_query(&expression)?;
179 let original_expression = expression.clone();
180
181 let mapping_schema = options
182 .schema
183 .as_ref()
184 .map(|schema| analysis_mapping_schema(schema, options.dialect));
185 let schema_info = options.schema.as_ref().map(AnalysisSchemaInfo::from_schema);
186 let cte_facts = top_level_cte_facts(&original_expression, options.dialect)?;
187 let star_projections = star_projection_facts(&original_expression, mapping_schema.as_ref());
188
189 if let Some(schema) = mapping_schema.as_ref() {
190 let qualify_options = QualifyColumnsOptions::new()
191 .with_dialect(options.dialect)
192 .with_allow_partial(true);
193 expression = qualify_columns(expression, schema, &qualify_options)
194 .map_err(|e| Error::internal(format!("query analysis qualification failed: {e}")))?;
195 }
196
197 annotate_types(
198 &mut expression,
199 mapping_schema.as_ref().map(|schema| schema as &dyn Schema),
200 Some(options.dialect),
201 );
202 crate::lineage::expand_cte_stars(
203 &mut expression,
204 mapping_schema.as_ref().map(|schema| schema as &dyn Schema),
205 );
206
207 let scope = build_scope(&expression);
208 let nullability_context = NullabilityContext {
209 schema: schema_info.as_ref(),
210 nullable_sources: nullable_source_names(&expression),
211 };
212 let shape = if is_set_operation(&expression) {
213 QueryShape::SetOperation
214 } else {
215 QueryShape::Select
216 };
217
218 Ok(QueryAnalysis {
219 shape,
220 ctes: collect_cte_names(&expression),
221 cte_facts,
222 projections: projection_facts_for_query(
223 &expression,
224 &scope,
225 options.dialect,
226 &nullability_context,
227 ),
228 relations: relation_facts(&scope, mapping_schema.as_ref()),
229 base_tables: base_table_facts(&scope, mapping_schema.as_ref()),
230 star_projections,
231 set_operations: set_operation_facts(&expression, &scope, options.dialect),
232 })
233}
234
235fn analysis_mapping_schema(schema: &ValidationSchema, dialect: DialectType) -> MappingSchema {
236 mapping_schema_from_validation_schema_with_dialect(schema, dialect)
237}
238
239fn validation_table_names(table: &crate::validation::SchemaTable) -> Vec<String> {
240 let mut names = Vec::new();
241
242 names.push(table.name.to_ascii_lowercase());
243 if let Some(schema_name) = &table.schema {
244 names.push(format!(
245 "{}.{}",
246 schema_name.to_ascii_lowercase(),
247 table.name.to_ascii_lowercase()
248 ));
249 }
250 for alias in &table.aliases {
251 names.push(alias.to_ascii_lowercase());
252 }
253
254 names.sort();
255 names.dedup();
256 names
257}
258
259#[derive(Debug, Clone)]
260struct AnalysisColumnInfo {
261 nullable: Option<bool>,
262 primary_key: bool,
263}
264
265#[derive(Debug, Clone)]
266struct AnalysisSchemaInfo {
267 columns: HashMap<(String, String), AnalysisColumnInfo>,
268}
269
270impl AnalysisSchemaInfo {
271 fn from_schema(schema: &ValidationSchema) -> Self {
272 let mut columns = HashMap::new();
273
274 for table in &schema.tables {
275 let table_names = validation_table_names(table);
276 let primary_keys: HashSet<String> = table
277 .primary_key
278 .iter()
279 .map(|column| column.to_ascii_lowercase())
280 .collect();
281
282 for column in &table.columns {
283 let info = AnalysisColumnInfo {
284 nullable: column.nullable,
285 primary_key: column.primary_key
286 || primary_keys.contains(&column.name.to_ascii_lowercase()),
287 };
288
289 for table_name in &table_names {
290 columns.insert(
291 (
292 normalize_lookup_name(table_name),
293 normalize_lookup_name(&column.name),
294 ),
295 info.clone(),
296 );
297 }
298 }
299 }
300
301 Self { columns }
302 }
303
304 fn column(&self, table: &str, column: &str) -> Option<&AnalysisColumnInfo> {
305 self.columns
306 .get(&(normalize_lookup_name(table), normalize_lookup_name(column)))
307 }
308}
309
310struct NullabilityContext<'a> {
311 schema: Option<&'a AnalysisSchemaInfo>,
312 nullable_sources: HashSet<String>,
313}
314
315fn top_level_cte_facts(expression: &Expression, dialect: DialectType) -> Result<Vec<CteFact>> {
316 let Some(with_clause) = with_clause(expression) else {
317 return Ok(Vec::new());
318 };
319
320 with_clause
321 .ctes
322 .iter()
323 .map(|cte| {
324 Ok(CteFact {
325 name: cte.alias.name.clone(),
326 columns: cte
327 .columns
328 .iter()
329 .map(|column| column.name.clone())
330 .collect(),
331 body_sql: Dialect::get(dialect).generate(&cte.this)?,
332 output_columns: get_output_column_names(&cte.this),
333 })
334 })
335 .collect()
336}
337
338fn star_projection_facts(
339 expression: &Expression,
340 mapping_schema: Option<&MappingSchema>,
341) -> Vec<StarProjectionFact> {
342 let scope = build_scope(expression);
343 let ordered_sources = ordered_source_names_for_query(expression);
344
345 select_expressions_for_query(expression)
346 .iter()
347 .enumerate()
348 .filter_map(|(index, projection)| {
349 let inner = unwrap_projection_alias(projection);
350 if !projection_is_star(inner) {
351 return None;
352 }
353
354 let table = projection_star_table(inner);
355 let expanded_columns =
356 expanded_star_columns(table.as_deref(), &scope, &ordered_sources, mapping_schema);
357
358 Some(StarProjectionFact {
359 index,
360 table,
361 expanded_columns,
362 })
363 })
364 .collect()
365}
366
367fn expanded_star_columns(
368 star_table: Option<&str>,
369 scope: &Scope,
370 ordered_sources: &[String],
371 mapping_schema: Option<&MappingSchema>,
372) -> Vec<String> {
373 let mut columns = Vec::new();
374 let mut source_names: Vec<String> = if ordered_sources.is_empty() {
375 let mut names: Vec<_> = scope.sources.keys().cloned().collect();
376 names.sort();
377 names
378 } else {
379 ordered_sources.to_vec()
380 };
381
382 source_names.dedup();
383
384 for source_name in source_names {
385 let Some(source) = scope.sources.get(&source_name) else {
386 continue;
387 };
388
389 if let Some(star_table) = star_table {
390 let matches = source_name.eq_ignore_ascii_case(star_table)
391 || source
392 .alias
393 .as_deref()
394 .is_some_and(|alias| alias.eq_ignore_ascii_case(star_table))
395 || source_table_name(source)
396 .is_some_and(|table| table.eq_ignore_ascii_case(star_table));
397
398 if !matches {
399 continue;
400 }
401 }
402
403 columns.extend(source_columns(source, mapping_schema));
404 }
405
406 columns
407}
408
409fn ordered_source_names_for_query(expression: &Expression) -> Vec<String> {
410 match expression {
411 Expression::Select(select) => ordered_source_names_for_select(select),
412 Expression::Union(union) => ordered_source_names_for_query(&union.left),
413 Expression::Intersect(intersect) => ordered_source_names_for_query(&intersect.left),
414 Expression::Except(except) => ordered_source_names_for_query(&except.left),
415 Expression::Subquery(subquery) => ordered_source_names_for_query(&subquery.this),
416 _ => Vec::new(),
417 }
418}
419
420fn ordered_source_names_for_select(select: &crate::expressions::Select) -> Vec<String> {
421 let mut sources = Vec::new();
422
423 if let Some(from) = &select.from {
424 for expression in &from.expressions {
425 if let Some(source_name) = expression_source_name(expression) {
426 sources.push(source_name);
427 }
428 }
429 }
430
431 for join in &select.joins {
432 if let Some(source_name) = expression_source_name(&join.this) {
433 sources.push(source_name);
434 }
435 }
436
437 sources
438}
439
440fn nullable_source_names(expression: &Expression) -> HashSet<String> {
441 match expression {
442 Expression::Select(select) => nullable_source_names_for_select(select),
443 Expression::Union(union) => nullable_source_names(&union.left),
444 Expression::Intersect(intersect) => nullable_source_names(&intersect.left),
445 Expression::Except(except) => nullable_source_names(&except.left),
446 Expression::Subquery(subquery) => nullable_source_names(&subquery.this),
447 _ => HashSet::new(),
448 }
449}
450
451fn nullable_source_names_for_select(select: &crate::expressions::Select) -> HashSet<String> {
452 let mut nullable = HashSet::new();
453 let mut left_sources = Vec::new();
454
455 if let Some(from) = &select.from {
456 for expression in &from.expressions {
457 if let Some(source_name) = expression_source_name(expression) {
458 left_sources.push(source_name);
459 }
460 }
461 }
462
463 for join in &select.joins {
464 let right_source = expression_source_name(&join.this);
465
466 if join_nullable_left(join.kind) {
467 for source_name in &left_sources {
468 nullable.insert(normalize_lookup_name(source_name));
469 }
470 }
471
472 if join_nullable_right(join.kind) {
473 if let Some(source_name) = &right_source {
474 nullable.insert(normalize_lookup_name(source_name));
475 }
476 }
477
478 if let Some(source_name) = right_source {
479 left_sources.push(source_name);
480 }
481 }
482
483 nullable
484}
485
486fn join_nullable_left(kind: JoinKind) -> bool {
487 matches!(
488 kind,
489 JoinKind::Right
490 | JoinKind::NaturalRight
491 | JoinKind::AsOfRight
492 | JoinKind::Full
493 | JoinKind::NaturalFull
494 | JoinKind::Outer
495 )
496}
497
498fn join_nullable_right(kind: JoinKind) -> bool {
499 matches!(
500 kind,
501 JoinKind::Left
502 | JoinKind::NaturalLeft
503 | JoinKind::AsOfLeft
504 | JoinKind::LeftLateral
505 | JoinKind::OuterApply
506 | JoinKind::LeftArray
507 | JoinKind::Full
508 | JoinKind::NaturalFull
509 | JoinKind::Outer
510 )
511}
512
513fn expression_source_name(expression: &Expression) -> Option<String> {
514 match expression {
515 Expression::Table(table) => table
516 .alias
517 .as_ref()
518 .map(|alias| alias.name.clone())
519 .or_else(|| Some(table.name.name.clone())),
520 Expression::Subquery(subquery) => subquery.alias.as_ref().map(|alias| alias.name.clone()),
521 Expression::Alias(alias) => Some(alias.alias.name.clone()),
522 Expression::Cte(cte) => Some(cte.alias.name.clone()),
523 _ => None,
524 }
525}
526
527fn normalize_lookup_name(name: &str) -> String {
528 name.to_ascii_lowercase()
529}
530
531fn effective_query(expression: Expression) -> Expression {
532 match expression {
533 Expression::Prepare(prepare) => prepare.statement,
534 Expression::Subquery(subquery) if subquery.alias.is_none() => subquery.this,
535 other => other,
536 }
537}
538
539fn ensure_query(expression: &Expression) -> Result<()> {
540 if matches!(
541 expression,
542 Expression::Select(_)
543 | Expression::Union(_)
544 | Expression::Intersect(_)
545 | Expression::Except(_)
546 ) {
547 Ok(())
548 } else {
549 Err(Error::internal(
550 "analyze_query requires a SELECT or set operation query",
551 ))
552 }
553}
554
555fn is_set_operation(expression: &Expression) -> bool {
556 matches!(
557 expression,
558 Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_)
559 )
560}
561
562fn collect_cte_names(expression: &Expression) -> Vec<String> {
563 let mut names = Vec::new();
564 let mut seen = HashSet::new();
565 collect_cte_names_inner(expression, &mut names, &mut seen);
566 names
567}
568
569fn collect_cte_names_inner(
570 expression: &Expression,
571 names: &mut Vec<String>,
572 seen: &mut HashSet<String>,
573) {
574 if let Some(with_clause) = with_clause(expression) {
575 collect_with_names(with_clause, names, seen);
576 }
577
578 match expression {
579 Expression::Union(union) => {
580 collect_cte_names_inner(&union.left, names, seen);
581 collect_cte_names_inner(&union.right, names, seen);
582 }
583 Expression::Intersect(intersect) => {
584 collect_cte_names_inner(&intersect.left, names, seen);
585 collect_cte_names_inner(&intersect.right, names, seen);
586 }
587 Expression::Except(except) => {
588 collect_cte_names_inner(&except.left, names, seen);
589 collect_cte_names_inner(&except.right, names, seen);
590 }
591 Expression::Subquery(subquery) => collect_cte_names_inner(&subquery.this, names, seen),
592 _ => {}
593 }
594}
595
596fn collect_with_names(with_clause: &With, names: &mut Vec<String>, seen: &mut HashSet<String>) {
597 for cte in &with_clause.ctes {
598 if seen.insert(cte.alias.name.clone()) {
599 names.push(cte.alias.name.clone());
600 }
601 collect_cte_names_inner(&cte.this, names, seen);
602 }
603}
604
605fn with_clause(expression: &Expression) -> Option<&With> {
606 match expression {
607 Expression::Select(select) => select.with.as_ref(),
608 Expression::Union(union) => union.with.as_ref(),
609 Expression::Intersect(intersect) => intersect.with.as_ref(),
610 Expression::Except(except) => except.with.as_ref(),
611 _ => None,
612 }
613}
614
615fn projection_facts_for_query(
616 expression: &Expression,
617 scope: &Scope,
618 dialect: DialectType,
619 nullability_context: &NullabilityContext<'_>,
620) -> Vec<ProjectionFact> {
621 let expressions = select_expressions_for_query(expression);
622 let names = get_output_column_names(expression);
623
624 expressions
625 .iter()
626 .enumerate()
627 .map(|(index, projection)| {
628 projection_fact(
629 index,
630 names
631 .get(index)
632 .cloned()
633 .or_else(|| projection_name(projection)),
634 projection,
635 expression,
636 scope,
637 dialect,
638 nullability_context,
639 )
640 })
641 .collect()
642}
643
644fn select_expressions_for_query(expression: &Expression) -> Vec<&Expression> {
645 match expression {
646 Expression::Select(select) => select.expressions.iter().collect(),
647 Expression::Union(union) => select_expressions_for_query(&union.left),
648 Expression::Intersect(intersect) => select_expressions_for_query(&intersect.left),
649 Expression::Except(except) => select_expressions_for_query(&except.left),
650 Expression::Subquery(subquery) => select_expressions_for_query(&subquery.this),
651 _ => Vec::new(),
652 }
653}
654
655fn projection_fact(
656 index: usize,
657 name: Option<String>,
658 projection: &Expression,
659 query: &Expression,
660 scope: &Scope,
661 dialect: DialectType,
662 nullability_context: &NullabilityContext<'_>,
663) -> ProjectionFact {
664 let inner = unwrap_projection_alias(projection);
665 let is_star = projection_is_star(inner);
666 let upstream = lineage_by_index_from_expression(index, query, Some(dialect), false)
667 .map(|node| terminal_references_from_lineage(&node))
668 .ok()
669 .filter(|refs| !refs.is_empty())
670 .unwrap_or_else(|| fallback_column_references(inner, scope));
671
672 ProjectionFact {
673 index,
674 name,
675 is_star,
676 star_table: projection_star_table(inner),
677 transform_kind: transform_kind(inner),
678 transform_function: transform_function_fact(inner, scope, dialect),
679 cast_type: cast_type(inner, dialect),
680 type_hint: projection
681 .inferred_type()
682 .or_else(|| inner.inferred_type())
683 .and_then(|data_type| render_data_type(data_type, dialect)),
684 nullability: projection_nullability(inner, scope, nullability_context),
685 upstream,
686 }
687}
688
689fn transform_function_fact(
690 expression: &Expression,
691 scope: &Scope,
692 dialect: DialectType,
693) -> Option<TransformFunctionFact> {
694 let mut matches = expression
695 .find_all(|candidate| transform_function_fact_for_node(candidate, scope, dialect).is_some())
696 .into_iter();
697
698 let first = matches.next()?;
699 if matches.next().is_some() {
700 return None;
701 }
702
703 transform_function_fact_for_node(first, scope, dialect)
704}
705
706fn transform_function_fact_for_node(
707 expression: &Expression,
708 scope: &Scope,
709 dialect: DialectType,
710) -> Option<TransformFunctionFact> {
711 match expression {
712 Expression::Function(function) => Some(transform_function_from_args(
713 &function.name,
714 &function.args,
715 scope,
716 dialect,
717 )),
718 Expression::AggregateFunction(function) => Some(transform_function_from_args(
719 &function.name,
720 &function.args,
721 scope,
722 dialect,
723 )),
724 Expression::DateTrunc(function) => Some(transform_function_from_parts(
725 "DATE_TRUNC",
726 vec![datetime_field_name(&function.unit)],
727 vec![&function.this],
728 scope,
729 dialect,
730 )),
731 Expression::TimestampTrunc(function) => Some(transform_function_from_parts(
732 "TIMESTAMP_TRUNC",
733 vec![datetime_field_name(&function.unit)],
734 vec![&function.this],
735 scope,
736 dialect,
737 )),
738 Expression::TimeTrunc(function) => {
739 let mut args = vec![function.this.as_ref()];
740 if let Some(zone) = function.zone.as_deref() {
741 args.push(zone);
742 }
743 Some(transform_function_from_parts(
744 "TIME_TRUNC",
745 vec![function.unit.clone()],
746 args,
747 scope,
748 dialect,
749 ))
750 }
751 Expression::Extract(function) => Some(transform_function_from_parts(
752 "EXTRACT",
753 vec![datetime_field_name(&function.field)],
754 vec![&function.this],
755 scope,
756 dialect,
757 )),
758 Expression::DateAdd(function) => Some(transform_function_from_parts(
759 "DATE_ADD",
760 Vec::new(),
761 vec![&function.this, &function.interval],
762 scope,
763 dialect,
764 )),
765 Expression::DateSub(function) => Some(transform_function_from_parts(
766 "DATE_SUB",
767 Vec::new(),
768 vec![&function.this, &function.interval],
769 scope,
770 dialect,
771 )),
772 Expression::DateDiff(function) => Some(transform_function_from_parts(
773 "DATE_DIFF",
774 Vec::new(),
775 vec![&function.this, &function.expression],
776 scope,
777 dialect,
778 )),
779 _ => None,
780 }
781}
782
783fn transform_function_from_args(
784 name: &str,
785 args: &[Expression],
786 scope: &Scope,
787 dialect: DialectType,
788) -> TransformFunctionFact {
789 let literal_args = args
790 .iter()
791 .filter_map(|arg| literal_argument(arg, dialect))
792 .collect();
793 transform_function_from_parts(name, literal_args, args.iter().collect(), scope, dialect)
794}
795
796fn transform_function_from_parts(
797 name: &str,
798 literal_args: Vec<String>,
799 args: Vec<&Expression>,
800 scope: &Scope,
801 _dialect: DialectType,
802) -> TransformFunctionFact {
803 let column_args = dedupe_column_refs(
804 args.into_iter()
805 .flat_map(|arg| fallback_column_references(arg, scope))
806 .collect(),
807 );
808
809 TransformFunctionFact {
810 name: name.to_string(),
811 literal_args,
812 column_args,
813 }
814}
815
816fn literal_argument(expression: &Expression, dialect: DialectType) -> Option<String> {
817 match expression {
818 Expression::Literal(literal) => Some(literal.value_str().to_string()),
819 Expression::Boolean(boolean) => Some(boolean.value.to_string()),
820 Expression::Null(_) => Some("NULL".to_string()),
821 Expression::Identifier(identifier) => Some(identifier.name.clone()),
822 Expression::Var(var) => Some(var.this.clone()),
823 Expression::DataType(data_type) => render_data_type(data_type, dialect),
824 _ => None,
825 }
826}
827
828fn datetime_field_name(field: &crate::expressions::DateTimeField) -> String {
829 match field {
830 crate::expressions::DateTimeField::Year => "year".to_string(),
831 crate::expressions::DateTimeField::Month => "month".to_string(),
832 crate::expressions::DateTimeField::Day => "day".to_string(),
833 crate::expressions::DateTimeField::Hour => "hour".to_string(),
834 crate::expressions::DateTimeField::Minute => "minute".to_string(),
835 crate::expressions::DateTimeField::Second => "second".to_string(),
836 crate::expressions::DateTimeField::Millisecond => "millisecond".to_string(),
837 crate::expressions::DateTimeField::Microsecond => "microsecond".to_string(),
838 crate::expressions::DateTimeField::DayOfWeek => "day_of_week".to_string(),
839 crate::expressions::DateTimeField::DayOfYear => "day_of_year".to_string(),
840 crate::expressions::DateTimeField::Week => "week".to_string(),
841 crate::expressions::DateTimeField::WeekWithModifier(modifier) => {
842 format!("week({modifier})")
843 }
844 crate::expressions::DateTimeField::Quarter => "quarter".to_string(),
845 crate::expressions::DateTimeField::Epoch => "epoch".to_string(),
846 crate::expressions::DateTimeField::Timezone => "timezone".to_string(),
847 crate::expressions::DateTimeField::TimezoneHour => "timezone_hour".to_string(),
848 crate::expressions::DateTimeField::TimezoneMinute => "timezone_minute".to_string(),
849 crate::expressions::DateTimeField::Date => "date".to_string(),
850 crate::expressions::DateTimeField::Time => "time".to_string(),
851 crate::expressions::DateTimeField::Custom(name) => name.clone(),
852 }
853}
854
855fn unwrap_projection_alias(expression: &Expression) -> &Expression {
856 match expression {
857 Expression::Alias(alias) => unwrap_projection_alias(&alias.this),
858 Expression::Annotated(annotated) => unwrap_projection_alias(&annotated.this),
859 Expression::Paren(paren) => unwrap_projection_alias(&paren.this),
860 _ => expression,
861 }
862}
863
864fn projection_name(expression: &Expression) -> Option<String> {
865 match expression {
866 Expression::Alias(alias) => Some(alias.alias.name.clone()),
867 Expression::Column(column) => Some(column.name.name.clone()),
868 Expression::Identifier(identifier) => Some(identifier.name.clone()),
869 Expression::Star(_) => Some("*".to_string()),
870 Expression::Annotated(annotated) => projection_name(&annotated.this),
871 _ => None,
872 }
873}
874
875fn projection_is_star(expression: &Expression) -> bool {
876 matches!(expression, Expression::Star(_))
877 || matches!(expression, Expression::Column(column) if column.name.name == "*")
878}
879
880fn projection_star_table(expression: &Expression) -> Option<String> {
881 match expression {
882 Expression::Star(star) => star
883 .table
884 .as_ref()
885 .map(|identifier| identifier.name.clone()),
886 Expression::Column(column) if column.name.name == "*" => column
887 .table
888 .as_ref()
889 .map(|identifier| identifier.name.clone()),
890 _ => None,
891 }
892}
893
894fn transform_kind(expression: &Expression) -> TransformKind {
895 if projection_is_star(expression) {
896 TransformKind::Star
897 } else if is_cast_expression(expression) {
898 TransformKind::Cast
899 } else if contains_aggregate(expression) {
900 TransformKind::Aggregation
901 } else if matches!(
902 expression,
903 Expression::Column(_) | Expression::Identifier(_)
904 ) {
905 TransformKind::Direct
906 } else if is_simple_constant(expression) {
907 TransformKind::Constant
908 } else {
909 TransformKind::Expression
910 }
911}
912
913fn is_cast_expression(expression: &Expression) -> bool {
914 matches!(
915 expression,
916 Expression::Cast(_) | Expression::TryCast(_) | Expression::SafeCast(_)
917 )
918}
919
920fn cast_type(expression: &Expression, dialect: DialectType) -> Option<String> {
921 match expression {
922 Expression::Cast(cast) | Expression::TryCast(cast) | Expression::SafeCast(cast) => {
923 render_data_type(&cast.to, dialect)
924 }
925 _ => None,
926 }
927}
928
929fn render_data_type(data_type: &DataType, dialect: DialectType) -> Option<String> {
930 Dialect::get(dialect)
931 .generate(&Expression::DataType(data_type.clone()))
932 .ok()
933}
934
935fn is_simple_constant(expression: &Expression) -> bool {
936 match expression {
937 Expression::Literal(_) | Expression::Boolean(_) | Expression::Null(_) => true,
938 Expression::Cast(cast) | Expression::TryCast(cast) | Expression::SafeCast(cast) => {
939 is_simple_constant(&cast.this)
940 }
941 Expression::Neg(unary) | Expression::BitwiseNot(unary) => is_simple_constant(&unary.this),
942 _ => false,
943 }
944}
945
946fn projection_nullability(
947 expression: &Expression,
948 scope: &Scope,
949 context: &NullabilityContext<'_>,
950) -> ProjectionNullability {
951 match expression {
952 Expression::Alias(alias) => projection_nullability(&alias.this, scope, context),
953 Expression::Annotated(annotated) => projection_nullability(&annotated.this, scope, context),
954 Expression::Paren(paren) => projection_nullability(&paren.this, scope, context),
955 Expression::Literal(_) | Expression::Boolean(_) => ProjectionNullability::NonNull,
956 Expression::Null(_) => ProjectionNullability::Nullable,
957 Expression::Count(_) | Expression::CountIf(_) => ProjectionNullability::NonNull,
958 Expression::Cast(cast) => projection_nullability(&cast.this, scope, context),
959 Expression::TryCast(_) | Expression::SafeCast(_) => ProjectionNullability::Unknown,
960 Expression::Column(column) => column_nullability(
961 &column.name.name,
962 column.table.as_ref().map(|table| table.name.as_str()),
963 scope,
964 context,
965 ),
966 Expression::Identifier(identifier) => {
967 column_nullability(&identifier.name, None, scope, context)
968 }
969 Expression::Coalesce(func) => coalesce_nullability(&func.expressions, scope, context),
970 _ => ProjectionNullability::Unknown,
971 }
972}
973
974fn column_nullability(
975 column_name: &str,
976 source_name: Option<&str>,
977 scope: &Scope,
978 context: &NullabilityContext<'_>,
979) -> ProjectionNullability {
980 let resolved_source_name = source_name
981 .map(str::to_string)
982 .or_else(|| single_scope_source_name(scope));
983
984 if let Some(source_name) = &resolved_source_name {
985 if context
986 .nullable_sources
987 .contains(&normalize_lookup_name(source_name))
988 {
989 return ProjectionNullability::Nullable;
990 }
991 }
992
993 let Some(schema) = context.schema else {
994 return ProjectionNullability::Unknown;
995 };
996
997 let table_name = resolved_source_name
998 .as_ref()
999 .and_then(|name| scope.sources.get(name).and_then(source_table_name))
1000 .or(resolved_source_name);
1001
1002 let Some(table_name) = table_name else {
1003 return ProjectionNullability::Unknown;
1004 };
1005
1006 match schema.column(&table_name, column_name) {
1007 Some(info) if info.primary_key || info.nullable == Some(false) => {
1008 ProjectionNullability::NonNull
1009 }
1010 Some(info) if info.nullable == Some(true) => ProjectionNullability::Nullable,
1011 Some(_) | None => ProjectionNullability::Unknown,
1012 }
1013}
1014
1015fn single_scope_source_name(scope: &Scope) -> Option<String> {
1016 if scope.sources.len() == 1 {
1017 scope.sources.keys().next().cloned()
1018 } else {
1019 None
1020 }
1021}
1022
1023fn coalesce_nullability(
1024 expressions: &[Expression],
1025 scope: &Scope,
1026 context: &NullabilityContext<'_>,
1027) -> ProjectionNullability {
1028 if expressions.is_empty() {
1029 return ProjectionNullability::Unknown;
1030 }
1031
1032 let mut all_nullable = true;
1033
1034 for expression in expressions {
1035 match projection_nullability(unwrap_projection_alias(expression), scope, context) {
1036 ProjectionNullability::NonNull => return ProjectionNullability::NonNull,
1037 ProjectionNullability::Nullable => {}
1038 ProjectionNullability::Unknown => all_nullable = false,
1039 }
1040 }
1041
1042 if all_nullable {
1043 ProjectionNullability::Nullable
1044 } else {
1045 ProjectionNullability::Unknown
1046 }
1047}
1048
1049fn terminal_references_from_lineage(node: &LineageNode) -> Vec<ColumnReferenceFact> {
1050 let mut refs = Vec::new();
1051 collect_terminal_references(node, &mut refs);
1052 dedupe_column_refs(refs)
1053}
1054
1055fn collect_terminal_references(node: &LineageNode, refs: &mut Vec<ColumnReferenceFact>) {
1056 if node.downstream.is_empty() {
1057 if let Some(reference) = column_reference_from_lineage_node(node) {
1058 refs.push(reference);
1059 }
1060 return;
1061 }
1062
1063 for child in &node.downstream {
1064 collect_terminal_references(child, refs);
1065 }
1066}
1067
1068fn column_reference_from_lineage_node(node: &LineageNode) -> Option<ColumnReferenceFact> {
1069 match &node.expression {
1070 Expression::Column(column) => {
1071 let source_name = non_empty_string(node.source_name.clone());
1072 let table =
1073 lineage_node_table(node).or_else(|| column.table.as_ref().map(|t| t.name.clone()));
1074 let confidence = if node.source_kind == SourceKind::Unknown && source_name.is_none() {
1075 ReferenceConfidence::Unknown
1076 } else {
1077 ReferenceConfidence::Resolved
1078 };
1079 Some(ColumnReferenceFact {
1080 source_name,
1081 source_alias: node.source_alias.clone(),
1082 source_kind: node.source_kind,
1083 table,
1084 column: column.name.name.clone(),
1085 unqualified: column.table.is_none(),
1086 confidence,
1087 })
1088 }
1089 Expression::Star(_) => Some(ColumnReferenceFact {
1090 source_name: non_empty_string(node.source_name.clone()),
1091 source_alias: node.source_alias.clone(),
1092 source_kind: node.source_kind,
1093 table: lineage_node_table(node),
1094 column: "*".to_string(),
1095 unqualified: true,
1096 confidence: if node.source_kind == SourceKind::Unknown {
1097 ReferenceConfidence::Unknown
1098 } else {
1099 ReferenceConfidence::Resolved
1100 },
1101 }),
1102 _ => None,
1103 }
1104}
1105
1106fn lineage_node_table(node: &LineageNode) -> Option<String> {
1107 match &node.source {
1108 Expression::Table(table) => Some(table_name(table)),
1109 _ => None,
1110 }
1111}
1112
1113fn fallback_column_references(expression: &Expression, scope: &Scope) -> Vec<ColumnReferenceFact> {
1114 let mut refs = Vec::new();
1115 let source_count = scope.sources.len();
1116 let single_source = if source_count == 1 {
1117 scope.sources.iter().next()
1118 } else {
1119 None
1120 };
1121
1122 for column_expr in expression.find_all(|candidate| matches!(candidate, Expression::Column(_))) {
1123 if let Expression::Column(column) = column_expr {
1124 if column.name.name == "*" {
1125 continue;
1126 }
1127 let source = column
1128 .table
1129 .as_ref()
1130 .and_then(|table| scope.sources.get(&table.name));
1131 let (source_name, source_alias, source_kind, table, confidence) =
1132 if let Some(table_identifier) = &column.table {
1133 if let Some(source) = source {
1134 (
1135 Some(table_identifier.name.clone()),
1136 source.alias.clone(),
1137 source.kind,
1138 source_table_name(source)
1139 .or_else(|| Some(table_identifier.name.clone())),
1140 ReferenceConfidence::Resolved,
1141 )
1142 } else {
1143 (
1144 Some(table_identifier.name.clone()),
1145 None,
1146 SourceKind::Unknown,
1147 Some(table_identifier.name.clone()),
1148 ReferenceConfidence::Unknown,
1149 )
1150 }
1151 } else if let Some((name, source)) = single_source {
1152 (
1153 Some(name.clone()),
1154 source.alias.clone(),
1155 source.kind,
1156 source_table_name(source).or_else(|| Some(name.clone())),
1157 ReferenceConfidence::Resolved,
1158 )
1159 } else if source_count > 1 {
1160 (
1161 None,
1162 None,
1163 SourceKind::Unknown,
1164 None,
1165 ReferenceConfidence::Ambiguous,
1166 )
1167 } else {
1168 (
1169 None,
1170 None,
1171 SourceKind::Unknown,
1172 None,
1173 ReferenceConfidence::Unknown,
1174 )
1175 };
1176
1177 refs.push(ColumnReferenceFact {
1178 source_name,
1179 source_alias,
1180 source_kind,
1181 table,
1182 column: column.name.name.clone(),
1183 unqualified: column.table.is_none(),
1184 confidence,
1185 });
1186 }
1187 }
1188
1189 dedupe_column_refs(refs)
1190}
1191
1192fn dedupe_column_refs(refs: Vec<ColumnReferenceFact>) -> Vec<ColumnReferenceFact> {
1193 let mut seen = HashSet::new();
1194 let mut deduped = Vec::new();
1195
1196 for reference in refs {
1197 let key = (
1198 reference.source_name.clone(),
1199 reference.source_alias.clone(),
1200 reference.table.clone(),
1201 reference.column.clone(),
1202 format!("{:?}", reference.source_kind),
1203 reference.unqualified,
1204 format!("{:?}", reference.confidence),
1205 );
1206 if seen.insert(key) {
1207 deduped.push(reference);
1208 }
1209 }
1210
1211 deduped
1212}
1213
1214fn relation_facts(
1215 scope: &Scope,
1216 mapping_schema: Option<&crate::schema::MappingSchema>,
1217) -> Vec<RelationFact> {
1218 let mut relations = Vec::new();
1219 let mut seen = HashSet::new();
1220 collect_relation_facts(scope, mapping_schema, &mut seen, &mut relations);
1221
1222 relations.sort_by(|left, right| {
1223 left.name
1224 .cmp(&right.name)
1225 .then_with(|| left.alias.cmp(&right.alias))
1226 });
1227 relations
1228}
1229
1230fn collect_relation_facts(
1231 scope: &Scope,
1232 mapping_schema: Option<&crate::schema::MappingSchema>,
1233 seen: &mut HashSet<String>,
1234 relations: &mut Vec<RelationFact>,
1235) {
1236 for relation in scope.sources.iter().map(|(source_name, source)| {
1237 let identity = source_table_identity(source);
1238 RelationFact {
1239 name: source
1240 .lineage_name
1241 .clone()
1242 .or_else(|| identity.as_ref().map(|identity| identity.name.clone()))
1243 .unwrap_or_else(|| source_name.clone()),
1244 alias: source.alias.clone().or_else(|| source_alias(source)),
1245 kind: source.kind,
1246 columns: source_columns(source, mapping_schema),
1247 catalog: identity
1248 .as_ref()
1249 .and_then(|identity| identity.catalog.clone()),
1250 schema: identity
1251 .as_ref()
1252 .and_then(|identity| identity.schema.clone()),
1253 table: identity
1254 .as_ref()
1255 .and_then(|identity| identity.table.clone()),
1256 }
1257 }) {
1258 let key = format!("{:?}|{}|{:?}", relation.kind, relation.name, relation.alias);
1259 if seen.insert(key) {
1260 relations.push(relation);
1261 }
1262 }
1263
1264 for branch_scope in &scope.union_scopes {
1265 collect_relation_facts(branch_scope, mapping_schema, seen, relations);
1266 }
1267}
1268
1269fn base_table_facts(
1270 scope: &Scope,
1271 mapping_schema: Option<&crate::schema::MappingSchema>,
1272) -> Vec<RelationFact> {
1273 let mut relations = Vec::new();
1274 let mut seen = HashSet::new();
1275
1276 collect_base_table_facts(scope, mapping_schema, &mut seen, &mut relations);
1277
1278 relations.sort_by(|left, right| left.name.cmp(&right.name));
1279 relations
1280}
1281
1282fn collect_base_table_facts(
1283 scope: &Scope,
1284 mapping_schema: Option<&crate::schema::MappingSchema>,
1285 seen: &mut HashSet<String>,
1286 relations: &mut Vec<RelationFact>,
1287) {
1288 for source in scope.sources.values() {
1289 if source.kind != SourceKind::Table {
1290 continue;
1291 }
1292
1293 let Some(identity) = source_table_identity(source) else {
1294 continue;
1295 };
1296
1297 if seen.insert(identity.name.clone()) {
1298 relations.push(RelationFact {
1299 name: identity.name,
1300 alias: source.alias.clone().or_else(|| source_alias(source)),
1301 kind: SourceKind::Table,
1302 columns: source_columns(source, mapping_schema),
1303 catalog: identity.catalog,
1304 schema: identity.schema,
1305 table: identity.table,
1306 });
1307 }
1308 }
1309
1310 for child_scope in scope
1311 .cte_scopes
1312 .iter()
1313 .chain(scope.union_scopes.iter())
1314 .chain(scope.table_scopes.iter())
1315 .chain(scope.derived_table_scopes.iter())
1316 .chain(scope.subquery_scopes.iter())
1317 {
1318 collect_base_table_facts(child_scope, mapping_schema, seen, relations);
1319 }
1320}
1321
1322fn source_columns(
1323 source: &SourceInfo,
1324 mapping_schema: Option<&crate::schema::MappingSchema>,
1325) -> Vec<String> {
1326 match &source.expression {
1327 Expression::Table(table) => mapping_schema
1328 .and_then(|schema| schema.column_names(&table_name(table)).ok())
1329 .unwrap_or_default(),
1330 Expression::Select(_)
1331 | Expression::Union(_)
1332 | Expression::Intersect(_)
1333 | Expression::Except(_) => get_output_column_names(&source.expression),
1334 Expression::Subquery(subquery) => get_output_column_names(&subquery.this),
1335 Expression::Cte(cte) if !cte.columns.is_empty() => cte
1336 .columns
1337 .iter()
1338 .map(|column| column.name.clone())
1339 .collect(),
1340 Expression::Cte(cte) => get_output_column_names(&cte.this),
1341 _ => Vec::new(),
1342 }
1343}
1344
1345fn source_table_name(source: &SourceInfo) -> Option<String> {
1346 source_table_identity(source).map(|identity| identity.name)
1347}
1348
1349fn source_alias(source: &SourceInfo) -> Option<String> {
1350 match &source.expression {
1351 Expression::Table(table) => table.alias.as_ref().map(|alias| alias.name.clone()),
1352 Expression::Subquery(subquery) => subquery.alias.as_ref().map(|alias| alias.name.clone()),
1353 _ => None,
1354 }
1355}
1356
1357fn table_name(table: &TableRef) -> String {
1358 let mut parts = Vec::new();
1359 if let Some(catalog) = &table.catalog {
1360 parts.push(catalog.name.clone());
1361 }
1362 if let Some(schema) = &table.schema {
1363 parts.push(schema.name.clone());
1364 }
1365 parts.push(table.name.name.clone());
1366 parts.join(".")
1367}
1368
1369#[derive(Debug, Clone)]
1370struct RelationIdentity {
1371 name: String,
1372 catalog: Option<String>,
1373 schema: Option<String>,
1374 table: Option<String>,
1375}
1376
1377fn source_table_identity(source: &SourceInfo) -> Option<RelationIdentity> {
1378 match &source.expression {
1379 Expression::Table(table) => Some(table_identity(table)),
1380 _ => None,
1381 }
1382}
1383
1384fn table_identity(table: &TableRef) -> RelationIdentity {
1385 RelationIdentity {
1386 name: table_name(table),
1387 catalog: table.catalog.as_ref().map(|catalog| catalog.name.clone()),
1388 schema: table.schema.as_ref().map(|schema| schema.name.clone()),
1389 table: Some(table.name.name.clone()),
1390 }
1391}
1392
1393fn set_operation_facts(
1394 expression: &Expression,
1395 scope: &Scope,
1396 dialect: DialectType,
1397) -> Vec<SetOperationFact> {
1398 let mut facts = Vec::new();
1399 collect_set_operation_facts(expression, scope, dialect, &mut facts);
1400 facts
1401}
1402
1403fn collect_set_operation_facts(
1404 expression: &Expression,
1405 scope: &Scope,
1406 dialect: DialectType,
1407 facts: &mut Vec<SetOperationFact>,
1408) {
1409 match expression {
1410 Expression::Union(union) => {
1411 facts.push(SetOperationFact {
1412 kind: "union".to_string(),
1413 all: union.all,
1414 distinct: union.distinct,
1415 output_columns: get_output_column_names(expression),
1416 branches: set_operation_branches(&union.left, &union.right, scope, dialect),
1417 });
1418 collect_set_operation_facts(&union.left, scope, dialect, facts);
1419 collect_set_operation_facts(&union.right, scope, dialect, facts);
1420 }
1421 Expression::Intersect(intersect) => {
1422 facts.push(SetOperationFact {
1423 kind: "intersect".to_string(),
1424 all: intersect.all,
1425 distinct: intersect.distinct,
1426 output_columns: get_output_column_names(expression),
1427 branches: set_operation_branches(&intersect.left, &intersect.right, scope, dialect),
1428 });
1429 collect_set_operation_facts(&intersect.left, scope, dialect, facts);
1430 collect_set_operation_facts(&intersect.right, scope, dialect, facts);
1431 }
1432 Expression::Except(except) => {
1433 facts.push(SetOperationFact {
1434 kind: "except".to_string(),
1435 all: except.all,
1436 distinct: except.distinct,
1437 output_columns: get_output_column_names(expression),
1438 branches: set_operation_branches(&except.left, &except.right, scope, dialect),
1439 });
1440 collect_set_operation_facts(&except.left, scope, dialect, facts);
1441 collect_set_operation_facts(&except.right, scope, dialect, facts);
1442 }
1443 Expression::Subquery(subquery) => {
1444 collect_set_operation_facts(&subquery.this, scope, dialect, facts);
1445 }
1446 _ => {}
1447 }
1448}
1449
1450fn set_operation_branches(
1451 left: &Expression,
1452 right: &Expression,
1453 scope: &Scope,
1454 dialect: DialectType,
1455) -> Vec<SetOperationBranchFact> {
1456 vec![
1457 SetOperationBranchFact {
1458 index: 0,
1459 projections: projection_facts_for_branch(left, scope, dialect),
1460 },
1461 SetOperationBranchFact {
1462 index: 1,
1463 projections: projection_facts_for_branch(right, scope, dialect),
1464 },
1465 ]
1466}
1467
1468fn projection_facts_for_branch(
1469 expression: &Expression,
1470 root_scope: &Scope,
1471 dialect: DialectType,
1472) -> Vec<ProjectionFact> {
1473 let branch_scope = build_scope(expression);
1474 let scope = if branch_scope.sources.is_empty() {
1475 root_scope
1476 } else {
1477 &branch_scope
1478 };
1479 let nullability_context = NullabilityContext {
1480 schema: None,
1481 nullable_sources: nullable_source_names(expression),
1482 };
1483 projection_facts_for_query(expression, scope, dialect, &nullability_context)
1484}
1485
1486fn non_empty_string(value: String) -> Option<String> {
1487 if value.is_empty() {
1488 None
1489 } else {
1490 Some(value)
1491 }
1492}