diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fffbd80..fe5380d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- 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. + ## [0.9.2] - 2026-09-24 ### Added diff --git a/crates/flowscope-core/src/analyzer/expression.rs b/crates/flowscope-core/src/analyzer/expression.rs index 123d858c..aff1f2e9 100644 --- a/crates/flowscope-core/src/analyzer/expression.rs +++ b/crates/flowscope-core/src/analyzer/expression.rs @@ -28,6 +28,50 @@ use tracing::debug; /// on maliciously crafted or deeply nested SQL expressions. pub(super) const MAX_RECURSION_DEPTH: usize = 100; +/// Returns the operands of a chain of binary operators, left to right. +/// +/// The parser builds `a || b || c` as `(a || b) || c`, so each operator of a +/// flat chain adds one level on the left. Walking that spine in a loop keeps +/// the length of a chain from counting against `MAX_RECURSION_DEPTH`, which +/// guards real nesting: otherwise a surrogate key concatenating fifty columns +/// and their separators loses its leftmost operands. Right operands are still +/// visited by recursion, so nesting on that side keeps counting. +fn binary_op_operands(expr: &Expr) -> Vec<&Expr> { + let mut operands = Vec::new(); + let mut current = expr; + while let Expr::BinaryOp { left, right, .. } = current { + operands.push(right.as_ref()); + current = left; + } + operands.push(current); + operands.reverse(); + operands +} + +/// Splits a field access into the name it starts from and the steps after it. +/// +/// The parser keeps `o.items[1]` as the root `o` followed by the steps `.items` +/// and `[1]`. The name is `o.items`, read as it would be without the subscript, +/// and the steps after it are fields and subscripts of its value. The name is +/// empty when the root is not an identifier. +fn split_access_name<'e>( + root: &'e Expr, + access_chain: &'e [ast::AccessExpr], +) -> (Vec<&'e ast::Ident>, &'e [ast::AccessExpr]) { + let Expr::Identifier(first) = root else { + return (Vec::new(), access_chain); + }; + let mut name = vec![first]; + for step in access_chain { + match step { + ast::AccessExpr::Dot(Expr::Identifier(part)) => name.push(part), + _ => break, + } + } + let steps = &access_chain[name.len() - 1..]; + (name, steps) +} + /// Analyzes SQL expressions to extract column references, detect aggregations, /// and capture filter predicates. /// @@ -114,9 +158,10 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { self.analyzer.analyze_query(self.ctx, subquery, None) } Expr::Exists { subquery, .. } => self.analyzer.analyze_query(self.ctx, subquery, None), - Expr::BinaryOp { left, right, .. } => { - self.visit_expression_for_subqueries(left, next_depth); - self.visit_expression_for_subqueries(right, next_depth); + Expr::BinaryOp { .. } => { + for operand in binary_op_operands(expr) { + self.visit_expression_for_subqueries(operand, next_depth); + } } Expr::UnaryOp { expr, .. } => self.visit_expression_for_subqueries(expr, next_depth), Expr::Nested(expr) => self.visit_expression_for_subqueries(expr, next_depth), @@ -230,9 +275,10 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { column, }); } - Expr::BinaryOp { left, right, .. } => { - depth_limited |= Self::collect_column_refs(left, refs, dialect, next_depth); - depth_limited |= Self::collect_column_refs(right, refs, dialect, next_depth); + Expr::BinaryOp { .. } => { + for operand in binary_op_operands(expr) { + depth_limited |= Self::collect_column_refs(operand, refs, dialect, next_depth); + } } Expr::UnaryOp { expr, .. } => { depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); @@ -255,6 +301,10 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { FunctionArg::Named { arg: FunctionArgExpr::Expr(e), .. + } + | FunctionArg::ExprNamed { + arg: FunctionArgExpr::Expr(e), + .. } => { depth_limited |= Self::collect_column_refs(e, refs, dialect, next_depth); @@ -292,7 +342,7 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { Expr::Nested(inner) => { depth_limited |= Self::collect_column_refs(inner, refs, dialect, next_depth); } - Expr::Subquery(_) => { + Expr::Subquery(_) | Expr::Exists { .. } => { // Subquery columns are handled separately } Expr::InList { expr, list, .. } => { @@ -326,9 +376,207 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { Expr::Extract { expr, .. } => { depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); } - _ => { - // Other expressions don't contain column references or are handled elsewhere + // The parser gives TRIM, SUBSTRING, CEIL, `col:path` and the like + // variants of their own rather than function calls. There is no + // catch-all arm, so a variant added to sqlparser is a compile error + // here instead of a column that silently drops out of lineage. + Expr::IsUnknown(e) + | Expr::IsNotUnknown(e) + | Expr::IsNormalized { expr: e, .. } + | Expr::Ceil { expr: e, .. } + | Expr::Floor { expr: e, .. } + | Expr::Collate { expr: e, .. } + | Expr::Prefixed { value: e, .. } + | Expr::Named { expr: e, .. } + | Expr::OuterJoin(e) + | Expr::Prior(e) => { + depth_limited |= Self::collect_column_refs(e, refs, dialect, next_depth); + } + Expr::Interval(interval) => { + depth_limited |= + Self::collect_column_refs(&interval.value, refs, dialect, next_depth); + } + Expr::IsDistinctFrom(left, right) + | Expr::IsNotDistinctFrom(left, right) + | Expr::AnyOp { left, right, .. } + | Expr::AllOp { left, right, .. } + | Expr::SimilarTo { + expr: left, + pattern: right, + .. + } + | Expr::RLike { + expr: left, + pattern: right, + .. + } + | Expr::Position { + expr: left, + r#in: right, + } + | Expr::AtTimeZone { + timestamp: left, + time_zone: right, + } + | Expr::InUnnest { + expr: left, + array_expr: right, + .. + } => { + depth_limited |= Self::collect_column_refs(left, refs, dialect, next_depth); + depth_limited |= Self::collect_column_refs(right, refs, dialect, next_depth); + } + Expr::MemberOf(member_of) => { + depth_limited |= + Self::collect_column_refs(&member_of.value, refs, dialect, next_depth); + depth_limited |= + Self::collect_column_refs(&member_of.array, refs, dialect, next_depth); + } + Expr::Trim { + expr, + trim_what, + trim_characters, + .. + } => { + depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); + if let Some(what) = trim_what { + depth_limited |= Self::collect_column_refs(what, refs, dialect, next_depth); + } + for characters in trim_characters.iter().flatten() { + depth_limited |= + Self::collect_column_refs(characters, refs, dialect, next_depth); + } + } + Expr::Substring { + expr, + substring_from, + substring_for, + .. + } => { + depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); + for bound in [substring_from, substring_for].into_iter().flatten() { + depth_limited |= Self::collect_column_refs(bound, refs, dialect, next_depth); + } + } + Expr::Overlay { + expr, + overlay_what, + overlay_from, + overlay_for, + } => { + depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); + depth_limited |= Self::collect_column_refs(overlay_what, refs, dialect, next_depth); + depth_limited |= Self::collect_column_refs(overlay_from, refs, dialect, next_depth); + if let Some(length) = overlay_for { + depth_limited |= Self::collect_column_refs(length, refs, dialect, next_depth); + } + } + Expr::Convert { expr, styles, .. } => { + depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); + for style in styles { + depth_limited |= Self::collect_column_refs(style, refs, dialect, next_depth); + } + } + // A dot step names a field of the value, not a column of the query; + // a subscript or bracket key is an expression and can read one. + Expr::JsonAccess { value, path } => { + depth_limited |= Self::collect_column_refs(value, refs, dialect, next_depth); + for elem in &path.path { + if let ast::JsonPathElem::Bracket { key } = elem { + depth_limited |= Self::collect_column_refs(key, refs, dialect, next_depth); + } + } + } + Expr::CompoundFieldAccess { root, access_chain } => { + let (name, steps) = split_access_name(root, access_chain); + match name.split_last() { + Some((column, table)) if !table.is_empty() => refs.push(ColumnRef { + table: Some( + table + .iter() + .map(|part| part.value.as_str()) + .collect::>() + .join("."), + ), + column: column.value.clone(), + }), + _ => { + depth_limited |= Self::collect_column_refs(root, refs, dialect, next_depth); + } + } + for access in steps { + match access { + ast::AccessExpr::Dot(_) => {} + ast::AccessExpr::Subscript(ast::Subscript::Index { index }) => { + depth_limited |= + Self::collect_column_refs(index, refs, dialect, next_depth); + } + ast::AccessExpr::Subscript(ast::Subscript::Slice { + lower_bound, + upper_bound, + stride, + }) => { + for bound in [lower_bound, upper_bound, stride].into_iter().flatten() { + depth_limited |= + Self::collect_column_refs(bound, refs, dialect, next_depth); + } + } + } + } } + Expr::Struct { values: exprs, .. } | Expr::Array(ast::Array { elem: exprs, .. }) => { + for e in exprs { + depth_limited |= Self::collect_column_refs(e, refs, dialect, next_depth); + } + } + Expr::GroupingSets(sets) | Expr::Cube(sets) | Expr::Rollup(sets) => { + for e in sets.iter().flatten() { + depth_limited |= Self::collect_column_refs(e, refs, dialect, next_depth); + } + } + Expr::Dictionary(fields) => { + for field in fields { + depth_limited |= + Self::collect_column_refs(&field.value, refs, dialect, next_depth); + } + } + Expr::Map(map) => { + for entry in &map.entries { + depth_limited |= + Self::collect_column_refs(&entry.key, refs, dialect, next_depth); + depth_limited |= + Self::collect_column_refs(&entry.value, refs, dialect, next_depth); + } + } + Expr::MatchAgainst { columns, .. } => { + for name in columns { + let mut parts = name.0.iter().filter_map(|part| part.as_ident()); + let Some(column) = parts.next_back() else { + continue; + }; + let table = parts.map(|i| i.value.as_str()).collect::>(); + refs.push(ColumnRef { + table: (!table.is_empty()).then(|| table.join(".")), + column: column.value.clone(), + }); + } + } + Expr::InSubquery { expr, .. } => { + // The value tested is read here; the subquery has its own scope + // and is analysed separately. + depth_limited |= Self::collect_column_refs(expr, refs, dialect, next_depth); + } + Expr::Lambda(_) => { + // Reading the body would take the lambda's parameters for + // columns of the enclosing query. + } + Expr::CompoundIdentifier(_) => { + // The parser never builds a compound name of fewer than two parts. + } + Expr::Value(_) + | Expr::TypedString(_) + | Expr::Wildcard(_) + | Expr::QualifiedWildcard(..) => {} } depth_limited @@ -424,9 +672,9 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { match expr { Expr::Function(func) => self.check_function_for_aggregate(func, next_depth), - Expr::BinaryOp { left, right, .. } => self - .find_aggregate_function(left, next_depth) - .or_else(|| self.find_aggregate_function(right, next_depth)), + Expr::BinaryOp { .. } => binary_op_operands(expr) + .into_iter() + .find_map(|operand| self.find_aggregate_function(operand, next_depth)), Expr::UnaryOp { expr, .. } | Expr::Nested(expr) | Expr::Cast { expr, .. } => { self.find_aggregate_function(expr, next_depth) } @@ -671,6 +919,9 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { identifiers } + /// Walks the same forms as `collect_column_refs`. A lateral alias that + /// lineage reads but this walk misses is never substituted, so it stays a + /// column reference and is resolved against the tables in scope. fn collect_simple_identifiers(expr: &Expr, identifiers: &mut HashSet, depth: usize) { if depth > MAX_RECURSION_DEPTH { #[cfg(feature = "tracing")] @@ -703,14 +954,25 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { | Expr::IsNotTrue(e) | Expr::IsUnknown(e) | Expr::IsNotUnknown(e) - | Expr::JsonAccess { value: e, .. } => { + | Expr::IsNormalized { expr: e, .. } + | Expr::Collate { expr: e, .. } + | Expr::Prefixed { value: e, .. } + | Expr::Named { expr: e, .. } + | Expr::OuterJoin(e) + | Expr::Prior(e) => { Self::collect_simple_identifiers(e, identifiers, next_depth); } + Expr::Interval(interval) => { + Self::collect_simple_identifiers(&interval.value, identifiers, next_depth); + } // Two expression patterns (left/right) - Expr::BinaryOp { left, right, .. } - | Expr::AnyOp { left, right, .. } - | Expr::AllOp { left, right, .. } => { + Expr::BinaryOp { .. } => { + for operand in binary_op_operands(expr) { + Self::collect_simple_identifiers(operand, identifiers, next_depth); + } + } + Expr::AnyOp { left, right, .. } | Expr::AllOp { left, right, .. } => { Self::collect_simple_identifiers(left, identifiers, next_depth); Self::collect_simple_identifiers(right, identifiers, next_depth); } @@ -746,6 +1008,10 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { Self::collect_simple_identifiers(e1, identifiers, next_depth); Self::collect_simple_identifiers(e2, identifiers, next_depth); } + Expr::MemberOf(member_of) => { + Self::collect_simple_identifiers(&member_of.value, identifiers, next_depth); + Self::collect_simple_identifiers(&member_of.array, identifiers, next_depth); + } // Three expression patterns Expr::Between { @@ -757,7 +1023,9 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { } // List patterns - Expr::Tuple(exprs) => { + Expr::Tuple(exprs) + | Expr::Struct { values: exprs, .. } + | Expr::Array(ast::Array { elem: exprs, .. }) => { for e in exprs { Self::collect_simple_identifiers(e, identifiers, next_depth); } @@ -768,6 +1036,22 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { Self::collect_simple_identifiers(e, identifiers, next_depth); } } + Expr::GroupingSets(sets) | Expr::Cube(sets) | Expr::Rollup(sets) => { + for e in sets.iter().flatten() { + Self::collect_simple_identifiers(e, identifiers, next_depth); + } + } + Expr::Dictionary(fields) => { + for field in fields { + Self::collect_simple_identifiers(&field.value, identifiers, next_depth); + } + } + Expr::Map(map) => { + for entry in &map.entries { + Self::collect_simple_identifiers(&entry.key, identifiers, next_depth); + Self::collect_simple_identifiers(&entry.value, identifiers, next_depth); + } + } // Function arguments Expr::Function(func) => { @@ -778,6 +1062,10 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { | FunctionArg::Named { arg: FunctionArgExpr::Expr(e), .. + } + | FunctionArg::ExprNamed { + arg: FunctionArgExpr::Expr(e), + .. } => Self::collect_simple_identifiers(e, identifiers, next_depth), _ => {} } @@ -820,8 +1108,84 @@ impl<'a, 'b> ExpressionAnalyzer<'a, 'b> { } } + Expr::Trim { + expr, + trim_what, + trim_characters, + .. + } => { + Self::collect_simple_identifiers(expr, identifiers, next_depth); + if let Some(what) = trim_what { + Self::collect_simple_identifiers(what, identifiers, next_depth); + } + for characters in trim_characters.iter().flatten() { + Self::collect_simple_identifiers(characters, identifiers, next_depth); + } + } + Expr::Overlay { + expr, + overlay_what, + overlay_from, + overlay_for, + } => { + Self::collect_simple_identifiers(expr, identifiers, next_depth); + Self::collect_simple_identifiers(overlay_what, identifiers, next_depth); + Self::collect_simple_identifiers(overlay_from, identifiers, next_depth); + if let Some(length) = overlay_for { + Self::collect_simple_identifiers(length, identifiers, next_depth); + } + } + Expr::Convert { expr, styles, .. } => { + Self::collect_simple_identifiers(expr, identifiers, next_depth); + for style in styles { + Self::collect_simple_identifiers(style, identifiers, next_depth); + } + } + Expr::JsonAccess { value, path } => { + Self::collect_simple_identifiers(value, identifiers, next_depth); + for elem in &path.path { + if let ast::JsonPathElem::Bracket { key } = elem { + Self::collect_simple_identifiers(key, identifiers, next_depth); + } + } + } + Expr::CompoundFieldAccess { root, access_chain } => { + let (name, steps) = split_access_name(root, access_chain); + // `o.items[1]` names a qualified column, which is no simple identifier. + if name.len() < 2 { + Self::collect_simple_identifiers(root, identifiers, next_depth); + } + for access in steps { + match access { + ast::AccessExpr::Dot(_) => {} + ast::AccessExpr::Subscript(ast::Subscript::Index { index }) => { + Self::collect_simple_identifiers(index, identifiers, next_depth); + } + ast::AccessExpr::Subscript(ast::Subscript::Slice { + lower_bound, + upper_bound, + stride, + }) => { + for bound in [lower_bound, upper_bound, stride].into_iter().flatten() { + Self::collect_simple_identifiers(bound, identifiers, next_depth); + } + } + } + } + } + Expr::MatchAgainst { columns, .. } => { + for name in columns { + if let [ast::ObjectNamePart::Identifier(ident)] = name.0.as_slice() { + identifiers.insert(ident.value.clone()); + } + } + } + Expr::InSubquery { expr, .. } => { + Self::collect_simple_identifiers(expr, identifiers, next_depth); + } + // Skip subqueries - they have their own scope - Expr::Subquery(_) | Expr::InSubquery { .. } | Expr::Exists { .. } => {} + Expr::Subquery(_) | Expr::Exists { .. } => {} // Skip qualified names (table.column) - not simple identifiers Expr::CompoundIdentifier(_) => {} @@ -852,4 +1216,172 @@ mod tests { "no column refs should be recorded when guard triggers" ); } + + fn column(index: usize) -> Expr { + Expr::Identifier(ast::Ident::new(format!("c{index}"))) + } + + fn concat(left: Expr, right: Expr) -> Expr { + Expr::BinaryOp { + left: Box::new(left), + op: ast::BinaryOperator::StringConcat, + right: Box::new(right), + } + } + + /// `c0 || c1 || ... || c{len - 1}`, left-deep as the parser builds it. + fn flat_concat_chain(len: usize) -> Expr { + (1..len).fold(column(0), |chain, index| concat(chain, column(index))) + } + + #[test] + fn collect_column_refs_walks_long_operator_chains() { + let len = MAX_RECURSION_DEPTH * 3; + let mut refs = Vec::new(); + let hit = ExpressionAnalyzer::collect_column_refs( + &flat_concat_chain(len), + &mut refs, + Dialect::Generic, + 0, + ); + + assert!(!hit, "a flat operator chain should not trigger the guard"); + let columns: Vec = refs.into_iter().map(|col_ref| col_ref.column).collect(); + let expected: Vec = (0..len).map(|index| format!("c{index}")).collect(); + assert_eq!( + columns, expected, + "operands should be collected left to right" + ); + } + + #[test] + fn collect_column_refs_still_limits_nesting_on_the_right() { + let expr = (0..=MAX_RECURSION_DEPTH) + .rev() + .fold(column(MAX_RECURSION_DEPTH + 1), |nested, index| { + concat(column(index), nested) + }); + let mut refs = Vec::new(); + let hit = ExpressionAnalyzer::collect_column_refs(&expr, &mut refs, Dialect::Generic, 0); + + assert!(hit, "nesting past the limit should still trigger the guard"); + } + + /// Expressions whose operands the parser keeps outside function arguments, + /// with the columns each one reads, qualified as written. + const DEDICATED_FORMS: &[(Dialect, &str, &[&str])] = &[ + ( + Dialect::Generic, + "TRIM(BOTH pad FROM c.name)", + &["c.name", "pad"], + ), + (Dialect::Snowflake, "TRIM(name, pad)", &["name", "pad"]), + ( + Dialect::Generic, + "SUBSTRING(name FROM start FOR len)", + &["name", "start", "len"], + ), + ( + Dialect::Generic, + "POSITION(needle IN hay)", + &["needle", "hay"], + ), + ( + Dialect::Generic, + "CEIL(amount) + FLOOR(fee)", + &["amount", "fee"], + ), + (Dialect::Generic, "ts AT TIME ZONE zone", &["ts", "zone"]), + (Dialect::Generic, "a IS DISTINCT FROM b", &["a", "b"]), + (Dialect::Generic, "flag IS UNKNOWN", &["flag"]), + ( + Dialect::Snowflake, + "payload:items[pos]", + &["payload", "pos"], + ), + (Dialect::Postgres, "items[pos].field", &["items", "pos"]), + (Dialect::Postgres, "t.items[pos]", &["t.items", "pos"]), + (Dialect::Duckdb, "s.t.items[1].field", &["s.t.items"]), + (Dialect::Postgres, "name COLLATE \"C\"", &["name"]), + ( + Dialect::Postgres, + "OVERLAY(name PLACING patch FROM 2 FOR len)", + &["name", "patch", "len"], + ), + ( + Dialect::Postgres, + "name SIMILAR TO pattern", + &["name", "pattern"], + ), + (Dialect::Mysql, "name RLIKE pattern", &["name", "pattern"]), + (Dialect::Postgres, "a = ANY(b)", &["a", "b"]), + (Dialect::Mysql, "CONVERT(amount, CHAR)", &["amount"]), + (Dialect::Postgres, "ARRAY[a, b]", &["a", "b"]), + (Dialect::Postgres, "f(arg => col)", &["col"]), + (Dialect::Mysql, "DATE_ADD(d, INTERVAL n DAY)", &["d", "n"]), + ( + Dialect::Mysql, + "MATCH (title, t.body) AGAINST ('x')", + &["title", "t.body"], + ), + (Dialect::Generic, "x IN (SELECT y FROM t)", &["x"]), + (Dialect::Generic, "EXISTS (SELECT y FROM t)", &[]), + (Dialect::Duckdb, "list_transform(xs, x -> x + 1)", &["xs"]), + ]; + + fn parse_expr(sql: &str, dialect: Dialect) -> Expr { + sqlparser::parser::Parser::new(dialect.to_sqlparser_dialect().as_ref()) + .try_with_sql(sql) + .and_then(|mut parser| parser.parse_expr()) + .unwrap_or_else(|err| panic!("{sql} should parse: {err}")) + } + + #[test] + fn collect_column_refs_reads_operands_of_dedicated_forms() { + for &(dialect, sql, expected) in DEDICATED_FORMS { + let (refs, hit) = ExpressionAnalyzer::extract_column_refs_with_dialect( + &parse_expr(sql, dialect), + dialect, + ); + + assert!(!hit, "{sql} should not trigger the guard"); + let columns: Vec = refs + .into_iter() + .map(|col_ref| match col_ref.table { + Some(table) => format!("{table}.{}", col_ref.column), + None => col_ref.column, + }) + .collect(); + assert_eq!(columns, expected, "columns read by {sql}"); + } + } + + #[test] + fn extract_simple_identifiers_sees_what_collect_column_refs_reads() { + for &(dialect, sql, expected) in DEDICATED_FORMS { + let unqualified: HashSet = expected + .iter() + .filter(|column| !column.contains('.')) + .map(|column| column.to_string()) + .collect(); + + assert_eq!( + ExpressionAnalyzer::extract_simple_identifiers(&parse_expr(sql, dialect)), + unqualified, + "bare identifiers of {sql}" + ); + } + } + + #[test] + fn extract_simple_identifiers_walks_long_operator_chains() { + let len = MAX_RECURSION_DEPTH * 3; + let identifiers = ExpressionAnalyzer::extract_simple_identifiers(&flat_concat_chain(len)); + + assert_eq!( + identifiers.len(), + len, + "every operand of a flat chain should be collected" + ); + } } diff --git a/crates/flowscope-core/tests/lineage_engine.rs b/crates/flowscope-core/tests/lineage_engine.rs index 645d3ec9..ae5f31f8 100644 --- a/crates/flowscope-core/tests/lineage_engine.rs +++ b/crates/flowscope-core/tests/lineage_engine.rs @@ -7258,6 +7258,239 @@ fn deeply_nested_case_expressions() { assert!(col.is_some(), "nested CASE produces column"); } +#[test] +fn long_concatenation_keeps_every_operand() { + // A generated surrogate key joins every field with a separator, so 120 + // fields parse as a left-deep chain of 238 `||` operators. + let fields: Vec = (0..120).map(|i| format!("field_{i:03}")).collect(); + let key = fields + .iter() + .map(|field| format!("COALESCE(CAST({field} AS VARCHAR), '')")) + .collect::>() + .join(" || '-' || "); + let sql = format!("SELECT MD5({key}) AS row_key FROM events"); + + let result = run_analysis(&sql, Dialect::Generic, None); + assert!( + !issue_codes_list(&result).contains(&issue_codes::APPROXIMATE_LINEAGE.to_string()), + "a flat operator chain is not deep nesting, issues: {:?}", + result.issues + ); + + let stmt = first_statement(&result); + let row_key = find_column_node(&stmt, "row_key").expect("row_key column should exist"); + let sources: HashSet<&str> = stmt + .edges + .iter() + .filter(|edge| edge.edge_type == EdgeType::Derivation && edge.to == row_key.id) + .filter_map(|edge| stmt.nodes.iter().find(|node| node.id == edge.from)) + .map(|node| &*node.label) + .collect(); + let missing: Vec<&str> = fields + .iter() + .map(String::as_str) + .filter(|field| !sources.contains(field)) + .collect(); + assert!( + missing.is_empty(), + "every concatenated field should derive row_key, missing: {missing:?}" + ); +} + +/// Sources of an output column as `relation.column`, lowercased, following +/// derivation and data-flow edges to the relation that owns each source. +fn source_columns_of(stmt: &StmtView<'_>, output: &str) -> HashSet { + let target = stmt + .nodes + .iter() + .find(|node| node.node_type == NodeType::Column && node.label.eq_ignore_ascii_case(output)) + .unwrap_or_else(|| panic!("output column {output} should exist")); + stmt.edges + .iter() + .filter(|edge| { + edge.to == target.id + && matches!(edge.edge_type, EdgeType::Derivation | EdgeType::DataFlow) + }) + .filter_map(|edge| { + let source = stmt.nodes.iter().find(|node| node.id == edge.from)?; + let owner = stmt + .edges + .iter() + .find(|owned| owned.edge_type == EdgeType::Ownership && owned.to == source.id) + .and_then(|owned| stmt.nodes.iter().find(|node| node.id == owned.from))?; + Some(format!("{}.{}", owner.label, source.label).to_lowercase()) + }) + .collect() +} + +fn expected_sources(sources: &[&str]) -> HashSet { + sources.iter().map(|source| source.to_string()).collect() +} + +#[test] +fn trim_reads_the_column_of_the_relation_it_names() { + // Both relations have `name`. The column inside TRIM belongs to the joined + // relation, and must not be credited to the one the query is driven from. + let sql = r#" + SELECT o.id, TRIM(c.name) AS customer_name + FROM orders o + JOIN customers c ON o.customer_id = c.id + "#; + let schema = SchemaMetadata { + allow_implied: false, + default_catalog: None, + default_schema: None, + search_path: None, + case_sensitivity: None, + tables: vec![ + schema_table(None, None, "orders", &["id", "customer_id", "name"]), + schema_table(None, None, "customers", &["id", "name"]), + ], + }; + + let result = run_analysis(sql, Dialect::Generic, Some(schema)); + let stmt = first_statement(&result); + assert_eq!( + source_columns_of(&stmt, "customer_name"), + expected_sources(&["customers.name"]) + ); +} + +#[test] +fn dedicated_expression_forms_read_every_operand() { + // SUBSTR, CEIL, POSITION and the `:` path operator parse to their own + // expression variants rather than to function calls. + let sql = r#" + SELECT + SUBSTR(code, 1, 2) || suffix AS short_code, + CEIL(amount) + fee AS total, + payload:kind::STRING || suffix AS tagged, + POSITION('-' IN code) + fee AS dash_at + FROM events + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + for (output, sources) in [ + ("short_code", ["events.code", "events.suffix"]), + ("total", ["events.amount", "events.fee"]), + ("tagged", ["events.payload", "events.suffix"]), + ("dash_at", ["events.code", "events.fee"]), + ] { + assert_eq!( + source_columns_of(&stmt, output), + expected_sources(&sources), + "sources of {output}" + ); + } +} + +#[test] +fn lateral_alias_inside_trim_resolves_to_its_sources() { + // Keys hashed from earlier keys of the same SELECT list: the lateral + // aliases are only visible inside TRIM. + let sql = r#" + SELECT + MD5(UPPER(TRIM(CAST(order_id AS VARCHAR)))) AS order_key, + MD5(UPPER(TRIM(CAST(customer_id AS VARCHAR)))) AS customer_key, + MD5(CONCAT( + UPPER(TRIM(CAST(order_key AS VARCHAR))), '||', + UPPER(TRIM(CAST(customer_key AS VARCHAR))) + )) AS link_key + FROM orders + "#; + + let result = run_analysis(sql, Dialect::Snowflake, None); + let stmt = first_statement(&result); + assert_eq!( + source_columns_of(&stmt, "link_key"), + expected_sources(&["orders.order_id", "orders.customer_id"]) + ); +} + +#[test] +fn filter_inside_trim_is_attached_to_its_table() { + let sql = r#" + SELECT o.id + FROM orders o + JOIN customers c ON o.customer_id = c.id + WHERE TRIM(c.status) = 'active' + "#; + + let result = run_analysis(sql, Dialect::Generic, None); + let stmt = first_statement(&result); + let customers = find_table_node(&stmt, "customers").expect("customers table not found"); + let orders = find_table_node(&stmt, "orders").expect("orders table not found"); + assert_eq!( + customers + .filters + .iter() + .map(|filter| filter.expression.as_str()) + .collect::>(), + ["TRIM(c.status) = 'active'"] + ); + assert!(orders.filters.is_empty(), "orders: {:?}", orders.filters); +} + +#[test] +fn in_subquery_reads_its_left_operand_only() { + // The subquery is analysed in its own scope, so its columns are not + // sources of the flag; the value tested against it is. + let sql = r#" + SELECT + o.id, + CASE WHEN o.customer_id IN (SELECT c.id FROM customers c) THEN 1 ELSE 0 END AS known + FROM orders o + "#; + + let result = run_analysis(sql, Dialect::Generic, None); + let stmt = first_statement(&result); + assert_eq!( + source_columns_of(&stmt, "known"), + expected_sources(&["orders.customer_id"]) + ); +} + +#[test] +fn lambda_parameters_are_not_read_as_columns() { + let sql = r#" + SELECT list_transform(prices, p -> p * rate) AS scaled + FROM products + "#; + + let result = run_analysis(sql, Dialect::Duckdb, None); + let stmt = first_statement(&result); + let sources = source_columns_of(&stmt, "scaled"); + assert!( + !sources.iter().any(|source| source.ends_with(".p")), + "a lambda parameter is not a column: {sources:?}" + ); + assert!(sources.contains("products.prices"), "sources: {sources:?}"); +} + +#[test] +fn subscript_of_a_qualified_column_reads_that_column() { + // `o.items[1]` parses as the root `o` with the steps `.items` and `[1]`. + // The root is the table alias, not a column of orders. + let sql = "SELECT o.id, o.items[1] AS first_item FROM orders o"; + let schema = SchemaMetadata { + allow_implied: false, + default_catalog: None, + default_schema: None, + search_path: None, + case_sensitivity: None, + tables: vec![schema_table(None, None, "orders", &["id", "items"])], + }; + + let result = run_analysis(sql, Dialect::Postgres, Some(schema)); + let stmt = first_statement(&result); + assert_eq!( + source_columns_of(&stmt, "first_item"), + expected_sources(&["orders.items"]) + ); + assert!(result.issues.is_empty(), "issues: {:?}", result.issues); +} + #[test] fn case_with_aggregate_in_condition() { // CASE with aggregate function in WHEN condition diff --git a/crates/flowscope-core/tests/snapshots/snapshots__run_postgres_snapshot_test@postgres_array_slicing.snap b/crates/flowscope-core/tests/snapshots/snapshots__run_postgres_snapshot_test@postgres_array_slicing.snap index 244d1a7a..64dd0f59 100644 --- a/crates/flowscope-core/tests/snapshots/snapshots__run_postgres_snapshot_test@postgres_array_slicing.snap +++ b/crates/flowscope-core/tests/snapshots/snapshots__run_postgres_snapshot_test@postgres_array_slicing.snap @@ -48,6 +48,32 @@ expression: clean_result ], "expression": "c[2:]" }, + { + "id": "column_95b826196d892dac", + "type": "column", + "label": "c", + "qualifiedName": "array_data.c", + "canonicalName": { + "schema": "array_data", + "name": "c" + }, + "statementIds": [ + 0 + ] + }, + { + "id": "column_b683e029cadda1d4", + "type": "column", + "label": "a", + "qualifiedName": "array_data.a", + "canonicalName": { + "schema": "array_data", + "name": "a" + }, + "statementIds": [ + 0 + ] + }, { "id": "column_bceb3c4d983131c5", "type": "column", @@ -60,6 +86,32 @@ expression: clean_result ], "expression": "d[2:3]" }, + { + "id": "column_c4a845893b35a6fa", + "type": "column", + "label": "b", + "qualifiedName": "array_data.b", + "canonicalName": { + "schema": "array_data", + "name": "b" + }, + "statementIds": [ + 0 + ] + }, + { + "id": "column_ea9f00345477a676", + "type": "column", + "label": "d", + "qualifiedName": "array_data.d", + "canonicalName": { + "schema": "array_data", + "name": "d" + }, + "statementIds": [ + 0 + ] + }, { "id": "output_74d22f755ec62dbc", "type": "output", @@ -100,8 +152,27 @@ expression: clean_result ], "edges": [ { - "id": "edge_550b27dba85a2465", + "id": "edge_16ba963d993199aa", + "from": "column_95b826196d892dac", + "to": "column_84c9f50405fdf52c", + "type": "derivation", + "expression": "c[2:]", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_1d5f16b94e24d605", "from": "table_eb6f82b7323d024e", + "to": "column_c4a845893b35a6fa", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_29829752b6d6e0ac", + "from": "column_ea9f00345477a676", "to": "column_bceb3c4d983131c5", "type": "derivation", "expression": "d[2:3]", @@ -110,17 +181,26 @@ expression: clean_result ] }, { - "id": "edge_77f5912a72ad7682", - "from": "output_74d22f755ec62dbc", - "to": "column_7748741d00eb4aa0", + "id": "edge_30772b2aeea20852", + "from": "table_eb6f82b7323d024e", + "to": "column_ea9f00345477a676", "type": "ownership", "statementIds": [ 0 ] }, { - "id": "edge_8128a9fd6e58dbf4", + "id": "edge_675b380196aa1111", "from": "table_eb6f82b7323d024e", + "to": "column_95b826196d892dac", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_75998b7fa2f4cdb8", + "from": "column_b683e029cadda1d4", "to": "column_7748741d00eb4aa0", "type": "derivation", "expression": "a[:]", @@ -129,11 +209,10 @@ expression: clean_result ] }, { - "id": "edge_91d9a3175fef27e4", - "from": "table_eb6f82b7323d024e", - "to": "column_6d76c5c7b69b69cb", - "type": "derivation", - "expression": "b[:1]", + "id": "edge_77f5912a72ad7682", + "from": "output_74d22f755ec62dbc", + "to": "column_7748741d00eb4aa0", + "type": "ownership", "statementIds": [ 0 ] @@ -148,11 +227,20 @@ expression: clean_result ] }, { - "id": "edge_ad3bc1a5b25c59cc", + "id": "edge_a4180dafa0264175", "from": "table_eb6f82b7323d024e", - "to": "column_84c9f50405fdf52c", + "to": "column_b683e029cadda1d4", + "type": "ownership", + "statementIds": [ + 0 + ] + }, + { + "id": "edge_dd32db879aa8c27f", + "from": "column_c4a845893b35a6fa", + "to": "column_6d76c5c7b69b69cb", "type": "derivation", - "expression": "c[2:]", + "expression": "b[:1]", "statementIds": [ 0 ] @@ -180,7 +268,7 @@ expression: clean_result "summary": { "statementCount": 1, "tableCount": 1, - "columnCount": 4, + "columnCount": 8, "joinCount": 0, "complexityScore": 5, "issueCount": {