From e404444949b5d8b06ff8beb08a9f26367862fbe3 Mon Sep 17 00:00:00 2001 From: William-Datadorelix <80535383+IL-William@users.noreply.github.com> Date: Tue, 29 Sep 2026 11:29:41 +0200 Subject: [PATCH] Fix lineage through LATERAL FLATTEN, reused derived aliases, predicate subqueries and star options (#2) * fix(core): read LATERAL FLATTEN as a relation with its own columns sqlparser gives `LATERAL FLATTEN(...)` its own table factor, `TableFactor::Function`, which the lineage visitor did not match. The alias was never registered and no issue was raised, so a column read through it lost its source: SELECT o.order_id, f.value::STRING AS tag FROM orders o, LATERAL FLATTEN(input => o.tags) f `f` was taken for a table name: tag came out derived from a column `value` of a table `F` that does not exist, the implied schema listed `F` beside ORDERS, and orders.tags was read neither for lineage nor for the implied schema. An unqualified `value` was reported as ambiguous across the tables in scope, and a FLATTEN over the value of another had nothing to follow. Register FLATTEN under its alias, or under its name when it has none, as a relation with the columns Snowflake documents: SEQ, KEY, PATH, INDEX, VALUE and THIS. VALUE and THIS carry the element, so they derive from the columns of the `INPUT =>` argument, or of the first positional one, with the call as their expression. SEQ, KEY, PATH and INDEX say where an element sits and read nothing. They are added without the relation-level edge a source-less projection gets, which would have made `'n' || f.index` depend on every base relation of the FROM clause. Each FLATTEN gets its own node, keyed by its scope, since CTEs commonly reuse one alias such as `f`. A star over the scope now includes the six columns. The alias is also recorded as a subquery alias, as a derived table's is, so that what is read through it stays out of the implied schema, and the qualified columns of the input are recorded there as projected columns are: the query above now implies ORDERS(order_id, tags) and nothing else. With column lineage disabled only that alias is recorded: at table level the input is a column of a relation already in the FROM clause, so a FLATTEN node would have no edge, and no column is emitted. Those six columns are all a FLATTEN has, so the scope records it as a relation with fixed columns. A bare name it lacks, with no schema to place it, still resolves to the one other relation of the FROM clause, as it did while FLATTEN went unregistered: `id` in `SELECT id FROM orders o, LATERAL FLATTEN(input => o.tags) f` stays orders.id instead of turning ambiguous, and in `FROM customers, LATERAL FLATTEN(input => addresses) a, LATERAL FLATTEN(input => phones) p` the second input is read from customers. A warning for a name that is ambiguous between tables no longer lists the FLATTEN beside them. VALUE and THIS read the same input, so an input that cannot be resolved is reported once. Any other `LATERAL (...)` now raises UNSUPPORTED_SYNTAX, as `TABLE((...))` already does, instead of being dropped silently. The snowflake_lateral_flatten snapshot changes accordingly: `value AS p_id` now derives from b.cool_ids through the FLATTEN node, B's implied columns gain cool_ids, and the ambiguity warning for `value` is gone. Tests: the query above with a decoy `value` column on orders, its implied schema without a schema given, chained flattens back to the first input, the position columns reading nothing under a CTE joined to another table, a literal input, one alias reused by two CTEs with named and positional input, a star over FLATTEN, a bare column and a bare filter beside a FLATTEN with no schema, a bare input to a second FLATTEN, an ambiguous input reported once, SPLIT_TO_TABLE reported as unsupported, and no FLATTEN node or column with column lineage disabled. Co-Authored-By: Claude Opus 5.5 (1M context) * fix(core): give each derived table its own node when an alias repeats A derived table's node was keyed by statement and alias. Two derived tables of one statement sharing an alias in different scopes, as a generated point in time query writes one `(...) AS v` per satellite, therefore shared one node and one set of column nodes, and each consumer received every producer's lineage: WITH a_intervals AS ( SELECT k, ts FROM (SELECT DISTINCT k, ts FROM sat_a) AS v ), b_intervals AS ( SELECT k, ts FROM (SELECT DISTINCT k, ts FROM sat_b) AS v ) SELECT a.ts AS a_ts, b.ts AS b_ts FROM a_intervals a JOIN b_intervals b ON a.k = b.k a_ts came out derived from sat_a.ts and sat_b.ts both, and so did b_ts, with no issue raised. Count the derived tables of each alias per statement. The first keeps the key it always had, so nothing changes for a statement that does not repeat an alias, and each later one is keyed `alias#n`, getting a node and column nodes of its own. The node keeps the alias as its label. Two derived tables of one alias in the same scope are invalid SQL, and lookups already go through the scope, so a per statement count is enough. A CTE name defined again in a nested WITH merges the same way, but CTEs are also looked up by name once the statement has been walked, so that takes more than a key and is left alone here. Tests: the query above, each output from its own satellite and two nodes labelled v; the same alias over twelve satellites, each output from its own. Both fail without the change. Co-Authored-By: Claude Opus 5.5 (1M context) * fix(core): keep a predicate's subquery and an unaliased derived table out of the outputs A subquery a predicate reads was analysed with no target node, so its select list fell back to whatever the enclosing select targets: the statement's output at the top, and inside a CTE that CTE's column list, which a star over it then returned. SELECT k FROM t WHERE x >= (SELECT MAX(y) AS floor_y FROM u) returned k and floor_y. The same held for IN, for EXISTS, whose `SELECT 1` came back as col_0, and for a subquery in HAVING, a join condition or a grouping expression. No issue said so. Such a subquery is now analysed against a node of its own, labelled `(subquery)` and keyed per occurrence, which owns its columns and their lineage, and what it projects is taken back from the projection buffer once it has been read. A scalar subquery in the select list is not visited there and is returned as before. A derived table without an alias, which Snowflake and others accept, had no node at all: its projection hung on the enclosing target in the same way, and the outer select found no table in scope and reported so. It now gets a node like an aliased one, registered in its scope under a name no query can write, `(derived)`, so that a star over it expands and a bare column resolves. Tests: a scalar comparison, IN, a correlated NOT EXISTS, HAVING and a join condition each return only k; inside a CTE the subquery's column stays on its own node and out of the star; a select list scalar subquery is still returned; an unaliased derived table reads like an aliased one, with no issue. All but the select list case fail without the change. Co-Authored-By: Claude Opus 5.5 (1M context) * fix(core): honour EXCLUDE, EXCEPT and RENAME in a wildcard's expansion expand_wildcard emitted every column of the relation whatever options the wildcard carried: `SELECT * EXCLUDE (b) FROM p` returned b, and `SELECT * RENAME (b AS bee) FROM p` returned b under its old name. The common idiom of replacing a column fared worse: SELECT * EXCLUDE (b), UPPER(b) AS b FROM p The star copy of b and the computed b shared one output node, the copy came first, and the UPPER(b) derivation was merged away: b came out as a plain copy of p.b. The wildcard's options now reach the expansion. A column named by EXCLUDE, or by EXCEPT where the dialect writes it so, is not expanded; a column named by RENAME is expanded under its new name, still read from the old one. Names are compared as the expanded columns' own names are. REPLACE and ILIKE are still not read. A plain `SELECT *, f(x) AS x`, which does write two columns named x, is left as it was. Tests: EXCLUDE with and without parentheses, qualified and inside a CTE; EXCEPT; RENAME with and without parentheses, read from the old column; the idiom above, whose b carries UPPER(b). All fail without the change. Co-Authored-By: Claude Opus 5.5 (1M context) * docs(changelog): note the relation and scope lineage fixes Co-Authored-By: Claude Opus 5.5 (1M context) --------- Co-authored-by: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 + crates/flowscope-core/src/analyzer/context.rs | 32 + .../flowscope-core/src/analyzer/expression.rs | 39 +- crates/flowscope-core/src/analyzer/query.rs | 101 ++- .../src/analyzer/select_analyzer.rs | 13 +- crates/flowscope-core/src/analyzer/visitor.rs | 207 +++++- crates/flowscope-core/tests/lineage_engine.rs | 674 ++++++++++++++++++ ...apshot_test@snowflake_lateral_flatten.snap | 209 +++++- 8 files changed, 1247 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe5380d7..ce95e924 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Walked left-deep chains of binary operators as chains, so that a generated surrogate key over more than about 100 operands keeps every operand in its lineage instead of losing the leftmost ones and reporting `APPROXIMATE_LINEAGE`. - Read columns inside TRIM, SUBSTRING, POSITION, CEIL and FLOOR, AT TIME ZONE, IS [NOT] DISTINCT FROM, path and subscript accessors, COLLATE, OVERLAY, SIMILAR TO, RLIKE, ANY and ALL as lineage sources. +- Read `LATERAL FLATTEN` as a relation with the SEQ, KEY, PATH, INDEX, VALUE and THIS columns, VALUE and THIS derived from its input, instead of as a table named after its alias. +- Kept two derived tables that share an alias in one statement apart, so that each consumer receives its own derived table's lineage. +- Kept what a subquery read by a predicate, a join condition or a grouping expression projects out of the statement's outputs, and gave a derived table without an alias a node of its own. +- Honoured `* EXCLUDE`, `* EXCEPT` and `* RENAME` in wildcard expansion. ## [0.9.2] - 2026-09-24 diff --git a/crates/flowscope-core/src/analyzer/context.rs b/crates/flowscope-core/src/analyzer/context.rs index 68cd4a62..c1b89b3a 100644 --- a/crates/flowscope-core/src/analyzer/context.rs +++ b/crates/flowscope-core/src/analyzer/context.rs @@ -58,6 +58,9 @@ pub(crate) struct Scope { /// True when the scope contains a table function relation whose output /// columns may be dialect-provided rather than schema-backed. pub(crate) has_table_function_relation: bool, + /// Node IDs of relations whose columns are all known, such as a FLATTEN. + /// An unqualified column missing from them belongs to another relation. + pub(crate) fixed_column_relations: HashSet>, } impl Scope { @@ -137,6 +140,13 @@ pub(crate) struct StatementContext { pub(crate) table_aliases: HashMap, /// Subquery aliases (for reference tracking) pub(crate) subquery_aliases: HashSet, + /// How many derived tables of each alias the statement has opened, so that + /// a derived table reusing the alias of an earlier one in another scope + /// gets a node of its own. + pub(crate) derived_occurrences: HashMap, + /// How many subqueries a predicate reads the statement has opened, each + /// analysed against a node of its own. + pub(crate) predicate_subqueries: usize, /// Last join/operation type for edge labeling pub(crate) last_operation: Option, /// Current join information (type + condition) for edge labeling @@ -232,6 +242,8 @@ impl StatementContext { relation_span_cursors: HashMap::new(), table_aliases: HashMap::new(), subquery_aliases: HashSet::new(), + derived_occurrences: HashMap::new(), + predicate_subqueries: 0, last_operation: None, current_join_info: JoinInfo::default(), table_node_ids: HashMap::new(), @@ -676,6 +688,26 @@ impl StatementContext { .is_some_and(|scope| scope.has_table_function_relation) } + /// Mark a relation of the current scope as having no columns beyond those + /// registered for it. + pub(crate) fn mark_fixed_columns_in_scope(&mut self, node_id: Arc) { + if let Some(scope) = self.current_scope_mut() { + scope.fixed_column_relations.insert(node_id); + } + } + + /// Relation instances of the current scope that may have a column missing + /// from every known column list, that is all but those with fixed columns. + pub(crate) fn open_relation_instances_in_current_scope(&self) -> Vec { + let Some(scope) = self.current_scope() else { + return Vec::new(); + }; + self.relation_instances_in_current_scope() + .into_iter() + .filter(|instance| !scope.fixed_column_relations.contains(&instance.node_id)) + .collect() + } + /// Register statement-global output columns for a named CTE definition. pub(crate) fn register_cte_output_columns(&mut self, name: String, columns: Vec) { self.aliased_subquery_columns.insert(name, columns); diff --git a/crates/flowscope-core/src/analyzer/expression.rs b/crates/flowscope-core/src/analyzer/expression.rs index aff1f2e9..e912741b 100644 --- a/crates/flowscope-core/src/analyzer/expression.rs +++ b/crates/flowscope-core/src/analyzer/expression.rs @@ -13,10 +13,10 @@ use super::context::{ColumnRef, StatementContext}; use super::functions; -use super::helpers::check_expr_types; +use super::helpers::{check_expr_types, generate_statement_scoped_node_id}; use super::Analyzer; use crate::generated; -use crate::types::{AggregationInfo, FilterClauseType}; +use crate::types::{AggregationInfo, FilterClauseType, Node, NodeType}; use crate::Dialect; use sqlparser::ast::{self, Expr, FunctionArg, FunctionArgExpr}; use std::collections::HashSet; @@ -24,6 +24,10 @@ use std::sync::Arc; #[cfg(feature = "tracing")] use tracing::debug; +/// The label of the node a subquery read by a predicate is analyzed against. A +/// query cannot write it, so nothing can qualify a column with it. +const PREDICATE_SUBQUERY: &str = "(subquery)"; + /// Maximum recursion depth for expression traversal to prevent stack overflow /// on maliciously crafted or deeply nested SQL expressions. pub(super) const MAX_RECURSION_DEPTH: usize = 100; @@ -135,6 +139,29 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { } } + /// Analyzes a subquery an expression other than a projection reads: a + /// predicate compares with it or tests it for rows, and a join condition or + /// a grouping expression reads it the same way. What it projects is never + /// returned, so it is analyzed against a node of its own, labelled + /// `(subquery)`, and its columns are taken back once it has been read. + /// Analyzed with no target, they would hang on whatever the enclosing + /// select targets, the statement's output at the top, as if the query + /// returned them, and would join the column list of an enclosing CTE. + fn analyze_predicate_subquery(&mut self, query: &ast::Query) { + let checkpoint = self.ctx.projection_checkpoint(); + self.ctx.predicate_subqueries += 1; + let key = format!("{PREDICATE_SUBQUERY}#{}", self.ctx.predicate_subqueries); + let node_id = self.ctx.add_node(Node { + id: generate_statement_scoped_node_id("subquery", self.ctx.statement_index, &key), + node_type: NodeType::Cte, + label: PREDICATE_SUBQUERY.into(), + qualified_name: Some(PREDICATE_SUBQUERY.into()), + ..Default::default() + }); + self.analyzer.analyze_query(self.ctx, query, Some(&node_id)); + self.ctx.take_output_columns_since(checkpoint); + } + /// Recursively visits an expression to find and analyze subqueries. /// /// The `depth` parameter tracks recursion depth to prevent stack overflow @@ -153,11 +180,9 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { let next_depth = depth + 1; match expr { - Expr::Subquery(query) => self.analyzer.analyze_query(self.ctx, query, None), - Expr::InSubquery { subquery, .. } => { - self.analyzer.analyze_query(self.ctx, subquery, None) - } - Expr::Exists { subquery, .. } => self.analyzer.analyze_query(self.ctx, subquery, None), + Expr::Subquery(query) => self.analyze_predicate_subquery(query), + Expr::InSubquery { subquery, .. } => self.analyze_predicate_subquery(subquery), + Expr::Exists { subquery, .. } => self.analyze_predicate_subquery(subquery), Expr::BinaryOp { .. } => { for operand in binary_op_operands(expr) { self.visit_expression_for_subqueries(operand, next_depth); diff --git a/crates/flowscope-core/src/analyzer/query.rs b/crates/flowscope-core/src/analyzer/query.rs index 4569bca6..9a233405 100644 --- a/crates/flowscope-core/src/analyzer/query.rs +++ b/crates/flowscope-core/src/analyzer/query.rs @@ -555,7 +555,47 @@ impl<'a> Analyzer<'a> { ctx: &mut StatementContext, table_qualifier: Option<&str>, target_node: Option<&str>, + options: &ast::WildcardAdditionalOptions, ) { + // What `* EXCLUDE (...)` or `* EXCEPT (...)` leaves out, and what + // `* RENAME (old AS new)` returns under another name, compared as the + // expanded columns' own names are. REPLACE and ILIKE are not read. + let mut excluded: HashSet = HashSet::new(); + match &options.opt_exclude { + Some(ast::ExcludeSelectItem::Single(ident)) => { + excluded.insert(self.normalize_identifier(&ident.to_string())); + } + Some(ast::ExcludeSelectItem::Multiple(idents)) => { + excluded.extend( + idents + .iter() + .map(|ident| self.normalize_identifier(&ident.to_string())), + ); + } + None => {} + } + if let Some(except) = &options.opt_except { + excluded.extend( + std::iter::once(&except.first_element) + .chain(&except.additional_elements) + .map(|ident| self.normalize_identifier(&ident.to_string())), + ); + } + let renames: Vec<&ast::IdentWithAlias> = match &options.opt_rename { + Some(ast::RenameSelectItem::Single(rename)) => vec![rename], + Some(ast::RenameSelectItem::Multiple(renames)) => renames.iter().collect(), + None => Vec::new(), + }; + let renamed: HashMap = renames + .into_iter() + .map(|rename| { + ( + self.normalize_identifier(&rename.ident.to_string()), + rename.alias.value.clone(), + ) + }) + .collect(); + // Resolve wildcard sources as (canonical, qualifier) pairs so repeated // relation instances in self-joins are expanded independently. let tables_to_expand: Vec<(String, String)> = if let Some(qualifier) = table_qualifier { @@ -615,13 +655,21 @@ impl<'a> Analyzer<'a> { if let Some(columns) = columns_to_add { // Expand from schema - NOT approximate. for col_info in columns { + let normalized = self.normalize_identifier(&col_info.name); + if excluded.contains(&normalized) { + continue; + } + let name = renamed + .get(&normalized) + .cloned() + .unwrap_or_else(|| col_info.name.clone()); let sources = vec![ColumnRef { table: Some(source_qualifier.clone()), column: col_info.name.clone(), }]; self.add_output_column( ctx, - &col_info.name, + &name, sources, None, col_info.data_type, @@ -865,13 +913,20 @@ impl<'a> Analyzer<'a> { } } 0 => { - // No candidates found - if there's only one relation instance in scope, use it - // (the column might exist but not be in our schema) - if relation_instances.len() == 1 { - return Some(relation_instances[0].canonical.clone()); + // No candidates found - if only one relation instance in scope may have + // columns we do not know of, use it (the column might exist but not be in + // our schema). A FLATTEN has none, so it cannot own the column. + let owners = ctx.open_relation_instances_in_current_scope(); + if owners.len() == 1 { + return Some(owners[0].canonical.clone()); } // Multiple tables but column not found in any - ambiguous - let mut sorted_tables: Vec<_> = relation_instances + let listed = if owners.is_empty() { + &relation_instances + } else { + &owners + }; + let mut sorted_tables: Vec<_> = listed .iter() .map(|instance| instance.canonical.clone()) .collect(); @@ -1212,6 +1267,40 @@ impl<'a> Analyzer<'a> { }); } + /// Adds a column that `target_node` produces without reading any column, + /// such as the position of an element in a flattened array. + /// + /// Unlike a source-less projection, the column is not attributed to the + /// relations in scope: it describes the relation that owns it. + pub(super) fn add_unsourced_column( + &mut self, + ctx: &mut StatementContext, + name: &str, + target_node: &Arc, + ) { + let normalized_name = self.normalize_identifier(name); + let node_id = generate_column_node_id(Some(target_node), &normalized_name); + ctx.add_node(Node { + id: node_id.clone(), + node_type: NodeType::Column, + label: normalized_name.clone().into(), + ..Default::default() + }); + let edge_id = generate_edge_id(target_node, &node_id); + if !ctx.edge_ids.contains(&edge_id) { + ctx.add_edge(Edge::ownership( + edge_id, + target_node.clone(), + node_id.clone(), + )); + } + ctx.output_columns.push(OutputColumn { + name: normalized_name, + data_type: None, + node_id, + }); + } + fn is_definitely_ambiguous_unqualified_column( &self, ctx: &StatementContext, diff --git a/crates/flowscope-core/src/analyzer/select_analyzer.rs b/crates/flowscope-core/src/analyzer/select_analyzer.rs index a1b55bfe..a14d5856 100644 --- a/crates/flowscope-core/src/analyzer/select_analyzer.rs +++ b/crates/flowscope-core/src/analyzer/select_analyzer.rs @@ -230,7 +230,7 @@ impl<'a, 'b> SelectAnalyzer<'a, 'b> { }, ); } - SelectItem::QualifiedWildcard(name, _) => { + SelectItem::QualifiedWildcard(name, options) => { let qualifier = name.to_string(); // SelectItemQualifiedWildcardKind::Display appends ".*" let qualifier = qualifier.strip_suffix(".*").unwrap_or(&qualifier); @@ -238,11 +238,16 @@ impl<'a, 'b> SelectAnalyzer<'a, 'b> { self.ctx, Some(qualifier), self.target_node.as_deref(), + options, ); } - SelectItem::Wildcard(_) => { - self.analyzer - .expand_wildcard(self.ctx, None, self.target_node.as_deref()); + SelectItem::Wildcard(options) => { + self.analyzer.expand_wildcard( + self.ctx, + None, + self.target_node.as_deref(), + options, + ); } } } diff --git a/crates/flowscope-core/src/analyzer/visitor.rs b/crates/flowscope-core/src/analyzer/visitor.rs index 13ccbb31..89e1a2b9 100644 --- a/crates/flowscope-core/src/analyzer/visitor.rs +++ b/crates/flowscope-core/src/analyzer/visitor.rs @@ -10,6 +10,7 @@ use super::helpers::{ alias_visibility_warning, find_cte_body_span, find_cte_definition_span, find_derived_table_alias_span, generate_statement_scoped_node_id, }; +use super::query::OutputColumnParams; use super::select_analyzer::SelectAnalyzer; use super::Analyzer; use crate::generated::is_value_table_function; @@ -20,6 +21,9 @@ use sqlparser::ast::{ }; use std::sync::Arc; +/// The name a derived table without an alias is registered under in its scope. +const UNALIASED_DERIVED_TABLE: &str = "(derived)"; + /// A visitor trait for traversing the SQL AST. /// /// This trait defines default behavior for visiting nodes (traversing children). @@ -465,6 +469,106 @@ impl<'a, 'b> LineageVisitor<'a, 'b> { } } + /// Registers Snowflake's `LATERAL FLATTEN(INPUT => expr) AS f` as a + /// relation with the columns FLATTEN returns, so that `f.value` resolves. + /// + /// VALUE and THIS carry the input, so they derive from the columns it + /// reads, with the whole call as their expression. SEQ, KEY, PATH and + /// INDEX say where an element sits rather than carry it, and read nothing. + fn visit_flatten( + &mut self, + name: &ast::ObjectName, + args: &[ast::FunctionArg], + alias: Option<&TableAlias>, + ) { + // Unaliased, the columns are still read unqualified (`value`), so the + // relation needs a name all the same. + let alias_name = alias.map_or_else(|| name.to_string(), |a| a.name.to_string()); + // What is read through the alias belongs to no table, so it stays out + // of the implied schema, as a derived table's columns do. + self.ctx + .register_subquery_alias_in_scope(alias_name.clone()); + // At table level the input is a column of a relation already in the + // FROM clause: a FLATTEN node would have nothing to connect to. + if !self.analyzer.column_lineage_enabled { + return; + } + // An alias such as `f` is often reused by every CTE that flattens, + // and each use is a different relation. + let scope_id = self.ctx.current_scope_id().unwrap_or_default(); + let node_id = self.ctx.add_node(Node { + id: generate_statement_scoped_node_id( + "flatten", + self.ctx.statement_index, + &format!("scope_{scope_id}::{alias_name}"), + ), + node_type: NodeType::Cte, + label: alias_name.clone().into(), + qualified_name: Some(alias_name.clone().into()), + ..Default::default() + }); + + let sources = match flatten_input(args) { + Some(input) => ExpressionAnalyzer::new(self.analyzer, self.ctx) + .extract_column_refs_with_warning(input), + None => Vec::new(), + }; + // The input is read as a projected column is, so its table learns of it. + for source in &sources { + let table = source.table.as_deref(); + if let Some(canonical) = table.and_then(|t| self.resolve_table_alias(Some(t))) { + self.ctx + .record_source_column(&canonical, &source.column, None); + } + } + let call = format!( + "{name}({})", + args.iter() + .map(ToString::to_string) + .collect::>() + .join(", ") + ); + + let checkpoint = self.ctx.projection_checkpoint(); + for column in ["SEQ", "KEY", "PATH", "INDEX", "VALUE", "THIS"] { + let carries_input = matches!(column, "VALUE" | "THIS"); + if carries_input && !sources.is_empty() { + let reported = self.analyzer.issues.len(); + self.analyzer.add_output_column_with_aggregation( + self.ctx, + OutputColumnParams { + name: column.to_string(), + sources: sources.clone(), + expression: Some(call.clone()), + data_type: None, + target_node: Some(node_id.to_string()), + approximate: false, + aggregation: None, + }, + ); + // THIS reads the input VALUE has just read, so whatever resolving + // it reports has been reported once already. + if column == "THIS" { + self.analyzer.issues.truncate(reported); + } + } else { + self.analyzer + .add_unsourced_column(self.ctx, column, &node_id); + } + } + let columns = self.ctx.take_output_columns_since(checkpoint); + + // The six columns are all FLATTEN returns: a bare name it lacks + // belongs to another relation of the FROM clause. + self.ctx.mark_fixed_columns_in_scope(node_id.clone()); + self.ctx + .register_table_in_scope(alias_name.clone(), node_id); + self.ctx + .register_alias_in_scope(alias_name.clone(), alias_name.clone()); + self.ctx + .register_subquery_columns_in_scope(alias_name, columns); + } + /// Emits a warning for unsupported alias usage in a clause. fn emit_alias_warning(&mut self, clause_name: &str, alias_name: &str) { let dialect = self.analyzer.request.dialect; @@ -674,21 +778,46 @@ impl<'a, 'b> Visitor for LineageVisitor<'a, 'b> { // We create a node for it in the graph, analyze its subquery to determine its // output columns, and then register its alias and columns in the current scope // so the outer query can reference it. - let alias_name = alias.as_ref().map(|a| a.name.to_string()); let projection_checkpoint = self.ctx.projection_checkpoint(); - let derived_span = alias_name + let derived_span = alias .as_ref() - .and_then(|name| self.locate_derived_alias_span(name)); + .and_then(|a| self.locate_derived_alias_span(&a.name.to_string())); + // A derived table without an alias, which Snowflake and others + // accept, is a relation the outer query reads all the same. + // Without a node its projection would hang on whatever the + // enclosing select targets, the statement's output at the top, and + // the outer select would find no table in scope. It is named so + // that no query can qualify a column with it. + let alias_name = Some(alias.as_ref().map_or_else( + || UNALIASED_DERIVED_TABLE.to_string(), + |a| a.name.to_string(), + )); // We model derived tables as CTEs in the graph since they are conceptually // similar: both are ephemeral, named result sets scoped to a single query. // This avoids introducing a separate NodeType for a very similar concept. + // + // Two derived tables of one statement can share an alias in + // different scopes, as `(...) AS v` in two CTEs. Keyed by the alias + // alone, the second would find the first's node and its columns + // would feed both. The first keeps the plain key, each later one + // gets its own. let derived_node_id = alias_name.as_ref().map(|name| { + let occurrence = self + .ctx + .derived_occurrences + .entry(name.clone()) + .or_insert(0); + *occurrence += 1; + let key = match *occurrence { + 1 => name.clone(), + n => format!("{name}#{n}"), + }; self.ctx.add_node(Node { id: generate_statement_scoped_node_id( "derived", self.ctx.statement_index, - name, + &key, ), node_type: NodeType::Cte, label: name.clone().into(), @@ -749,6 +878,34 @@ impl<'a, 'b> Visitor for LineageVisitor<'a, 'b> { .with_statement(self.ctx.statement_index), ); } + TableFactor::Function { + name, args, alias, .. + } if name.to_string().eq_ignore_ascii_case("flatten") => { + self.visit_flatten(name, args, alias.as_ref()); + } + TableFactor::Function { + name, args, alias, .. + } => { + for expr in args.iter().filter_map(function_arg_expr) { + self.extract_identifiers_from_expr(expr); + } + if is_value_table_function(self.analyzer.request.dialect, &name.to_string()) { + self.ctx.mark_table_function_in_scope(); + } + if let Some(a) = alias { + self.ctx + .register_subquery_alias_in_scope(a.name.to_string()); + } + self.analyzer.issues.push( + Issue::info( + issue_codes::UNSUPPORTED_SYNTAX, + format!( + "Table function '{name}' lineage extracted with best-effort identifier matching" + ), + ) + .with_statement(self.ctx.statement_index), + ); + } TableFactor::Pivot { table, aggregate_functions, @@ -846,3 +1003,45 @@ impl<'a, 'b> Visitor for LineageVisitor<'a, 'b> { } } } + +/// The expression passed to a table function argument, named or not. +fn function_arg_expr(arg: &ast::FunctionArg) -> Option<&Expr> { + match arg { + ast::FunctionArg::Named { + arg: ast::FunctionArgExpr::Expr(expr), + .. + } + | ast::FunctionArg::ExprNamed { + arg: ast::FunctionArgExpr::Expr(expr), + .. + } + | ast::FunctionArg::Unnamed(ast::FunctionArgExpr::Expr(expr)) => Some(expr), + _ => None, + } +} + +/// The value FLATTEN expands: its `INPUT =>` argument, or the first positional +/// one. PATH, OUTER, RECURSIVE and MODE are constants. +fn flatten_input(args: &[ast::FunctionArg]) -> Option<&Expr> { + let named = args.iter().find_map(|arg| { + let name = match arg { + ast::FunctionArg::Named { name, .. } => name, + ast::FunctionArg::ExprNamed { + name: Expr::Identifier(name), + .. + } => name, + _ => return None, + }; + if name.value.eq_ignore_ascii_case("input") { + function_arg_expr(arg) + } else { + None + } + }); + named.or_else(|| { + args.iter().find_map(|arg| match arg { + ast::FunctionArg::Unnamed(ast::FunctionArgExpr::Expr(expr)) => Some(expr), + _ => None, + }) + }) +} diff --git a/crates/flowscope-core/tests/lineage_engine.rs b/crates/flowscope-core/tests/lineage_engine.rs index ae5f31f8..27cea89a 100644 --- a/crates/flowscope-core/tests/lineage_engine.rs +++ b/crates/flowscope-core/tests/lineage_engine.rs @@ -9824,6 +9824,446 @@ fn snowflake_lateral_flatten_tracks_source_table() { ); } +/// Table columns an output column is computed from, as `relation.column`, +/// lowercased, following derivation and data-flow edges through CTEs, +/// derived tables and table functions. +fn base_sources_of(stmt: &StmtView<'_>, output: &str) -> HashSet { + let owner_of = |id: &str| { + stmt.edges + .iter() + .find(|edge| edge.edge_type == EdgeType::Ownership && &*edge.to == id) + .and_then(|edge| stmt.nodes.iter().copied().find(|node| node.id == edge.from)) + }; + let target = stmt + .nodes + .iter() + .find(|node| { + node.node_type == NodeType::Column + && node.label.eq_ignore_ascii_case(output) + && owner_of(&node.id).is_some_and(|owner| owner.node_type == NodeType::Output) + }) + .unwrap_or_else(|| panic!("output column {output} should exist")); + + let mut sources = HashSet::new(); + let mut seen = HashSet::new(); + let mut pending = vec![target.id.clone()]; + while let Some(id) = pending.pop() { + if !seen.insert(id.clone()) { + continue; + } + for edge in stmt.edges.iter().filter(|edge| { + edge.to == id && matches!(edge.edge_type, EdgeType::Derivation | EdgeType::DataFlow) + }) { + let Some(source) = stmt + .nodes + .iter() + .find(|node| node.id == edge.from && node.node_type == NodeType::Column) + else { + continue; + }; + match owner_of(&source.id) { + Some(owner) if owner.node_type == NodeType::Table => { + sources.insert(format!("{}.{}", owner.label, source.label).to_lowercase()); + } + _ => pending.push(source.id.clone()), + } + } + } + sources +} + +/// Edges that carry data into the columns of the relation labelled `alias`, +/// keyed by the lowercased column label. +fn flows_into_columns_of<'a>(stmt: &StmtView<'a>, alias: &str) -> Vec<(String, &'a Edge)> { + let relation = find_cte_node(stmt, alias).unwrap_or_else(|| panic!("{alias} should exist")); + stmt.edges + .iter() + .filter(|owned| owned.edge_type == EdgeType::Ownership && owned.from == relation.id) + .flat_map(|owned| { + let label = stmt + .nodes + .iter() + .find(|node| node.id == owned.to) + .map(|node| node.label.to_lowercase()) + .unwrap_or_default(); + stmt.edges + .iter() + .copied() + .filter(move |edge| { + edge.to == owned.to + && matches!(edge.edge_type, EdgeType::Derivation | EdgeType::DataFlow) + }) + .map(move |edge| (label.clone(), edge)) + }) + .collect() +} + +fn schema_of(tables: Vec) -> SchemaMetadata { + SchemaMetadata { + allow_implied: false, + default_catalog: None, + default_schema: None, + search_path: None, + case_sensitivity: None, + tables, + } +} + +#[test] +fn flatten_value_comes_from_its_input_not_from_a_namesake() { + // `value` is also a column of orders: `f.value` must not resolve to it. + let sql = r#" + SELECT o.order_id, f.value::STRING AS tag + FROM orders o, + LATERAL FLATTEN(input => o.tags) f + "#; + let schema = schema_of(vec![schema_table( + None, + None, + "orders", + &["order_id", "tags", "value"], + )]); + + let result = run_analysis(sql, Dialect::Snowflake, Some(schema)); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "tag"), + expected_sources(&["orders.tags"]) + ); + + let flows = flows_into_columns_of(&stmt, "f"); + let expressions: HashSet<(&str, Option<&str>)> = flows + .iter() + .map(|(column, edge)| (column.as_str(), edge.expression.as_deref())) + .collect(); + assert_eq!( + expressions, + HashSet::from([ + ("value", Some("FLATTEN(input => o.tags)")), + ("this", Some("FLATTEN(input => o.tags)")), + ]) + ); +} + +#[test] +fn flatten_alias_is_not_an_implied_table() { + // Without a schema the tables are implied from what the query reads: + // orders has the flattened tags, and `f` is no table at all. + let sql = r#" + SELECT o.order_id, f.value::STRING AS tag, f.key AS k + FROM orders o, + LATERAL FLATTEN(input => o.tags) f + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let resolved = result + .resolved_schema + .expect("resolved schema should be present"); + let tables: Vec<(String, Vec)> = resolved + .tables + .iter() + .map(|table| { + ( + table.name.to_lowercase(), + table.columns.iter().map(|col| col.name.clone()).collect(), + ) + }) + .collect(); + assert_eq!( + tables, + vec![( + "orders".to_string(), + vec!["order_id".to_string(), "tags".to_string()] + )] + ); +} + +#[test] +fn chained_flattens_reach_the_first_input() { + let sql = r#" + SELECT m.value::STRING AS month + FROM events e, + LATERAL FLATTEN(input => PARSE_JSON(e.payload)) a, + LATERAL FLATTEN(input => SPLIT(a.value:"1"::STRING, ',')) m + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "month"), + expected_sources(&["events.payload"]) + ); +} + +#[test] +fn flatten_position_columns_read_nothing() { + // `f.index` numbers the elements of the array: it is not a value of + // orders, and the join to audit feeds nothing. + let sql = r#" + WITH base AS ( + SELECT o.id, o.a, o.b + FROM orders o + JOIN audit k ON o.id = k.id + ), + opts AS ( + SELECT 'n' || (f.index + 1) AS n, b.* + FROM base b, + LATERAL FLATTEN(input => ARRAY_CONSTRUCT(b.a, b.b)) f + ) + SELECT CONCAT_WS('/', id, n) AS label + FROM opts + "#; + let schema = schema_of(vec![ + schema_table(None, None, "orders", &["id", "a", "b"]), + schema_table(None, None, "audit", &["id", "note"]), + ]); + + let result = run_analysis(sql, Dialect::Snowflake, Some(schema)); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "label"), + expected_sources(&["orders.id"]) + ); + + let fed: HashSet = flows_into_columns_of(&stmt, "f") + .into_iter() + .map(|(column, _)| column) + .collect(); + assert_eq!(fed, expected_sources(&["value", "this"])); +} + +#[test] +fn flatten_of_a_literal_reads_nothing() { + let sql = r#" + SELECT o.id, f.value AS step + FROM orders o, + LATERAL FLATTEN(input => ARRAY_CONSTRUCT(1, 2, 3)) f + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + assert!(flows_into_columns_of(&stmt, "f").is_empty()); + assert!(base_sources_of(&stmt, "step").is_empty()); +} + +#[test] +fn flatten_alias_reused_across_ctes_keeps_each_input() { + // Each CTE flattens its own column under the same alias. The second one + // passes the input positionally. + let sql = r#" + WITH order_tags AS ( + SELECT f.value::STRING AS tag + FROM orders o, + LATERAL FLATTEN(input => o.tags) f + ), + ticket_notes AS ( + SELECT f.value::STRING AS note + FROM tickets t, + LATERAL FLATTEN(t.notes) f + ) + SELECT tag, note + FROM order_tags, ticket_notes + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "tag"), + expected_sources(&["orders.tags"]) + ); + assert_eq!( + base_sources_of(&stmt, "note"), + expected_sources(&["tickets.notes"]) + ); +} + +#[test] +fn star_over_flatten_returns_its_columns() { + let sql = r#" + SELECT * + FROM items i, + LATERAL FLATTEN(input => i.arr) f + "#; + let schema = schema_of(vec![schema_table(None, None, "items", &["id", "arr"])]); + + let result = run_analysis(sql, Dialect::Snowflake, Some(schema)); + let stmt = first_statement(&result); + let output = stmt + .nodes + .iter() + .find(|node| node.node_type == NodeType::Output) + .expect("output node"); + let columns: HashSet = stmt + .edges + .iter() + .filter(|edge| edge.edge_type == EdgeType::Ownership && edge.from == output.id) + .filter_map(|edge| stmt.nodes.iter().find(|node| node.id == edge.to)) + .map(|node| node.label.to_lowercase()) + .collect(); + assert_eq!( + columns, + expected_sources(&["id", "arr", "seq", "key", "path", "index", "value", "this"]) + ); + for carried in ["value", "this"] { + assert_eq!( + base_sources_of(&stmt, carried), + expected_sources(&["items.arr"]), + "sources of {carried}" + ); + } + for position in ["seq", "key", "path", "index"] { + assert!( + base_sources_of(&stmt, position).is_empty(), + "{position} should read nothing" + ); + } +} + +#[test] +fn bare_column_beside_flatten_belongs_to_the_table() { + // No schema: `id` and `status` are not columns of FLATTEN, so they can + // only be columns of orders. + let sql = r#" + SELECT id, f.value AS v + FROM orders o, + LATERAL FLATTEN(input => o.tags) f + WHERE status = 'open' + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "id"), + expected_sources(&["orders.id"]) + ); + assert_eq!( + base_sources_of(&stmt, "v"), + expected_sources(&["orders.tags"]) + ); + let orders = stmt + .nodes + .iter() + .find(|node| node.node_type == NodeType::Table) + .expect("orders node"); + assert!( + orders + .filters + .iter() + .any(|filter| filter.expression.contains("status")), + "{:?}", + orders.filters + ); + assert!( + !issue_codes_list(&result).contains(&issue_codes::UNRESOLVED_REFERENCE.to_string()), + "{:?}", + result.issues + ); +} + +#[test] +fn bare_input_of_a_second_flatten_belongs_to_the_table() { + let sql = r#" + SELECT customer_id, phone.value::STRING AS phone_number + FROM customers, + LATERAL FLATTEN(input => addresses) addr, + LATERAL FLATTEN(input => phone_numbers) phone + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "customer_id"), + expected_sources(&["customers.customer_id"]) + ); + assert_eq!( + base_sources_of(&stmt, "phone_number"), + expected_sources(&["customers.phone_numbers"]) + ); + assert!( + !issue_codes_list(&result).contains(&issue_codes::UNRESOLVED_REFERENCE.to_string()), + "{:?}", + result.issues + ); +} + +#[test] +fn ambiguous_flatten_input_is_reported_once() { + // Without a schema `tags` may belong to either table. VALUE and THIS both + // carry it, yet it is one reference. FLATTEN has no `note`, so it is not + // named among the relations `note` may belong to. + let sql = r#" + SELECT note, f.value AS v + FROM orders o, customers c, + LATERAL FLATTEN(input => tags) f + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let unresolved: Vec<&str> = result + .issues + .iter() + .filter(|issue| issue.code == issue_codes::UNRESOLVED_REFERENCE) + .map(|issue| issue.message.as_str()) + .collect(); + assert_eq!( + unresolved, + vec![ + "Column 'tags' is ambiguous across tables in scope: CUSTOMERS, ORDERS", + "Column 'note' is ambiguous across tables in scope: CUSTOMERS, ORDERS", + ] + ); +} + +#[test] +fn lateral_table_function_other_than_flatten_is_reported() { + let sql = r#" + SELECT s.value AS part + FROM files t, + LATERAL SPLIT_TO_TABLE(t.csv, ',') s + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + assert!( + issue_codes_list(&result).contains(&issue_codes::UNSUPPORTED_SYNTAX.to_string()), + "{:?}", + result.issues + ); +} + +#[test] +fn flatten_adds_no_column_when_column_lineage_is_disabled() { + let sql = r#" + SELECT o.id, f.value AS v + FROM orders o, + LATERAL FLATTEN(input => o.tags) f + "#; + + let result = run_analysis_with_options( + sql, + Dialect::Snowflake, + None, + AnalysisOptions { + enable_column_lineage: Some(false), + ..Default::default() + }, + ); + let stmt = first_statement(&result); + + let tables: Vec = stmt + .nodes + .iter() + .filter(|node| node.node_type == NodeType::Table) + .map(|node| node.label.to_lowercase()) + .collect(); + assert_eq!(tables, vec!["orders".to_string()]); + assert!( + stmt.nodes + .iter() + .all(|node| matches!(node.node_type, NodeType::Table | NodeType::Output)), + "only relations should be emitted: {:?}", + stmt.nodes + ); +} + #[test] fn snowflake_higher_order_functions_track_source() { let sql = r#" @@ -12502,3 +12942,237 @@ fn oracle_insert_from_view_with_schema() { fn oracle_view_from_view_with_schema() { assert_oracle_no_unresolved_with_schema("view_from_view.sql"); } + +#[test] +fn derived_tables_sharing_an_alias_in_two_ctes_keep_their_own_columns() { + // A generated point in time query opens one `(...) AS v` per satellite. + let sql = r#" + WITH a_intervals AS ( + SELECT k, ts FROM (SELECT DISTINCT k, ts FROM sat_a) AS v + ), + b_intervals AS ( + SELECT k, ts FROM (SELECT DISTINCT k, ts FROM sat_b) AS v + ) + SELECT a.k, a.ts AS a_ts, b.ts AS b_ts + FROM a_intervals a + JOIN b_intervals b ON a.k = b.k + "#; + let schema = schema_of(vec![ + schema_table(None, None, "sat_a", &["k", "ts"]), + schema_table(None, None, "sat_b", &["k", "ts"]), + ]); + + let result = run_analysis(sql, Dialect::Snowflake, Some(schema)); + let stmt = first_statement(&result); + assert_eq!( + base_sources_of(&stmt, "a_ts"), + expected_sources(&["sat_a.ts"]) + ); + assert_eq!( + base_sources_of(&stmt, "b_ts"), + expected_sources(&["sat_b.ts"]) + ); + + let derived: Vec<&Node> = stmt + .nodes + .iter() + .copied() + .filter(|node| node.node_type == NodeType::Cte && &*node.label == "v") + .collect(); + assert_eq!(derived.len(), 2, "one node per derived table"); + assert_ne!(derived[0].id, derived[1].id); +} + +#[test] +fn a_derived_alias_reused_in_twelve_scopes_crosses_nothing() { + let ctes: Vec = (0..12) + .map(|i| format!("c{i} AS (SELECT ts FROM (SELECT DISTINCT ts FROM sat_{i}) AS v)")) + .collect(); + let outputs: Vec = (0..12).map(|i| format!("c{i}.ts AS ts_{i}")).collect(); + let from: Vec = (0..12).map(|i| format!("c{i}")).collect(); + let sql = format!( + "WITH {} SELECT {} FROM {}", + ctes.join(", "), + outputs.join(", "), + from.join(" CROSS JOIN ") + ); + let schema = schema_of( + (0..12) + .map(|i| schema_table(None, None, &format!("sat_{i}"), &["ts"])) + .collect(), + ); + + let result = run_analysis(&sql, Dialect::Snowflake, Some(schema)); + let stmt = first_statement(&result); + for i in 0..12 { + assert_eq!( + base_sources_of(&stmt, &format!("ts_{i}")), + expected_sources(&[&format!("sat_{i}.ts")]), + "ts_{i}" + ); + } +} + +/// Labels of the columns the statement returns, lowercased and sorted. +fn returned_columns(stmt: &StmtView<'_>) -> Vec { + let output_ids: HashSet<&str> = stmt + .nodes + .iter() + .filter(|node| node.node_type == NodeType::Output) + .map(|node| &*node.id) + .collect(); + let mut labels: Vec = stmt + .edges + .iter() + .filter(|edge| edge.edge_type == EdgeType::Ownership && output_ids.contains(&*edge.from)) + .filter_map(|edge| stmt.nodes.iter().find(|node| node.id == edge.to)) + .map(|node| node.label.to_lowercase()) + .collect(); + labels.sort(); + labels.dedup(); + labels +} + +fn predicate_schema() -> SchemaMetadata { + schema_of(vec![ + schema_table(None, None, "t", &["k", "x"]), + schema_table(None, None, "u", &["k", "y"]), + ]) +} + +#[test] +fn a_subquery_a_predicate_reads_returns_nothing() { + for sql in [ + "SELECT k FROM t WHERE x >= (SELECT MAX(y) AS floor_y FROM u)", + "SELECT k FROM t WHERE x IN (SELECT y FROM u)", + "SELECT k FROM t WHERE NOT EXISTS (SELECT 1 FROM u WHERE u.k = t.k)", + "SELECT k FROM t GROUP BY k HAVING MAX(x) > (SELECT MAX(y) FROM u)", + "SELECT t.k FROM t JOIN u ON u.k = t.k AND u.y IN (SELECT y FROM u)", + ] { + let result = run_analysis(sql, Dialect::Snowflake, Some(predicate_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["k"], "{sql}"); + } +} + +#[test] +fn a_subquery_a_predicate_reads_keeps_its_lineage_on_its_own_node() { + // Inside a CTE, the subquery's column must not join the CTE's columns + // either, or the star below would return it. + let sql = "WITH c AS (SELECT k FROM t WHERE x IN (SELECT y FROM u)) SELECT * FROM c"; + let result = run_analysis(sql, Dialect::Snowflake, Some(predicate_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["k"]); + let flows: Vec = flows_into_columns_of(&stmt, "(subquery)") + .into_iter() + .map(|(column, _)| column) + .collect(); + assert_eq!(flows, vec!["y"]); +} + +#[test] +fn a_scalar_subquery_in_the_select_list_is_still_returned() { + let sql = "SELECT k, (SELECT MAX(y) FROM u) AS m FROM t"; + let result = run_analysis(sql, Dialect::Snowflake, Some(predicate_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["k", "m"]); +} + +#[test] +fn a_derived_table_without_an_alias_is_read_like_one_with_it() { + let schema = schema_of(vec![schema_table(None, None, "orders", &["id", "amount"])]); + for sql in [ + "SELECT * FROM (SELECT id, amount FROM orders) WHERE amount > 0", + "SELECT * FROM (SELECT id, amount FROM orders) AS o WHERE amount > 0", + ] { + let result = run_analysis(sql, Dialect::Snowflake, Some(schema.clone())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["amount", "id"], "{sql}"); + assert_eq!( + base_sources_of(&stmt, "amount"), + expected_sources(&["orders.amount"]), + "{sql}" + ); + assert!( + issue_codes_list(&result).is_empty(), + "{sql}: {:?}", + issue_codes_list(&result) + ); + } +} + +fn star_options_schema() -> SchemaMetadata { + schema_of(vec![schema_table(None, None, "p", &["a", "b", "c"])]) +} + +#[test] +fn star_exclude_leaves_the_excluded_columns_out() { + for sql in [ + "SELECT * EXCLUDE (b) FROM p", + "SELECT * EXCLUDE b FROM p", + "SELECT p.* EXCLUDE (b) FROM p", + "WITH c AS (SELECT * EXCLUDE (b) FROM p) SELECT * FROM c", + ] { + let result = run_analysis(sql, Dialect::Snowflake, Some(star_options_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["a", "c"], "{sql}"); + } +} + +#[test] +fn a_column_excluded_and_computed_again_keeps_its_expression() { + // With the star copy of b still expanded, the two projections named b + // shared one node, the copy came first, and the expression was lost. + let sql = "SELECT * EXCLUDE (b), UPPER(b) AS b FROM p"; + let result = run_analysis(sql, Dialect::Snowflake, Some(star_options_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["a", "b", "c"]); + assert_eq!(base_sources_of(&stmt, "b"), expected_sources(&["p.b"])); + let output_b = stmt + .nodes + .iter() + .find(|node| { + node.node_type == NodeType::Column + && node.label.eq_ignore_ascii_case("b") + && stmt.edges.iter().any(|edge| { + edge.edge_type == EdgeType::Ownership + && edge.to == node.id + && stmt.nodes.iter().any(|owner| { + owner.id == edge.from && owner.node_type == NodeType::Output + }) + }) + }) + .expect("b is returned"); + let expressions: Vec> = stmt + .edges + .iter() + .filter(|edge| edge.to == output_b.id && edge.edge_type != EdgeType::Ownership) + .map(|edge| edge.expression.as_deref()) + .collect(); + assert_eq!(expressions, vec![Some("UPPER(b)")]); +} + +#[test] +fn star_except_leaves_the_excepted_columns_out() { + let sql = "SELECT * EXCEPT (b) FROM p"; + let result = run_analysis(sql, Dialect::Bigquery, Some(star_options_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["a", "c"]); +} + +#[test] +fn star_rename_returns_the_column_under_its_new_name() { + for sql in [ + "SELECT * RENAME (b AS bee) FROM p", + "SELECT * RENAME b AS bee FROM p", + ] { + let result = run_analysis(sql, Dialect::Snowflake, Some(star_options_schema())); + let stmt = first_statement(&result); + assert_eq!(returned_columns(&stmt), vec!["a", "bee", "c"], "{sql}"); + assert_eq!( + base_sources_of(&stmt, "bee"), + expected_sources(&["p.b"]), + "{sql}" + ); + } +} diff --git a/crates/flowscope-core/tests/snapshots/snapshots__run_snowflake_snapshot_test@snowflake_lateral_flatten.snap b/crates/flowscope-core/tests/snapshots/snapshots__run_snowflake_snapshot_test@snowflake_lateral_flatten.snap index bd12afa9..46fd0e0b 100644 --- a/crates/flowscope-core/tests/snapshots/snapshots__run_snowflake_snapshot_test@snowflake_lateral_flatten.snap +++ b/crates/flowscope-core/tests/snapshots/snapshots__run_snowflake_snapshot_test@snowflake_lateral_flatten.snap @@ -8,10 +8,80 @@ expression: clean_result "statementIndex": 0, "statementType": "SELECT", "joinCount": 1, - "complexityScore": 20 + "complexityScore": 28 } ], "nodes": [ + { + "id": "column_1f95f87efeb20682", + "type": "column", + "label": "KEY", + "canonicalName": { + "name": "KEY" + }, + "statementIds": [ + 0 + ] + }, + { + "id": "column_268ea64f9e076ed3", + "type": "column", + "label": "cool_ids", + "qualifiedName": "B.cool_ids", + "canonicalName": { + "schema": "B", + "name": "cool_ids" + }, + "statementIds": [ + 0 + ] + }, + { + "id": "column_2f1758652518e9a9", + "type": "column", + "label": "THIS", + "canonicalName": { + "name": "THIS" + }, + "statementIds": [ + 0 + ], + "expression": "FLATTEN(input => b.cool_ids)" + }, + { + "id": "column_4e2ed24f2245005f", + "type": "column", + "label": "VALUE", + "canonicalName": { + "name": "VALUE" + }, + "statementIds": [ + 0 + ], + "expression": "FLATTEN(input => b.cool_ids)" + }, + { + "id": "column_7bfb651f3214d1d7", + "type": "column", + "label": "INDEX", + "canonicalName": { + "name": "INDEX" + }, + "statementIds": [ + 0 + ] + }, + { + "id": "column_8b8a696c1d07ba23", + "type": "column", + "label": "PATH", + "canonicalName": { + "name": "PATH" + }, + "statementIds": [ + 0 + ] + }, { "id": "column_9312b887682fcd76", "type": "column", @@ -23,6 +93,17 @@ expression: clean_result 0 ] }, + { + "id": "column_c2f34f4bce303b03", + "type": "column", + "label": "SEQ", + "canonicalName": { + "name": "SEQ" + }, + "statementIds": [ + 0 + ] + }, { "id": "column_ffa7c99c403f959a", "type": "column", @@ -34,6 +115,18 @@ expression: clean_result 0 ] }, + { + "id": "flatten_c1bebf2d8b83f913", + "type": "cte", + "label": "FLATTEN", + "qualifiedName": "FLATTEN", + "canonicalName": { + "name": "FLATTEN" + }, + "statementIds": [ + 0 + ] + }, { "id": "output_74d22f755ec62dbc", "type": "output", @@ -99,6 +192,42 @@ expression: clean_result } ], "edges": [ + { + "id": "edge_01f4f9bc314d367e", + "from": "flatten_c1bebf2d8b83f913", + "to": "column_1f95f87efeb20682", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_09d6cf174d070d13", + "from": "flatten_c1bebf2d8b83f913", + "to": "column_7bfb651f3214d1d7", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_10b868357df4cc6e", + "from": "column_4e2ed24f2245005f", + "to": "column_9312b887682fcd76", + "type": "data_flow", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_338f68a009f32040", + "from": "flatten_c1bebf2d8b83f913", + "to": "column_2f1758652518e9a9", + "type": "ownership", + "statementIds": [ + 0 + ] + }, { "id": "edge_671cdc7a5a898ccd", "from": "output_74d22f755ec62dbc", @@ -108,6 +237,27 @@ expression: clean_result 0 ] }, + { + "id": "edge_6a0a97f0ea1ad2ab", + "from": "table_14fd3f715fc3ae74", + "to": "column_268ea64f9e076ed3", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_7823aeb8ad256f9d", + "from": "column_268ea64f9e076ed3", + "to": "column_4e2ed24f2245005f", + "type": "derivation", + "expression": "FLATTEN(input => b.cool_ids)", + "joinType": "INNER", + "joinCondition": "b.c_id = a.c_id", + "statementIds": [ + 0 + ] + }, { "id": "edge_7f6d7c7173448b60", "from": "table_14fd3f715fc3ae74", @@ -119,6 +269,15 @@ expression: clean_result 0 ] }, + { + "id": "edge_8ac209680be08157", + "from": "flatten_c1bebf2d8b83f913", + "to": "column_8b8a696c1d07ba23", + "type": "ownership", + "statementIds": [ + 0 + ] + }, { "id": "edge_8d1b60030155f6f4", "from": "output_74d22f755ec62dbc", @@ -127,6 +286,36 @@ expression: clean_result "statementIds": [ 0 ] + }, + { + "id": "edge_9c212231bc72c292", + "from": "flatten_c1bebf2d8b83f913", + "to": "column_c2f34f4bce303b03", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_a4e289214a89eb54", + "from": "column_268ea64f9e076ed3", + "to": "column_2f1758652518e9a9", + "type": "derivation", + "expression": "FLATTEN(input => b.cool_ids)", + "joinType": "INNER", + "joinCondition": "b.c_id = a.c_id", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_cb4d48c39a8c8c26", + "from": "flatten_c1bebf2d8b83f913", + "to": "column_4e2ed24f2245005f", + "type": "ownership", + "statementIds": [ + 0 + ] } ], "issues": [ @@ -135,23 +324,17 @@ expression: clean_result "code": "UNRESOLVED_REFERENCE", "message": "Column 'name' is ambiguous across tables in scope: A, B", "statementIndex": 0 - }, - { - "severity": "warning", - "code": "UNRESOLVED_REFERENCE", - "message": "Column 'value' is ambiguous across tables in scope: A, B", - "statementIndex": 0 } ], "summary": { "statementCount": 1, - "tableCount": 2, - "columnCount": 2, + "tableCount": 3, + "columnCount": 9, "joinCount": 1, - "complexityScore": 20, + "complexityScore": 28, "issueCount": { "errors": 0, - "warnings": 2, + "warnings": 1, "infos": 0 }, "hasErrors": false @@ -196,6 +379,10 @@ expression: clean_result "table": "A", "column": "c_id" } + }, + { + "name": "cool_ids", + "origin": "implied" } ], "origin": "implied",