Fix lineage through LATERAL FLATTEN, reused derived aliases, predicate subqueries and star options

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
 

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<Arc<str>>,
 }
 
 impl Scope {
@@ -137,6 +140,13 @@ pub(crate) struct StatementContext {
     pub(crate) table_aliases: HashMap<String, String>,
     /// Subquery aliases (for reference tracking)
     pub(crate) subquery_aliases: HashSet<String>,
+    /// 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<String, usize>,
+    /// 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<String>,
     /// 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<str>) {
+        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<RelationInstance> {
+        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<OutputColumn>) {
         self.aliased_subquery_columns.insert(name, columns);

crates/flowscope-core/src/analyzer/expression.rs

@@ -13,17 +13,21 @@
 
 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;
 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);

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<String> = 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<String, String> = 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<str>,
+    ) {
+        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,

crates/flowscope-core/src/analyzer/select_analyzer.rs

@@ -230,19 +230,24 @@ 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);
                     self.analyzer.expand_wildcard(
                         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,
+                    );
                 }
             }
         }