Fix: check ambiguous column reference

Author: HuSen8891

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,35 @@ 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))),
+            _ => {
+                let fields_with_qualifier = matches
+                    .iter()
+                    .filter(|(q, _)| q.is_some())
+                    .collect::<Vec<_>>();
+                if fields_with_qualifier.len() == 1 {
+                    Ok((fields_with_qualifier[0].0, fields_with_qualifier[0].1))
+                } else {
+                    _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

@@ -20,7 +20,7 @@ use sqlparser::ast::{Expr as SQLExpr, Ident};
 
 use datafusion_common::{
     internal_err, not_impl_err, plan_datafusion_err, Column, DFSchema, DataFusionError,
-    Result, TableReference,
+    Result, SchemaError, TableReference,
 };
 use datafusion_expr::planner::PlannerResult;
 use datafusion_expr::{Case, Expr};
@@ -186,7 +186,22 @@ 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();
-                            Ok(Expr::Column(Column::new(relation, column_name)))
+                            // sanity check on column
+                            match schema
+                                .qualified_field_with_name(relation.as_ref(), column_name)
+                            {
+                                Err(e) => match e {
+                                    DataFusionError::SchemaError(
+                                        SchemaError::AmbiguousReference { .. },
+                                        _,
+                                    ) => Err(e),
+                                    _ => Ok(Expr::Column(Column::new(
+                                        relation,
+                                        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;