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",