Read a column over an earlier alias of its name

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

@@ -839,7 +839,11 @@ impl<'a> Analyzer<'a> {
             .or_else(|| ctx.cte_definitions.get(&canonical).cloned())
     }
 
-    fn candidate_tables_for_column(&self, ctx: &StatementContext, column: &str) -> Vec<String> {
+    pub(super) fn candidate_tables_for_column(
+        &self,
+        ctx: &StatementContext,
+        column: &str,
+    ) -> Vec<String> {
         let tables_in_scope = ctx.tables_in_current_scope();
         let normalized_col = self.normalize_identifier(column);
         let mut candidate_tables: Vec<String> = Vec::new();

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

@@ -263,6 +263,12 @@ impl<'a, 'b> SelectAnalyzer<'a, 'b> {
     /// When processing `c`, the identifier `b` matches the lateral alias, so we
     /// resolve `c`'s sources to include `t.a` (via `b`) instead of treating `b`
     /// as an unresolved column reference.
+    ///
+    /// A column of the `FROM` wins over an alias of its name: in
+    /// `SELECT a AS b, b AS c FROM t` where `t` has `b`, Snowflake and DuckDB
+    /// read `t.b` into `c`. So an identifier a source in scope is known to
+    /// have stays a column reference. When no source is known to have it, the
+    /// alias is the only name in sight, and it is taken.
     fn resolve_lateral_alias_sources(
         &self,
         expr: &sqlparser::ast::Expr,
@@ -280,6 +286,13 @@ impl<'a, 'b> SelectAnalyzer<'a, 'b> {
         for ident in &identifiers {
             let normalized_ident = self.analyzer.normalize_identifier(ident);
             if let Some(alias_sources) = lateral_alias_sources.get(&normalized_ident) {
+                if !self
+                    .analyzer
+                    .candidate_tables_for_column(self.ctx, ident)
+                    .is_empty()
+                {
+                    continue;
+                }
                 // This identifier is a lateral alias reference. Add the alias's sources
                 // to our sources list, and remove any ColumnRef that just has the alias
                 // name without a table (since it's not a real table column).

crates/flowscope-core/tests/lineage_engine.rs

@@ -13611,3 +13611,35 @@ fn star_replace_computes_the_column_it_replaces() {
         );
     }
 }
+
+#[test]
+fn a_column_wins_over_an_earlier_alias_of_its_name() {
+    // Snowflake and DuckDB read p.b into c: the table's column wins over the
+    // alias b, alone or inside an expression, through a CTE as well.
+    for dialect in [Dialect::Snowflake, Dialect::Duckdb] {
+        for (sql, output, sources) in [
+            ("SELECT a AS b, b AS c FROM p", "c", vec!["p.b"]),
+            ("SELECT a AS b, UPPER(b) AS c FROM p", "c", vec!["p.b"]),
+            (
+                "SELECT a || c AS b, b || a AS d FROM p",
+                "d",
+                vec!["p.a", "p.b"],
+            ),
+            (
+                "WITH w AS (SELECT a, b FROM p) SELECT a AS b, b AS c FROM w",
+                "c",
+                vec!["p.b"],
+            ),
+            // An alias no source has a column for is still read through.
+            ("SELECT a AS x, x AS y FROM p", "y", vec!["p.a"]),
+        ] {
+            let result = run_analysis(sql, dialect, Some(star_options_schema()));
+            let stmt = first_statement(&result);
+            assert_eq!(
+                base_sources_of(&stmt, output),
+                expected_sources(&sources),
+                "{dialect:?}: {sql}"
+            );
+        }
+    }
+}