Fix: check ambiguous column reference

datafusion/common/src/dfschema.rs

@@ -419,10 +419,7 @@ impl DFSchema {
         name: &str,
     ) -> Result<(Option<&TableReference>, &Field)> {
         if let Some(qualifier) = qualifier {
-            let idx = self
-                .index_of_column_by_name(Some(qualifier), name)
-                .ok_or_else(|| field_not_found(Some(qualifier.clone()), name, self))?;
-            Ok((self.field_qualifiers[idx].as_ref(), self.field(idx)))
+            self.qualified_field_with_qualified_name(qualifier, name)
         } else {
             self.qualified_field_with_unqualified_name(name)
         }
@@ -467,6 +464,36 @@ impl DFSchema {
             .collect()
     }
 
+    /// Find all fields that match the given name with qualifier and return them with their qualifier
+    pub fn qualified_fields_with_qualified_name(
+        &self,
+        qualifier: &TableReference,
+        name: &str,
+    ) -> Vec<(Option<&TableReference>, &Field)> {
+        self.iter()
+            .filter(|(q, f)| match (qualifier, q) {
+                // field to lookup is qualified.
+                // current field is qualified and not shared between relations, compare both
+                // qualifier and name.
+                (q, Some(field_q)) => q.resolved_eq(field_q) && f.name() == name,
+                // field to lookup is qualified but current field is unqualified.
+                (qq, None) => {
+                    // the original field may now be aliased with a name that matches the
+                    // original qualified name
+                    let column = Column::from_qualified_name(f.name());
+                    match column {
+                        Column {
+                            relation: Some(r),
+                            name: column_name,
+                        } => &r == qq && column_name == name,
+                        _ => false,
+                    }
+                }
+            })
+            .map(|(qualifier, field)| (qualifier, field.as_ref()))
+            .collect()
+    }
+
     /// Find all fields that match the given name and convert to column
     pub fn columns_with_unqualified_name(&self, name: &str) -> Vec<Column> {
         self.iter()
@@ -519,6 +546,25 @@ impl DFSchema {
         }
     }
 
+    /// Find the qualified field with the given qualified name
+    pub fn qualified_field_with_qualified_name(
+        &self,
+        qualifier: &TableReference,
+        name: &str,
+    ) -> Result<(Option<&TableReference>, &Field)> {
+        let matches = self.qualified_fields_with_qualified_name(qualifier, name);
+        match matches.len() {
+            0 => Err(field_not_found(Some(qualifier.clone()), name, self)),
+            1 => Ok((matches[0].0, (matches[0].1))),
+            _ => _schema_err!(SchemaError::AmbiguousReference {
+                field: Column {
+                    relation: None,
+                    name: name.to_string(),
+                },
+            }),
+        }
+    }
+
     /// Find the field with the given name
     pub fn field_with_unqualified_name(&self, name: &str) -> Result<&Field> {
         self.qualified_field_with_unqualified_name(name)

datafusion/sql/src/expr/identifier.rs

@@ -186,6 +186,11 @@ 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.qualified_field_with_name(
+                                relation.as_ref(),
+                                column_name,
+                            )?;
                             Ok(Expr::Column(Column::new(relation, column_name)))
                         }
                     }

datafusion/sqllogictest/test_files/join.slt

@@ -1209,3 +1209,17 @@ 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);
+
+## 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;
+
+statement ok
+drop table t1;