Fix: check ambiguous column reference

Author: HuSen8891

datafusion/common/src/dfschema.rs

@@ -412,6 +412,33 @@ impl DFSchema {
         }
     }
 
+    /// Check whether the column reference is ambiguous
+    pub fn check_ambiguous_name(
+        &self,
+        qualifier: Option<&TableReference>,
+        name: &str,
+    ) -> Result<()> {
+        let count = self
+            .iter()
+            .filter(|(field_q, f)| match (field_q, qualifier) {
+                (Some(q1), Some(q2)) => q1.resolved_eq(q2) && f.name() == name,
+                (None, None) => f.name() == name,
+                _ => false,
+            })
+            .take(2)
+            .count();
+        if count > 1 {
+            _schema_err!(SchemaError::AmbiguousReference {
+                field: Column {
+                    relation: None,
+                    name: name.to_string(),
+                },
+            })
+        } else {
+            Ok(())
+        }
+    }
+
     /// Find the qualified field with the given name
     pub fn qualified_field_with_name(
         &self,

datafusion/sql/src/expr/identifier.rs

@@ -132,7 +132,12 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                             nested_names,
                         ) {
                             match planner_result {
-                                PlannerResult::Planned(expr) => return Ok(expr),
+                                PlannerResult::Planned(expr) => {
+                                    // sanity check on column
+                                    schema
+                                        .check_ambiguous_name(qualifier, field.name())?;
+                                    return Ok(expr);
+                                }
                                 PlannerResult::Original(_args) => {}
                             }
                         }
@@ -143,6 +148,8 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                 }
                 // found matching field with no spare identifier(s)
                 Some((field, qualifier, _nested_names)) => {
+                    // sanity check on column
+                    schema.check_ambiguous_name(qualifier, field.name())?;
                     Ok(Expr::Column(Column::from((qualifier, field))))
                 }
                 None => {
@@ -186,6 +193,9 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                             let s = &ids[0..ids.len()];
                             // safe unwrap as s can never be empty or exceed the bounds
                             let (relation, column_name) = form_identifier(s).unwrap();
+                            // sanity check on column
+                            schema
+                                .check_ambiguous_name(relation.as_ref(), column_name)?;
                             Ok(Expr::Column(Column::new(relation, column_name)))
                         }
                     }

datafusion/sqllogictest/test_files/join.slt

@@ -1209,3 +1209,20 @@ drop table t1;
 
 statement ok
 drop table t2;
+
+# Test SQLancer issue: https://github.com/apache/datafusion/issues/12337
+statement ok
+create table t1(v1 int) as values(100);
+
+## Query with Ambiguous column reference
+query error DataFusion error: Schema error: Ambiguous reference to unqualified field v1
+select count(*)
+from t1
+right outer join t1
+on t1.v1 > 0;
+
+query error DataFusion error: Schema error: Ambiguous reference to unqualified field v1
+select t1.v1 from t1 join t1 using(v1) cross join (select struct('foo' as v1) as t1);
+
+statement ok
+drop table t1;