Repository: apache/datafusion (language: rust)

Issue:
Column for primary key not found in schema if constraint column in uppercase
### Describe the bug

Query
```sql
create table test (COLUMN_NAME TEXT NOT NULL,  constraint COLUMN_NAME_PK primary key (COLUMN_NAME))
```
triggers the error 
```Column for primary key not found in schema: COLUMN_NAME```

When creating a table, we build a schema and apply [normalization](https://github.com/apache/datafusion/blob/main/datafusion/sql/src/planner.rs#L260) to column names with to_ascii_lowercase call, which is enabled by default.
But during constraint check we don't do this for the [constraint column name](https://github.com/apache/datafusion/blob/main/datafusion/sql/src/statement.rs#L1459).




### To Reproduce

_No response_

### Expected behavior

The query should work without errors
```sql
create table test (COLUMN_NAME TEXT NOT NULL,  constraint COLUMN_NAME_PK primary key (COLUMN_NAME))
```

### Additional context

Suggestion to use just [to_lowercase](https://github.com/apache/datafusion/blob/main/datafusion/sql/src/statement.rs#L1459) (since we don't pass column names with ascii symbols)
```rust
.position(|item| *item.to_lowercase() == pk.value.to_lowercase())
```


Discussion / hints:
I think we should also apply normalization to constraint columns. Using `to_lowercase` may not work when normalization is disabled.

Fix patch:
diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs
index 74055d979145..fbe6d6501c86 100644
--- a/datafusion/sql/src/statement.rs
+++ b/datafusion/sql/src/statement.rs
@@ -441,7 +441,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                             plan
                         };
 
-                        let constraints = Self::new_constraint_from_table_constraints(
+                        let constraints = self.new_constraint_from_table_constraints(
                             &all_constraints,
                             plan.schema(),
                         )?;
@@ -465,7 +465,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                             schema,
                         };
                         let plan = LogicalPlan::EmptyRelation(plan);
-                        let constraints = Self::new_constraint_from_table_constraints(
+                        let constraints = self.new_constraint_from_table_constraints(
                             &all_constraints,
                             plan.schema(),
                         )?;
@@ -1434,7 +1434,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
 
         let name = self.object_name_to_table_reference(name)?;
         let constraints =
-            Self::new_constraint_from_table_constraints(&all_constraints, &df_schema)?;
+            self.new_constraint_from_table_constraints(&all_constraints, &df_schema)?;
         Ok(LogicalPlan::Ddl(DdlStatement::CreateExternalTable(
             PlanCreateExternalTable {
                 schema: df_schema,
@@ -1454,8 +1454,34 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         )))
     }
 
+    /// Get the indices of the constraint columns in the schema.
+    /// If any column is not found, return an error.
+    fn get_constraint_column_indices(
+        &self,
+        df_schema: &DFSchemaRef,
+        columns: &[Ident],
+        constraint_name: &str,
+    ) -> Result<Vec<usize>> {
+        let field_names = df_schema.field_names();
+        columns
+            .iter()
+            .map(|ident| {
+                let column = self.ident_normalizer.normalize(ident.clone());
+                field_names
+                    .iter()
+                    .position(|item| *item == column)
+                    .ok_or_else(|| {
+                        plan_datafusion_err!(
+                            "Column for {constraint_name} not found in schema: {column}"
+                        )
+                    })
+            })
+            .collect::<Result<Vec<_>>>()
+    }
+
     /// Convert each [TableConstraint] to corresponding [Constraint]
     fn new_constraint_from_table_constraints(
+        &self,
         constraints: &[TableConstraint],
         df_schema: &DFSchemaRef,
     ) -> Result<Constraints> {
@@ -1463,46 +1489,25 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             .iter()
             .map(|c: &TableConstraint| match c {
                 TableConstraint::Unique { name, columns, .. } => {
-                    let field_names = df_schema.field_names();
-                    // Get unique constraint indices in the schema:
-                    let indices = columns
-                        .iter()
-                        .map(|u| {
-                            let idx = field_names
-                                .iter()
-                                .position(|item| *item == u.value)
-                                .ok_or_else(|| {
-                                    let name = name
-                                        .as_ref()
-                                        .map(|name| format!("with name '{name}' "))
-                                        .unwrap_or("".to_string());
-                                    DataFusionError::Execution(
-                                        format!("Column for unique constraint {}not found in schema: {}", name,u.value)
-                                    )
-                                })?;
-                            Ok(idx)
-                        })
-                        .collect::<Result<Vec<_>>>()?;
+                    let constraint_name = match name {
+                        Some(name) => &format!("unique constraint with name '{name}'"),
+                        None => "unique constraint",
+                    };
+                    // Get unique constraint indices in the schema
+                    let indices = self.get_constraint_column_indices(
+                        df_schema,
+                        columns,
+                        constraint_name,
+                    )?;
                     Ok(Constraint::Unique(indices))
                 }
                 TableConstraint::PrimaryKey { columns, .. } => {
-                    let field_names = df_schema.field_names();
-                    // Get primary key indices in the schema:
-                    let indices = columns
-                        .iter()
-                        .map(|pk| {
-                            let idx = field_names
-                                .iter()
-                                .position(|item| *item == pk.value)
-                                .ok_or_else(|| {
-                                    DataFusionError::Execution(format!(
-                                        "Column for primary key not found in schema: {}",
-                                        pk.value
-                                    ))
-                                })?;
-                            Ok(idx)
-                        })
-                        .collect::<Result<Vec<_>>>()?;
+                    // Get primary key indices in the schema
+                    let indices = self.get_constraint_column_indices(
+                        df_schema,
+                        columns,
+                        "primary key",
+                    )?;
                     Ok(Constraint::PrimaryKey(indices))
                 }
                 TableConstraint::ForeignKey { .. } => {


Test patch:
diff --git a/datafusion/sqllogictest/test_files/ddl.slt b/datafusion/sqllogictest/test_files/ddl.slt
index aefc2672b539..6f75a7d7f8fd 100644
--- a/datafusion/sqllogictest/test_files/ddl.slt
+++ b/datafusion/sqllogictest/test_files/ddl.slt
@@ -827,4 +827,3 @@ drop table table_with_pk;
 
 statement ok
 set datafusion.catalog.information_schema = false;
-
diff --git a/datafusion/sqllogictest/test_files/ident_normalization.slt b/datafusion/sqllogictest/test_files/ident_normalization.slt
index 996093c3ad9c..ac2f460ebc43 100644
--- a/datafusion/sqllogictest/test_files/ident_normalization.slt
+++ b/datafusion/sqllogictest/test_files/ident_normalization.slt
@@ -132,3 +132,44 @@ query T
 SELECT CONCAT('Hello', 'World')
 ----
 HelloWorld
+
+# Restore default setting
+statement ok
+set datafusion.sql_parser.enable_ident_normalization = true;
+
+##########
+## Constraint Column Name Normalization
+##########
+
+# Test issue https://github.com/apache/datafusion/issues/14340
+statement ok
+create table test_pk_constraint(COLUMN_NAME TEXT NOT NULL, constraint COLUMN_NAME_PK primary key (COLUMN_NAME))
+
+statement ok
+create table test_unique_constraint(cOlUmn_name TEXT NOT NULL, unique(COLUMN_NAME))
+
+statement ok
+drop table test_pk_constraint;
+
+statement ok
+drop table test_unique_constraint;
+
+# Test with normalization disabled
+statement ok
+set datafusion.sql_parser.enable_ident_normalization = false;
+
+statement error Error during planning: Column for primary key not found in schema: COLUMN_NAME
+create table test_pk_constraint(column_name TEXT NOT NULL, constraint COLUMN_NAME_PK primary key (COLUMN_NAME))
+
+statement ok
+create table test_pk_constraint(COLUMN_NAME TEXT NOT NULL, constraint COLUMN_NAME_PK primary key (COLUMN_NAME))
+
+statement error Error during planning: Column for unique constraint not found in schema: COLUMN_NAME
+create table test_unique_constraint(column_name TEXT NOT NULL, unique(COLUMN_NAME))
+
+statement ok
+drop table test_pk_constraint;
+
+# Restore default setting
+statement ok
+set datafusion.sql_parser.enable_ident_normalization = true;