Fix normalization of columns in JOIN ... USING. (#16560)

* Fix Column mgmt when parsing USING joins.

In SqlToRel::parse_join(), when handling JoinContraint::Using, the
identifiers are normalized using IdentNormalizer::normalize().

That normalization lower-cases unquoted identifiers, and keeps the case
otherwise (but not the quotes).

Until this commit, the normalized column names were passed to
LogicalPlanBuilder::join_using() as strings. When each goes through
LogicalPlanBuilder::normalize(), Column::From<String>() is called,
leading to Column::from_qualified_named(). As it gets an unqualified
column, it lower-cases it.

This means that if a join is USING("SOME_COLUMN_NAME"), we end up with a
Column { name: "some_column_name", ..}. In the end, the join fails, as
that lower-case column does not exist.

With this commit, SqlToRel::parse_join() calls Column::from_name() on
each normalized column and passed those to
LogicalPlanBuilder::join_using(). Downstream, in
LogicalPlanBuilder::normalize(), there is no need to create the Column
objects from strings, and the bug does not happen.

This fixes #16120.

* Remove genericity from LogicalPlanBuilder::join_using().

Until this commit, LogicalPlanBuilder::join_using() accepted using_keys:
Vec<impl Into<Column> + Clone>.

This commit removes this, only allowing Vec<Column>.

Motivation: passing e.g. Vec<String> for using_keys is bug-prone, as the
Strings can get (their case) modified when made into Column. That logic
is admissible with a common column name that can be qualified, but some
column names cannot (e.g. USING keys).

This commit changes the API. However, potential users can trivially fix
their code by calling Column::from/from_qualified_name on their
using_keys. This forces them to things about what their identifier
represent and that removes a class of potential bugs.

Additional bonus: shorter compilation time & binary size.

---------

Co-authored-by: Bruno Cauet <[email protected]>

Author: brunal

datafusion/core/tests/sql/joins.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use datafusion::assert_batches_eq;
 use datafusion::datasource::stream::{FileStreamProvider, StreamConfig, StreamTable};
 use datafusion::test_util::register_unbounded_file_with_ordering;
 
@@ -235,3 +236,50 @@ async fn join_change_in_planner_without_sort_not_allowed() -> Result<()> {
     }
     Ok(())
 }
+
+#[tokio::test]
+async fn join_using_uppercase_column() -> Result<()> {
+    let schema = Arc::new(Schema::new(vec![Field::new(
+        "UPPER",
+        DataType::UInt32,
+        false,
+    )]));
+    let tmp_dir = TempDir::new()?;
+    let file_path = tmp_dir.path().join("uppercase-column.csv");
+    let mut file = File::create(file_path.clone())?;
+    file.write_all("0".as_bytes())?;
+    drop(file);
+
+    let ctx = SessionContext::new();
+    ctx.register_csv(
+        "test",
+        file_path.to_str().unwrap(),
+        CsvReadOptions::new().schema(&schema).has_header(false),
+    )
+    .await?;
+
+    let dataframe = ctx
+        .sql(
+            r#"
+        SELECT test."UPPER" FROM "test"
+        INNER JOIN (
+            SELECT test."UPPER" FROM "test"
+        ) AS selection USING ("UPPER")
+        ;
+        "#,
+        )
+        .await?;
+
+    assert_batches_eq!(
+        [
+            "+-------+",
+            "| UPPER |",
+            "+-------+",
+            "| 0     |",
+            "+-------+",
+        ],
+        &dataframe.collect().await?
+    );
+
+    Ok(())
+}

datafusion/expr/src/logical_plan/builder.rs

@@ -970,11 +970,7 @@ impl LogicalPlanBuilder {
         )
     }
 
-    pub(crate) fn normalize(
-        plan: &LogicalPlan,
-        column: impl Into<Column>,
-    ) -> Result<Column> {
-        let column = column.into();
+    pub(crate) fn normalize(plan: &LogicalPlan, column: Column) -> Result<Column> {
         if column.relation.is_some() {
             // column is already normalized
             return Ok(column);
@@ -1127,7 +1123,7 @@ impl LogicalPlanBuilder {
         self,
         right: LogicalPlan,
         join_type: JoinType,
-        using_keys: Vec<impl Into<Column> + Clone>,
+        using_keys: Vec<Column>,
     ) -> Result<Self> {
         let left_keys: Vec<Column> = using_keys
             .clone()

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -1827,7 +1827,7 @@ mod tests {
         let table2_scan = scan_empty(Some("test2"), &schema, None)?.build()?;
 
         let plan = LogicalPlanBuilder::from(table_scan)
-            .join_using(table2_scan, JoinType::Left, vec!["a"])?
+            .join_using(table2_scan, JoinType::Left, vec!["a".into()])?
             .project(vec![col("a"), col("b")])?
             .build()?;
 

datafusion/sql/src/relation/join.rs

@@ -142,7 +142,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                                         "Expected identifier in USING clause"
                                     )
                                 })
-                                .map(|ident| self.ident_normalizer.normalize(ident.clone()))
+                                .map(|ident| Column::from_name(self.ident_normalizer.normalize(ident.clone())))
                         }
                     })
                     .collect::<Result<Vec<_>>>()?;