fix: propagate ambiguous column names through projections and joins

Author: xiedeyantu

datafusion/expr/src/logical_plan/builder.rs

@@ -1745,7 +1745,30 @@ pub fn build_join_schema(
         .collect();
 
     let dfschema = DFSchema::new_with_metadata(qualified_fields, metadata)?;
-    dfschema.with_functional_dependencies(func_dependencies)
+    let dfschema = dfschema.with_functional_dependencies(func_dependencies)?;
+
+    // Propagate ambiguous names from both input schemas.  A name that was
+    // already ambiguous on either side of the join (e.g. because the left
+    // input is itself a subquery that wrapped a JOIN) remains ambiguous in
+    // the output.  We only propagate names that actually appear as field
+    // names in the output schema so we don't accumulate stale entries.
+    let output_field_names: HashSet<&str> = dfschema
+        .fields()
+        .iter()
+        .map(|f| f.name().as_str())
+        .collect();
+    let inherited_ambiguous: HashSet<String> = left
+        .ambiguous_names()
+        .iter()
+        .chain(right.ambiguous_names())
+        .filter(|n| output_field_names.contains(n.as_str()))
+        .cloned()
+        .collect();
+    if inherited_ambiguous.is_empty() {
+        Ok(dfschema)
+    } else {
+        Ok(dfschema.with_ambiguous_names(inherited_ambiguous))
+    }
 }
 
 /// (Re)qualify the sides of a join if needed, i.e. if the columns from one side would otherwise

datafusion/expr/src/logical_plan/plan.rs

@@ -2375,6 +2375,29 @@ pub fn projection_schema(input: &LogicalPlan, exprs: &[Expr]) -> Result<Arc<DFSc
                 exprs, input,
             )?)?;
 
+    // Propagate ambiguous names from the input for any column passed through
+    // unchanged.  This prevents a `SELECT * FROM (...) AS alias` wrapper from
+    // silently dropping the ambiguity marker set by an inner JOIN or alias.
+    let input_ambiguous = input.schema().ambiguous_names();
+    if !input_ambiguous.is_empty() {
+        // A column is a pass-through when it is `Expr::Column(c)` and `c.name`
+        // appears in the input's ambiguous set.
+        let inherited: HashSet<String> = exprs
+            .iter()
+            .filter_map(|e| {
+                if let Expr::Column(col) = e {
+                    if input_ambiguous.contains(&col.name) {
+                        return Some(col.name.clone());
+                    }
+                }
+                None
+            })
+            .collect();
+        if !inherited.is_empty() {
+            return Ok(Arc::new(schema.with_ambiguous_names(inherited)));
+        }
+    }
+
     Ok(Arc::new(schema))
 }
 
@@ -2406,23 +2429,37 @@ impl SubqueryAlias {
         let aliases = unique_field_aliases(plan.schema().fields());
         let is_projection_needed = aliases.iter().any(Option::is_some);
 
-        // Collect the set of unqualified field names that are ambiguous in this
-        // subquery alias's output schema.  A name is ambiguous when two or more
-        // input columns share the same unqualified name (they come, say, from
-        // different sides of a JOIN).  `unique_field_aliases` renames the
-        // duplicates to keep the Arrow schema free of duplicates, but we still
-        // need to reject unqualified references to those names from outer
-        // queries.
+        // Collect unqualified field names that are ambiguous in this alias's
+        // output schema.  `unique_field_aliases` renames duplicates (e.g. to
+        // "id:1") to keep Arrow happy, but outer queries must still be
+        // prevented from referencing those names without qualification.
+        // We also inherit names already marked ambiguous by the input schema
+        // so nested `SELECT * FROM (...) AS sN` wrappers don't lose the marker.
         let ambiguous_names: HashSet<String> = {
             let mut name_counts: HashMap<&str, usize> = HashMap::new();
             for field in plan.schema().fields() {
                 *name_counts.entry(field.name().as_str()).or_insert(0) += 1;
             }
-            name_counts
+            let mut names: HashSet<String> = name_counts
                 .into_iter()
                 .filter(|&(_, count)| count >= 2)
                 .map(|(name, _)| name.to_string())
-                .collect()
+                .collect();
+
+            // Inherit names still visible in the output (the first occurrence
+            // of a renamed duplicate like "id:1" still keeps the name "id").
+            let output_field_names: HashSet<&str> = plan
+                .schema()
+                .fields()
+                .iter()
+                .map(|f| f.name().as_str())
+                .collect();
+            for inherited in plan.schema().ambiguous_names() {
+                if output_field_names.contains(inherited.as_str()) {
+                    names.insert(inherited.clone());
+                }
+            }
+            names
         };
 
         // Insert a projection node, if needed, to make sure aliases are applied.

datafusion/sqllogictest/test_files/joins.slt

@@ -5594,6 +5594,52 @@ FROM (SELECT t_left.id, t_left.age, t_right.score, t_extra.dept
 100 eng
 200 sales
 
+# Nested derived table: double SELECT * wrapper must preserve ambiguity.
+# SELECT age FROM ( SELECT * FROM ( SELECT * FROM t_left JOIN t_right ON t_left.id = t_right.id ) AS s1 ) AS s2
+# "age" is ambiguous in s1 (from both t_left and t_right) and must stay
+# ambiguous when s1 is wrapped in another SELECT * … AS s2.
+query error DataFusion error: Schema error: Ambiguous reference to unqualified field age
+SELECT age FROM (
+  SELECT * FROM (
+    SELECT * FROM t_left JOIN t_right ON t_left.id = t_right.id
+  ) AS s1
+) AS s2;
+
+# Nested derived table: same for "id"
+query error DataFusion error: Schema error: Ambiguous reference to unqualified field id
+SELECT id FROM (
+  SELECT * FROM (
+    SELECT * FROM t_left JOIN t_right ON t_left.id = t_right.id
+  ) AS s1
+) AS s2;
+
+# Join over subquery + table: ambiguous names from the subquery side must
+# propagate into the outer join schema so that bare "age" is still rejected.
+# Set up a seed table with a single column so only the subquery side has "age".
+statement ok
+CREATE TABLE seed(val INT) AS VALUES (1), (2);
+
+query error DataFusion error: Schema error: Ambiguous reference to unqualified field age
+SELECT age FROM (SELECT * FROM t_left JOIN t_right ON t_left.id = t_right.id) sub
+JOIN seed ON true;
+
+# Qualified access through the subquery alias is still fine even after joining
+# with another table.
+query II rowsort
+SELECT sub.id, sub.score FROM (
+  SELECT t_left.id, t_right.score
+  FROM t_left JOIN t_right ON t_left.id = t_right.id
+) sub
+JOIN seed ON true;
+----
+1 100
+1 100
+2 200
+2 200
+
+statement ok
+DROP TABLE seed;
+
 statement ok
 DROP TABLE t_left;