fix: Consider column names' case when aliasing tables

datafusion/sql/src/planner.rs

@@ -23,19 +23,19 @@ use std::vec;
 
 use crate::utils::make_decimal_type;
 use arrow::datatypes::*;
-use datafusion_common::TableReference;
 use datafusion_common::config::SqlParserOptions;
 use datafusion_common::datatype::{DataTypeExt, FieldExt};
 use datafusion_common::error::add_possible_columns_to_diag;
+use datafusion_common::{Column, TableReference};
 use datafusion_common::{DFSchema, DataFusionError, Result, not_impl_err, plan_err};
 use datafusion_common::{
     DFSchemaRef, Diagnostic, SchemaError, field_not_found, internal_err,
     plan_datafusion_err,
 };
+use datafusion_expr::Expr;
 use datafusion_expr::logical_plan::{LogicalPlan, LogicalPlanBuilder};
 pub use datafusion_expr::planner::ContextProvider;
 use datafusion_expr::utils::find_column_exprs;
-use datafusion_expr::{Expr, col};
 use sqlparser::ast::{ArrayElemTypeDef, ExactNumberInfo, TimezoneInfo};
 use sqlparser::ast::{ColumnDef as SQLColumnDef, ColumnOption};
 use sqlparser::ast::{DataType as SQLDataType, Ident, ObjectName, TableAlias};
@@ -575,7 +575,8 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             let fields = plan.schema().fields().clone();
             LogicalPlanBuilder::from(plan)
                 .project(fields.iter().zip(idents).map(|(field, ident)| {
-                    col(field.name()).alias(self.ident_normalizer.normalize(ident))
+                    Expr::Column(Column::from_qualified_name_ignore_case(field.name()))
+                        .alias(self.ident_normalizer.normalize(ident))
                 }))?
                 .build()
         }

datafusion/sqllogictest/test_files/alias.slt

@@ -57,3 +57,50 @@ drop table t1;
 
 statement count 0
 drop table t2;
+
+
+# Test table-aliasing a subquery with a case sensitive column
+# (https://github.com/apache/datafusion/issues/22916)
+
+statement ok
+create table t ("A" int);
+
+query I
+select * from (select * from t) t_(x);
+----
+
+query TT
+explain select * from (select * from t) t_(x);
+----
+logical_plan
+01)SubqueryAlias: t_
+02)--Projection: t.A AS x
+03)----TableScan: t projection=[A]
+physical_plan
+01)ProjectionExec: expr=[A@0 as x]
+02)--DataSourceExec: partitions=1, partition_sizes=[0]
+
+query TT
+explain select * from (select * from t) t_(X);
+----
+logical_plan
+01)SubqueryAlias: t_
+02)--Projection: t.A AS x
+03)----TableScan: t projection=[A]
+physical_plan
+01)ProjectionExec: expr=[A@0 as x]
+02)--DataSourceExec: partitions=1, partition_sizes=[0]
+
+query TT
+explain select * from (select * from t) t_("X");
+----
+logical_plan
+01)SubqueryAlias: t_
+02)--Projection: t.A AS X
+03)----TableScan: t projection=[A]
+physical_plan
+01)ProjectionExec: expr=[A@0 as X]
+02)--DataSourceExec: partitions=1, partition_sizes=[0]
+
+statement ok
+drop table t;