diff --git a/crates/integrations/datafusion/src/merge_into.rs b/crates/integrations/datafusion/src/merge_into.rs index 97c360795..1080eeaee 100644 --- a/crates/integrations/datafusion/src/merge_into.rs +++ b/crates/integrations/datafusion/src/merge_into.rs @@ -35,7 +35,7 @@ use datafusion::error::{DataFusionError, Result as DFResult}; use datafusion::prelude::{DataFrame, SessionContext}; use datafusion::sql::sqlparser::ast::{ AssignmentTarget, BinaryOperator, Expr as SqlExpr, Ident, Merge, MergeAction, MergeClauseKind, - MergeInsertKind, TableFactor, + MergeInsertExpr, MergeInsertKind, TableFactor, }; use futures::TryStreamExt; @@ -43,7 +43,7 @@ use paimon::spec::{datums_to_binary_row, extract_datum_from_arrow, CoreOptions, use paimon::table::{CopyOnWriteMergeWriter, DataSplitBuilder, Table, WriteBuilder}; use crate::error::to_datafusion_error; -use crate::sql_context::SQLContext; +use crate::sql_context::{normalize_schema_identifier, SQLContext}; /// Maximum number of retries when DML conflicts with concurrent compaction. const DML_MAX_RETRIES: u32 = 5; @@ -128,14 +128,15 @@ pub(crate) async fn execute_merge_into( ctx: &SQLContext, merge: &Merge, table: Table, + enable_ident_normalization: bool, ) -> DFResult { let schema = table.schema(); let core_options = CoreOptions::new(schema.options()); if core_options.data_evolution_enabled() { - execute_data_evolution_merge(ctx, merge, table).await + execute_data_evolution_merge(ctx, merge, table, enable_ident_normalization).await } else if schema.trimmed_primary_keys().is_empty() { - execute_cow_merge(ctx, merge, table).await + execute_cow_merge(ctx, merge, table, enable_ident_normalization).await } else { Err(DataFusionError::Plan( "MERGE INTO on primary-key tables without data-evolution is not supported".to_string(), @@ -168,9 +169,10 @@ async fn execute_data_evolution_merge( ctx: &SQLContext, merge: &Merge, table: Table, + enable_ident_normalization: bool, ) -> DFResult { retry_on_conflict("MERGE INTO", is_row_id_conflict, || { - execute_merge_into_once(ctx, merge, &table) + execute_merge_into_once(ctx, merge, &table, enable_ident_normalization) }) .await } @@ -198,7 +200,10 @@ enum CowMatchedAction { } /// Parse MERGE clauses for the CoW path (supports DELETE unlike the data-evolution parser). -fn extract_cow_merge_clauses(merge: &Merge) -> DFResult { +fn extract_cow_merge_clauses( + merge: &Merge, + enable_ident_normalization: bool, +) -> DFResult { let mut matched: Vec = Vec::new(); let mut inserts: Vec = Vec::new(); @@ -216,7 +221,12 @@ fn extract_cow_merge_clauses(merge: &Merge) -> DFResult { .0 .last() .and_then(|p| p.as_ident()) - .map(|id| id.value.clone()) + .map(|ident| { + normalize_schema_identifier( + ident, + enable_ident_normalization, + ) + }) .ok_or_else(|| { DataFusionError::Plan(format!( "Invalid column name in SET: {name}" @@ -253,25 +263,11 @@ fn extract_cow_merge_clauses(merge: &Merge) -> DFResult { MergeClauseKind::NotMatched | MergeClauseKind::NotMatchedByTarget => { match &clause.action { MergeAction::Insert(insert_expr) => { - let columns: Vec = - insert_expr.columns.iter().map(|c| c.to_string()).collect(); - let value_exprs = match &insert_expr.kind { - MergeInsertKind::Values(values) => { - if values.rows.is_empty() { - return Err(DataFusionError::Plan( - "INSERT VALUES must have at least one row".to_string(), - )); - } - values.rows[0].iter().map(|e| e.to_string()).collect() - } - MergeInsertKind::Row => Vec::new(), - }; - let predicate = clause.predicate.as_ref().map(|p| p.to_string()); - inserts.push(MergeInsertClause { - columns, - value_exprs, - predicate, - }); + inserts.push(extract_merge_insert_clause( + insert_expr, + clause.predicate.as_ref(), + enable_ident_normalization, + )?); } _ => { return Err(DataFusionError::Plan( @@ -299,9 +295,14 @@ fn extract_cow_merge_clauses(merge: &Merge) -> DFResult { } /// Execute MERGE INTO on an append-only table with retry on delete conflict. -async fn execute_cow_merge(ctx: &SQLContext, merge: &Merge, table: Table) -> DFResult { +async fn execute_cow_merge( + ctx: &SQLContext, + merge: &Merge, + table: Table, + enable_ident_normalization: bool, +) -> DFResult { retry_on_conflict("CoW MERGE INTO", is_delete_conflict, || { - execute_cow_merge_once(ctx, merge, &table) + execute_cow_merge_once(ctx, merge, &table, enable_ident_normalization) }) .await } @@ -311,8 +312,15 @@ async fn execute_cow_merge_once( ctx: &SQLContext, merge: &Merge, table: &Table, + enable_ident_normalization: bool, ) -> DFResult { - let mut clauses = extract_cow_merge_clauses(merge)?; + let mut clauses = extract_cow_merge_clauses(merge, enable_ident_normalization)?; + validate_merge_insert_columns(&clauses.inserts, table.schema().fields())?; + for matched in &clauses.matched { + if let CowMatchedAction::Update(update) = &matched.action { + validate_update_columns(&update.columns, table.schema().fields())?; + } + } // Collect the union of all update columns across matched clauses (preserving order) let mut update_columns: Vec = Vec::new(); @@ -625,9 +633,14 @@ async fn execute_merge_into_once( ctx: &SQLContext, merge: &Merge, table: &Table, + enable_ident_normalization: bool, ) -> DFResult { // 1. Parse all MERGE clauses - let parsed = extract_merge_clauses(merge)?; + let parsed = extract_merge_clauses(merge, enable_ident_normalization)?; + validate_merge_insert_columns(&parsed.inserts, table.schema().fields())?; + if let Some(update) = &parsed.update { + validate_update_columns(&update.columns, table.schema().fields())?; + } // Validate preconditions early and create writer (before executing any SQL) let wb = table.new_write_builder(); @@ -988,19 +1001,18 @@ fn insert_select_clause(ins: &MergeInsertClause, table_fields: &[DataField]) -> "*".to_string() } else { // Build column_name -> expression mapping from the INSERT clause - let col_expr_map: HashMap = ins + let col_expr_map: HashMap<&str, &str> = ins .columns .iter() .zip(ins.value_exprs.iter()) - .map(|(col, expr)| (col.to_lowercase(), expr.as_str())) + .map(|(col, expr)| (col.as_str(), expr.as_str())) .collect(); // Emit SELECT in table schema order table_fields .iter() .map(|field| { - let key = field.name().to_lowercase(); - match col_expr_map.get(&key) { + match col_expr_map.get(field.name()) { Some(expr) => format!("{expr} AS {}", quote_identifier(field.name())), // Column not in INSERT list — fill with NULL None => format!("NULL AS {}", quote_identifier(field.name())), @@ -1021,6 +1033,87 @@ struct MergeInsertClause { predicate: Option, } +fn extract_merge_insert_clause( + insert_expr: &MergeInsertExpr, + predicate: Option<&SqlExpr>, + enable_ident_normalization: bool, +) -> DFResult { + let columns = insert_expr + .columns + .iter() + .map(|name| match name.0.as_slice() { + [part] => part + .as_ident() + .map(|ident| normalize_schema_identifier(ident, enable_ident_normalization)) + .ok_or_else(|| { + DataFusionError::Plan(format!("Invalid column name in MERGE INSERT: {name}")) + }), + _ => Err(DataFusionError::Plan(format!( + "Expected a simple column name in MERGE INSERT, got: {name}" + ))), + }) + .collect::>>()?; + + let value_exprs = match &insert_expr.kind { + MergeInsertKind::Values(values) => { + let row = values.rows.first().ok_or_else(|| { + DataFusionError::Plan("INSERT VALUES must have at least one row".to_string()) + })?; + row.iter().map(ToString::to_string).collect() + } + // INSERT ROW — BigQuery syntax, treat as INSERT *. + MergeInsertKind::Row => Vec::new(), + }; + + Ok(MergeInsertClause { + columns, + value_exprs, + predicate: predicate.map(ToString::to_string), + }) +} + +fn validate_merge_insert_columns( + inserts: &[MergeInsertClause], + table_fields: &[DataField], +) -> DFResult<()> { + for insert in inserts { + validate_target_columns(&insert.columns, table_fields, "MERGE INSERT")?; + } + + Ok(()) +} + +pub(crate) fn validate_update_columns( + columns: &[String], + table_fields: &[DataField], +) -> DFResult<()> { + validate_target_columns(columns, table_fields, "UPDATE") +} + +fn validate_target_columns( + columns: &[String], + table_fields: &[DataField], + operation: &str, +) -> DFResult<()> { + let target_columns: HashSet<&str> = table_fields.iter().map(|field| field.name()).collect(); + let mut seen = HashSet::new(); + + for column in columns { + if !seen.insert(column.as_str()) { + return Err(DataFusionError::Plan(format!( + "Duplicate column '{column}' in {operation}" + ))); + } + if !target_columns.contains(column.as_str()) { + return Err(DataFusionError::Plan(format!( + "Unknown column '{column}' in {operation}" + ))); + } + } + + Ok(()) +} + /// Parsed WHEN MATCHED THEN UPDATE clause. struct MergeUpdateClause { columns: Vec, @@ -1035,7 +1128,10 @@ struct ParsedMergeClauses { } /// Extract UPDATE and INSERT clauses from the MERGE AST. -fn extract_merge_clauses(merge: &Merge) -> DFResult { +fn extract_merge_clauses( + merge: &Merge, + enable_ident_normalization: bool, +) -> DFResult { let mut update: Option = None; let mut delete = false; let mut inserts: Vec = Vec::new(); @@ -1063,7 +1159,12 @@ fn extract_merge_clauses(merge: &Merge) -> DFResult { .0 .last() .and_then(|p| p.as_ident()) - .map(|id| id.value.clone()) + .map(|ident| { + normalize_schema_identifier( + ident, + enable_ident_normalization, + ) + }) .ok_or_else(|| { DataFusionError::Plan(format!( "Invalid column name in SET: {name}" @@ -1094,31 +1195,11 @@ fn extract_merge_clauses(merge: &Merge) -> DFResult { MergeClauseKind::NotMatched | MergeClauseKind::NotMatchedByTarget => { match &clause.action { MergeAction::Insert(insert_expr) => { - let columns: Vec = - insert_expr.columns.iter().map(|c| c.to_string()).collect(); - - let value_exprs = match &insert_expr.kind { - MergeInsertKind::Values(values) => { - if values.rows.is_empty() { - return Err(DataFusionError::Plan( - "INSERT VALUES must have at least one row".to_string(), - )); - } - values.rows[0].iter().map(|e| e.to_string()).collect() - } - MergeInsertKind::Row => { - // INSERT ROW — BigQuery syntax, treat as INSERT * - Vec::new() - } - }; - - let predicate = clause.predicate.as_ref().map(|p| p.to_string()); - - inserts.push(MergeInsertClause { - columns, - value_exprs, - predicate, - }); + inserts.push(extract_merge_insert_clause( + insert_expr, + clause.predicate.as_ref(), + enable_ident_normalization, + )?); } _ => { return Err(DataFusionError::Plan( @@ -1650,6 +1731,49 @@ mod tests { } } + #[test] + fn test_merge_update_assignments_normalize_identifiers() { + for (target, enabled, expected) in [ + ("VALUE", true, "value"), + ("\"VALUE\"", true, "VALUE"), + ("VALUE", false, "VALUE"), + ] { + let merge = parse_merge(&format!( + "MERGE INTO target t USING source s ON t.id = s.id \ + WHEN MATCHED THEN UPDATE SET {target} = s.value" + )); + + let parsed = extract_merge_clauses(&merge, enabled).unwrap(); + assert_eq!(parsed.update.unwrap().columns, [expected]); + + let cow = extract_cow_merge_clauses(&merge, enabled).unwrap(); + let CowMatchedAction::Update(update) = &cow.matched[0].action else { + panic!("expected update action"); + }; + assert_eq!(update.columns, [expected]); + } + } + + #[test] + fn test_merge_insert_columns_normalize_identifiers() { + for (target, enabled, expected) in [ + ("MIXEDCASE", true, "mixedcase"), + ("\"MixedCase\"", true, "MixedCase"), + ("MixedCase", false, "MixedCase"), + ] { + let merge = parse_merge(&format!( + "MERGE INTO target t USING source s ON t.id = s.id \ + WHEN NOT MATCHED THEN INSERT ({target}) VALUES (s.value)" + )); + + let parsed = extract_merge_clauses(&merge, enabled).unwrap(); + assert_eq!(parsed.inserts[0].columns, [expected]); + + let cow = extract_cow_merge_clauses(&merge, enabled).unwrap(); + assert_eq!(cow.inserts[0].columns, [expected]); + } + } + #[test] fn test_normalize_merge_insert_batch_uses_position() { let table_fields = vec![ @@ -1758,9 +1882,9 @@ mod tests { // Execute MERGE INTO let merge = parse_merge( "MERGE INTO paimon.test_db.t_merge t USING paimon.test_db.source s ON t.id = s.id \ - WHEN MATCHED THEN UPDATE SET name = s.name", + WHEN MATCHED THEN UPDATE SET NAME = s.name", ); - execute_merge_into(&sql_context, &merge, table) + execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); @@ -1828,7 +1952,7 @@ mod tests { "MERGE INTO paimon.test_db.t_merge2 t USING paimon.test_db.source s ON t.id = s.id \ WHEN MATCHED THEN UPDATE SET name = s.name", ); - let result = execute_merge_into(&sql_context, &merge, table) + let result = execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); let batches = result.collect().await.unwrap(); @@ -1874,7 +1998,7 @@ mod tests { "MERGE INTO t USING s ON t.id = s.id \ WHEN MATCHED THEN UPDATE SET id = s.id", ); - let result = execute_merge_into(&sql_context, &merge, table).await; + let result = execute_merge_into(&sql_context, &merge, table, true).await; assert!(result.is_err()); assert!(result .unwrap_err() @@ -1958,9 +2082,9 @@ mod tests { let merge = parse_merge( "MERGE INTO paimon.test_db.t_cow_upd t USING paimon.test_db.source s ON t.id = s.id \ - WHEN MATCHED THEN UPDATE SET name = s.name", + WHEN MATCHED THEN UPDATE SET NAME = s.name", ); - execute_merge_into(&sql_context, &merge, table) + execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); @@ -2003,7 +2127,7 @@ mod tests { "MERGE INTO paimon.test_db.t_cow_del t USING paimon.test_db.source s ON t.id = s.id \ WHEN MATCHED THEN DELETE", ); - execute_merge_into(&sql_context, &merge, table) + execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); @@ -2042,7 +2166,7 @@ mod tests { "MERGE INTO paimon.test_db.t_cow_ins t USING paimon.test_db.source s ON t.id = s.id \ WHEN NOT MATCHED THEN INSERT (id, name, value) VALUES (s.id, s.name, s.value)", ); - execute_merge_into(&sql_context, &merge, table) + execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); @@ -2088,7 +2212,7 @@ mod tests { WHEN MATCHED THEN UPDATE SET name = s.name, value = s.value \ WHEN NOT MATCHED THEN INSERT (id, name, value) VALUES (s.id, s.name, s.value)", ); - execute_merge_into(&sql_context, &merge, table) + execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); @@ -2128,7 +2252,7 @@ mod tests { "MERGE INTO paimon.test_db.t_cow_nomatch t USING paimon.test_db.source s ON t.id = s.id \ WHEN MATCHED THEN UPDATE SET name = s.name", ); - let result = execute_merge_into(&sql_context, &merge, table) + let result = execute_merge_into(&sql_context, &merge, table, true) .await .unwrap(); let batches = result.collect().await.unwrap(); diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index 1803892b2..cd54a08e6 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -62,7 +62,7 @@ use datafusion::sql::planner::IdentNormalizer; use datafusion::sql::sqlparser::ast::{ AlterColumnOperation, AlterTableOperation, BinaryLength, CharacterLength, ColumnDef, ColumnOption, CreateFunction, CreateFunctionBody, CreateTable, CreateTableOptions, CreateView, - Delete, Expr as SqlExpr, FromTable, FunctionBehavior, FunctionReturnType, Insert, Merge, + Delete, Expr as SqlExpr, FromTable, FunctionBehavior, FunctionReturnType, Ident, Insert, Merge, ObjectName, ObjectType, RenameTableNameKind, Reset, ResetStatement, Set, ShowCreateObject, SqlOption, Statement, TableFactor, TableObject, Truncate, Update, Use, Value as SqlValue, }; @@ -368,8 +368,9 @@ impl SQLContext { /// directly; everything else is delegated to DataFusion. pub async fn sql(&self, sql: &str) -> DFResult { let is_create_table = looks_like_create_table(sql); + let enable_ident_normalization = self.ctx.enable_ident_normalization(); let (rewritten_sql, partition_keys) = if is_create_table { - extract_partition_by(sql)? + extract_partition_by(sql, enable_ident_normalization)? } else { (sql.to_string(), vec![]) }; @@ -435,8 +436,13 @@ impl SQLContext { } else { let (catalog, _catalog_name, _) = self.resolve_catalog_and_table(&create_table.name)?; - self.handle_create_table(&catalog, create_table, partition_keys) - .await + self.handle_create_table( + &catalog, + create_table, + partition_keys, + enable_ident_normalization, + ) + .await } } Statement::ShowCreate { @@ -451,17 +457,24 @@ impl SQLContext { &alter_table.name, &alter_table.operations, alter_table.if_exists, + enable_ident_normalization, ) .await } - Statement::Merge(merge) => self.handle_merge_into(merge).await, - Statement::Update(update) => self.handle_update(update).await, + Statement::Merge(merge) => { + self.handle_merge_into(merge, enable_ident_normalization) + .await + } + Statement::Update(update) => { + self.handle_update(update, enable_ident_normalization).await + } Statement::Delete(delete) => self.handle_delete(delete).await, Statement::Insert(insert) if insert.overwrite && insert.partitioned.as_ref().is_some_and(|p| !p.is_empty()) => { - self.handle_insert_overwrite_partition(insert).await + self.handle_insert_overwrite_partition(insert, enable_ident_normalization) + .await } Statement::Set(Set::SingleAssignment { variable, values, .. @@ -497,7 +510,10 @@ impl SQLContext { } self.ctx.sql(sql).await } - Statement::Truncate(truncate) => self.handle_truncate_table(truncate).await, + Statement::Truncate(truncate) => { + self.handle_truncate_table(truncate, enable_ident_normalization) + .await + } Statement::CreateView(create_view) => { if create_view.temporary { // Temporary views are always handled by us (Paimon catalog temp storage) @@ -845,6 +861,7 @@ impl SQLContext { catalog: &Arc, ct: &CreateTable, partition_keys: Vec, + enable_ident_normalization: bool, ) -> DFResult { if ct.external { return Err(DataFusionError::Plan( @@ -866,12 +883,17 @@ impl SQLContext { let mut builder = paimon::spec::Schema::builder(); let table_options = extract_options(&ct.table_options)?; + let column_names: Vec = ct + .columns + .iter() + .map(|column| normalize_schema_identifier(&column.name, enable_ident_normalization)) + .collect(); // Columns - for col in &ct.columns { + for (col, column_name) in ct.columns.iter().zip(&column_names) { let paimon_type = column_def_to_paimon_type(col)?; let comment = column_def_comment(col); - builder = builder.column_with_description(col.name.value.clone(), paimon_type, comment); + builder = builder.column_with_description(column_name.clone(), paimon_type, comment); } // Primary key from constraints: PRIMARY KEY (col, ...) @@ -880,7 +902,7 @@ impl SQLContext { let pk_cols: Vec = pk .columns .iter() - .map(|c| primary_key_column_name(&c.column.expr)) + .map(|c| primary_key_column_name(&c.column.expr, enable_ident_normalization)) .collect(); builder = builder.primary_key(pk_cols); } @@ -888,9 +910,8 @@ impl SQLContext { // Partition keys (extracted and validated before parsing) if !partition_keys.is_empty() { - let col_names: Vec<&str> = ct.columns.iter().map(|c| c.name.value.as_str()).collect(); for pk in &partition_keys { - if !col_names.contains(&pk.as_str()) { + if !column_names.contains(pk) { return Err(DataFusionError::Plan(format!( "PARTITIONED BY column '{pk}' is not defined in the table" ))); @@ -1164,6 +1185,7 @@ impl SQLContext { name: &ObjectName, operations: &[AlterTableOperation], if_exists: bool, + enable_ident_normalization: bool, ) -> DFResult { Self::ensure_main_branch_write_target(name, "ALTER TABLE")?; let identifier = self.resolve_table_name(name)?; @@ -1188,11 +1210,17 @@ impl SQLContext { for op in operations { match op { AlterTableOperation::AddColumn { column_def, .. } => { - changes.push(column_def_to_add_column(column_def)?); + changes.push(column_def_to_add_column( + column_def, + enable_ident_normalization, + )?); } AlterTableOperation::DropColumn { column_names, .. } => { for col in column_names { - changes.push(SchemaChange::drop_column(col.value.clone())); + changes.push(SchemaChange::drop_column(normalize_schema_identifier( + col, + enable_ident_normalization, + ))); } } AlterTableOperation::RenameColumn { @@ -1200,12 +1228,16 @@ impl SQLContext { new_column_name, } => { changes.push(SchemaChange::rename_column( - old_column_name.value.clone(), - new_column_name.value.clone(), + normalize_schema_identifier(old_column_name, enable_ident_normalization), + normalize_schema_identifier(new_column_name, enable_ident_normalization), )); } AlterTableOperation::AlterColumn { column_name, op } => { - changes.push(alter_column_to_schema_change(&column_name.value, op)?); + changes.push(alter_column_to_schema_change( + column_name, + op, + enable_ident_normalization, + )?); } AlterTableOperation::RenameTable { table_name } => { let new_name = match table_name { @@ -1238,6 +1270,7 @@ impl SQLContext { &identifier, partitions, if_exists || *partition_if_exists, + enable_ident_normalization, ) .await; } @@ -1298,7 +1331,11 @@ impl SQLContext { Ok(()) } - async fn handle_merge_into(&self, merge: &Merge) -> DFResult { + async fn handle_merge_into( + &self, + merge: &Merge, + enable_ident_normalization: bool, + ) -> DFResult { self.ensure_no_time_travel_for_write("MERGE INTO")?; let table_name = match &merge.table { TableFactor::Table { name, .. } => name.clone(), @@ -1316,10 +1353,14 @@ impl SQLContext { .await .map_err(to_datafusion_error)?; - crate::merge_into::execute_merge_into(self, merge, table).await + crate::merge_into::execute_merge_into(self, merge, table, enable_ident_normalization).await } - async fn handle_update(&self, update: &Update) -> DFResult { + async fn handle_update( + &self, + update: &Update, + enable_ident_normalization: bool, + ) -> DFResult { self.ensure_no_time_travel_for_write("UPDATE")?; let table_name = match &update.table.relation { TableFactor::Table { name, .. } => name.clone(), @@ -1337,7 +1378,7 @@ impl SQLContext { .await .map_err(to_datafusion_error)?; - crate::update::execute_update(self, update, table).await + crate::update::execute_update(self, update, table, enable_ident_normalization).await } async fn handle_delete(&self, delete: &Delete) -> DFResult { @@ -1369,7 +1410,11 @@ impl SQLContext { crate::delete::execute_delete(self, delete, table, &table_ref).await } - async fn handle_insert_overwrite_partition(&self, insert: &Insert) -> DFResult { + async fn handle_insert_overwrite_partition( + &self, + insert: &Insert, + enable_ident_normalization: bool, + ) -> DFResult { self.ensure_no_time_travel_for_write("INSERT OVERWRITE")?; let table_name = match &insert.table { TableObject::TableName(name) => name.clone(), @@ -1390,8 +1435,12 @@ impl SQLContext { DataFusionError::Plan("INSERT OVERWRITE PARTITION requires a PARTITION clause".into()) })?; let partition_fields = table.schema().partition_fields(); - let static_partitions = - parse_static_partitions(partition_exprs, &partition_fields, table.schema().fields())?; + let static_partitions = parse_static_partitions( + partition_exprs, + &partition_fields, + table.schema().fields(), + enable_ident_normalization, + )?; let source = insert.source.as_ref().ok_or_else(|| { DataFusionError::Plan("INSERT OVERWRITE requires a source query".into()) @@ -1412,7 +1461,7 @@ impl SQLContext { insert .columns .iter() - .map(object_name_to_single_identifier) + .map(|name| object_name_to_single_identifier(name, enable_ident_normalization)) .collect::>()?, ) } else if !insert.after_columns.is_empty() { @@ -1420,7 +1469,7 @@ impl SQLContext { insert .after_columns .iter() - .map(|ident| ident.value.clone()) + .map(|ident| normalize_schema_identifier(ident, enable_ident_normalization)) .collect(), ) } else { @@ -1514,7 +1563,11 @@ impl SQLContext { crate::merge_into::ok_result(&self.ctx, row_count) } - async fn handle_truncate_table(&self, truncate: &Truncate) -> DFResult { + async fn handle_truncate_table( + &self, + truncate: &Truncate, + enable_ident_normalization: bool, + ) -> DFResult { self.ensure_no_time_travel_for_write("TRUNCATE TABLE")?; if truncate.table_names.len() > 1 { return Err(DataFusionError::Plan( @@ -1547,6 +1600,7 @@ impl SQLContext { partitions, table.schema().fields(), table.schema().partition_keys(), + enable_ident_normalization, )?; commit .truncate_partitions(partition_values) @@ -1803,6 +1857,7 @@ impl SQLContext { identifier: &Identifier, partitions: &[SqlExpr], if_exists: bool, + enable_ident_normalization: bool, ) -> DFResult { if partitions.is_empty() { return Err(DataFusionError::Plan( @@ -1821,6 +1876,7 @@ impl SQLContext { partitions, table.schema().fields(), table.schema().partition_keys(), + enable_ident_normalization, )?; let wb = table.new_write_builder(); @@ -2406,7 +2462,7 @@ fn find_partitioned_by(sql: &str) -> Option<(usize, usize)> { } /// Parse a single partition column token, handling quoted identifiers. -fn parse_partition_column(token: &str) -> DFResult { +fn parse_partition_column(token: &str, enable_ident_normalization: bool) -> DFResult { let trimmed = token.trim(); if trimmed.is_empty() { return Err(DataFusionError::Plan( @@ -2434,7 +2490,10 @@ fn parse_partition_column(token: &str) -> DFResult { } if let Some(end) = end { if trimmed[end..].trim().is_empty() { - return Ok(value); + return Ok(normalize_schema_identifier( + &Ident::with_quote(first as char, value), + enable_ident_normalization, + )); } } return Err(DataFusionError::Plan(format!( @@ -2444,7 +2503,10 @@ fn parse_partition_column(token: &str) -> DFResult { let parts: Vec<&str> = trimmed.split_whitespace().collect(); match parts.len() { - 1 => Ok(parts[0].to_string()), + 1 => Ok(normalize_schema_identifier( + &Ident::new(parts[0]), + enable_ident_normalization, + )), _ => Err(DataFusionError::Plan(format!( "PARTITIONED BY column '{}' should not specify a type. \ Use column references only, e.g. PARTITIONED BY ({})", @@ -2491,7 +2553,10 @@ fn split_partition_columns(inner: &str) -> DFResult> { /// Since sqlparser's GenericDialect requires types in column definitions, /// we extract and validate the clause ourselves, then strip it from the SQL /// so sqlparser can parse the rest. -fn extract_partition_by(sql: &str) -> DFResult<(String, Vec)> { +fn extract_partition_by( + sql: &str, + enable_ident_normalization: bool, +) -> DFResult<(String, Vec)> { let Some((kw_start, by_end)) = find_partitioned_by(sql) else { return Ok((sql.to_string(), vec![])); }; @@ -2545,7 +2610,7 @@ fn extract_partition_by(sql: &str) -> DFResult<(String, Vec)> { let mut partition_keys = Vec::new(); for token in split_partition_columns(inner)? { - partition_keys.push(parse_partition_column(token)?); + partition_keys.push(parse_partition_column(token, enable_ident_normalization)?); } let clause_end = paren_end + 1; @@ -2555,13 +2620,26 @@ fn extract_partition_by(sql: &str) -> DFResult<(String, Vec)> { Ok((rewritten, partition_keys)) } +pub(crate) fn normalize_schema_identifier( + identifier: &Ident, + enable_ident_normalization: bool, +) -> String { + IdentNormalizer::new(enable_ident_normalization).normalize(identifier.clone()) +} + /// Convert a sqlparser [`ColumnDef`] to a Paimon [`SchemaChange::AddColumn`]. -fn column_def_to_add_column(col: &ColumnDef) -> DFResult { +fn column_def_to_add_column( + col: &ColumnDef, + enable_ident_normalization: bool, +) -> DFResult { let paimon_type = column_def_to_paimon_type(col)?; let comment = column_def_comment(col); Ok(SchemaChange::AddColumn { - field_names: vec![col.name.value.clone()], + field_names: vec![normalize_schema_identifier( + &col.name, + enable_ident_normalization, + )], data_type: paimon_type, comment, column_move: None, @@ -2569,18 +2647,19 @@ fn column_def_to_add_column(col: &ColumnDef) -> DFResult { } fn alter_column_to_schema_change( - column_name: &str, + column_name: &Ident, operation: &AlterColumnOperation, + enable_ident_normalization: bool, ) -> DFResult { + let column_name = normalize_schema_identifier(column_name, enable_ident_normalization); + match operation { - AlterColumnOperation::SetNotNull => Ok(SchemaChange::update_column_nullability( - column_name.to_string(), - false, - )), - AlterColumnOperation::DropNotNull => Ok(SchemaChange::update_column_nullability( - column_name.to_string(), - true, - )), + AlterColumnOperation::SetNotNull => { + Ok(SchemaChange::update_column_nullability(column_name, false)) + } + AlterColumnOperation::DropNotNull => { + Ok(SchemaChange::update_column_nullability(column_name, true)) + } AlterColumnOperation::SetDataType { data_type, using, .. } => { @@ -2591,7 +2670,7 @@ fn alter_column_to_schema_change( } let new_data_type = sql_data_type_to_paimon_type(data_type, true)?; Ok(SchemaChange::UpdateColumnType { - field_names: vec![column_name.to_string()], + field_names: vec![column_name], new_data_type, // A type-only SQL change must not change the column's nullability. keep_nullability: true, @@ -2614,9 +2693,11 @@ fn column_def_comment(col: &ColumnDef) -> Option { }) } -fn primary_key_column_name(expr: &SqlExpr) -> String { +fn primary_key_column_name(expr: &SqlExpr, enable_ident_normalization: bool) -> String { match expr { - SqlExpr::Identifier(ident) => ident.value.clone(), + SqlExpr::Identifier(ident) => { + normalize_schema_identifier(ident, enable_ident_normalization) + } _ => expr.to_string(), } } @@ -2829,11 +2910,14 @@ fn object_name_to_string(name: &ObjectName) -> String { .join(".") } -fn object_name_to_single_identifier(name: &ObjectName) -> DFResult { +fn object_name_to_single_identifier( + name: &ObjectName, + enable_ident_normalization: bool, +) -> DFResult { match name.0.as_slice() { [part] => part .as_ident() - .map(|id| id.value.clone()) + .map(|ident| normalize_schema_identifier(ident, enable_ident_normalization)) .ok_or_else(|| DataFusionError::Plan(format!("Invalid column name: {name}"))), _ => Err(DataFusionError::Plan(format!( "Expected a simple column name, got: {name}" @@ -2883,11 +2967,13 @@ fn parse_partition_values( exprs: &[SqlExpr], all_fields: &[PaimonDataField], partition_keys: &[String], + enable_ident_normalization: bool, ) -> DFResult>>> { let field_map: HashMap<&str, &PaimonDataField> = all_fields.iter().map(|f| (f.name(), f)).collect(); let mut partition = HashMap::new(); + let mut seen_columns = HashSet::new(); for expr in exprs { let (col_name, val_expr) = match expr { SqlExpr::BinaryOp { @@ -2896,7 +2982,9 @@ fn parse_partition_values( right, } => { let col = match left.as_ref() { - SqlExpr::Identifier(ident) => ident.value.clone(), + SqlExpr::Identifier(ident) => { + normalize_schema_identifier(ident, enable_ident_normalization) + } other => { return Err(DataFusionError::Plan(format!( "Expected column name in partition spec, got: {other}" @@ -2912,6 +3000,11 @@ fn parse_partition_values( } }; + if !seen_columns.insert(col_name.clone()) { + return Err(DataFusionError::Plan(format!( + "Duplicate partition column '{col_name}'" + ))); + } if !partition_keys.iter().any(|k| k == &col_name) { return Err(DataFusionError::Plan(format!( "Column '{col_name}' is not a partition column" @@ -2947,8 +3040,10 @@ fn parse_static_partitions( exprs: &[SqlExpr], partition_fields: &[PaimonDataField], all_fields: &[PaimonDataField], + enable_ident_normalization: bool, ) -> DFResult>> { let mut result = HashMap::new(); + let mut seen_columns = HashSet::new(); let field_map: HashMap<&str, &PaimonDataField> = all_fields.iter().map(|f| (f.name(), f)).collect(); let partition_names: Vec<&str> = partition_fields.iter().map(|f| f.name()).collect(); @@ -2961,25 +3056,20 @@ fn parse_static_partitions( right, } => { let col = match left.as_ref() { - SqlExpr::Identifier(ident) => ident.value.clone(), + SqlExpr::Identifier(ident) => { + normalize_schema_identifier(ident, enable_ident_normalization) + } other => { return Err(DataFusionError::Plan(format!( "Expected column name in PARTITION clause, got: {other}" ))) } }; - (col, right.as_ref()) + (col, Some(right.as_ref())) } - // Dynamic partition: bare column name without value — skip it, - // the column will be read from the source query. SqlExpr::Identifier(ident) => { - let col_name = &ident.value; - if !partition_names.contains(&col_name.as_str()) { - return Err(DataFusionError::Plan(format!( - "Column '{col_name}' is not a partition column" - ))); - } - continue; + let col = normalize_schema_identifier(ident, enable_ident_normalization); + (col, None) } other => { return Err(DataFusionError::Plan(format!( @@ -2988,12 +3078,21 @@ fn parse_static_partitions( } }; + if !seen_columns.insert(col_name.clone()) { + return Err(DataFusionError::Plan(format!( + "Duplicate partition column '{col_name}'" + ))); + } if !partition_names.contains(&col_name.as_str()) { return Err(DataFusionError::Plan(format!( "Column '{col_name}' is not a partition column" ))); } + // Dynamic partition columns are read from the source query. + let Some(val_expr) = val_expr else { + continue; + }; let field = field_map.get(col_name.as_str()).ok_or_else(|| { DataFusionError::Plan(format!("Column '{col_name}' not found in table schema")) })?; @@ -6137,6 +6236,50 @@ mod tests { } } + #[tokio::test] + async fn test_create_table_normalizes_schema_identifiers() { + let catalog = Arc::new(MockCatalog::new()); + let sql_context = make_sql_context(catalog.clone()).await; + + sql_context + .sql( + "CREATE TABLE mydb.t1 ( + ID INT NOT NULL, + PART STRING, + PRIMARY KEY (ID) + ) PARTITIONED BY (PART)", + ) + .await + .unwrap(); + + let calls = catalog.take_calls(); + let [CatalogCall::CreateTable { schema, .. }] = calls.as_slice() else { + panic!("expected one CreateTable call, got {calls:?}"); + }; + let field_names: Vec<&str> = schema.fields().iter().map(|field| field.name()).collect(); + assert_eq!(field_names, ["id", "part"]); + assert_eq!(schema.primary_keys(), ["id"]); + assert_eq!(schema.partition_keys(), ["part"]); + } + + #[tokio::test] + async fn test_create_table_preserves_quoted_schema_identifiers() { + let catalog = Arc::new(MockCatalog::new()); + let sql_context = make_sql_context(catalog.clone()).await; + + sql_context + .sql("CREATE TABLE mydb.t1 (\"ID\" INT NOT NULL, PRIMARY KEY (\"ID\"))") + .await + .unwrap(); + + let calls = catalog.take_calls(); + let [CatalogCall::CreateTable { schema, .. }] = calls.as_slice() else { + panic!("expected one CreateTable call, got {calls:?}"); + }; + assert_eq!(schema.fields()[0].name(), "ID"); + assert_eq!(schema.primary_keys(), ["ID"]); + } + #[tokio::test] async fn test_create_table_if_not_exists() { let catalog = Arc::new(MockCatalog::new()); @@ -6284,7 +6427,7 @@ mod tests { let sql_context = make_sql_context(catalog.clone()).await; sql_context - .sql("ALTER TABLE mydb.t1 ADD COLUMN age INT") + .sql("ALTER TABLE mydb.t1 ADD COLUMN AGE INT") .await .unwrap(); @@ -6370,7 +6513,7 @@ mod tests { let sql_context = make_sql_context(catalog.clone()).await; sql_context - .sql("ALTER TABLE mydb.t1 DROP COLUMN age") + .sql("ALTER TABLE mydb.t1 DROP COLUMN AGE") .await .unwrap(); @@ -6392,7 +6535,7 @@ mod tests { let sql_context = make_sql_context(catalog.clone()).await; sql_context - .sql("ALTER TABLE mydb.t1 RENAME COLUMN old_name TO new_name") + .sql("ALTER TABLE mydb.t1 RENAME COLUMN OLD_NAME TO NEW_NAME") .await .unwrap(); @@ -6416,8 +6559,8 @@ mod tests { let sql_context = make_sql_context(catalog.clone()).await; for sql in [ - "ALTER TABLE mydb.t1 ALTER COLUMN value TYPE BIGINT", - "ALTER TABLE mydb.t1 ALTER COLUMN value SET DATA TYPE BIGINT", + "ALTER TABLE mydb.t1 ALTER COLUMN VALUE TYPE BIGINT", + "ALTER TABLE mydb.t1 ALTER COLUMN VALUE SET DATA TYPE BIGINT", ] { sql_context.sql(sql).await.unwrap(); @@ -6445,8 +6588,8 @@ mod tests { let sql_context = make_sql_context(catalog.clone()).await; for (sql, expected_nullability) in [ - ("ALTER TABLE mydb.t1 ALTER COLUMN value SET NOT NULL", false), - ("ALTER TABLE mydb.t1 ALTER COLUMN value DROP NOT NULL", true), + ("ALTER TABLE mydb.t1 ALTER COLUMN VALUE SET NOT NULL", false), + ("ALTER TABLE mydb.t1 ALTER COLUMN VALUE DROP NOT NULL", true), ] { sql_context.sql(sql).await.unwrap(); @@ -6484,11 +6627,64 @@ mod tests { panic!("expected ALTER COLUMN operation"); }; - let err = alter_column_to_schema_change(&column_name.value, op).unwrap_err(); + let err = alter_column_to_schema_change(column_name, op, true).unwrap_err(); assert!(err.to_string().contains("USING is not supported")); } + #[tokio::test] + async fn test_alter_table_preserves_quoted_schema_identifier() { + let catalog = Arc::new(MockCatalog::new()); + let sql_context = make_sql_context(catalog.clone()).await; + + sql_context + .sql("ALTER TABLE mydb.t1 ALTER COLUMN \"MixedCase\" TYPE BIGINT") + .await + .unwrap(); + + let calls = catalog.take_calls(); + let [CatalogCall::AlterTable { changes, .. }] = calls.as_slice() else { + panic!("expected one AlterTable call, got {calls:?}"); + }; + assert!(matches!( + &changes[0], + SchemaChange::UpdateColumnType { field_names, .. } + if field_names == &["MixedCase"] + )); + } + + #[tokio::test] + async fn test_schema_identifier_normalization_can_be_disabled() { + let catalog = Arc::new(MockCatalog::new()); + let sql_context = make_sql_context(catalog.clone()).await; + + sql_context + .sql("SET datafusion.sql_parser.enable_ident_normalization = false") + .await + .unwrap(); + sql_context + .sql("CREATE TABLE mydb.t1 (VALUE INT)") + .await + .unwrap(); + sql_context + .sql("ALTER TABLE mydb.t1 ALTER COLUMN VALUE TYPE BIGINT") + .await + .unwrap(); + + let calls = catalog.take_calls(); + let [CatalogCall::CreateTable { schema, .. }, CatalogCall::AlterTable { changes, .. }] = + calls.as_slice() + else { + panic!("expected CreateTable and AlterTable calls, got {calls:?}"); + }; + assert_eq!(schema.fields()[0].name(), "VALUE"); + assert!(matches!( + &changes[0], + SchemaChange::UpdateColumnType { field_names, .. } + if field_names == &["VALUE"] + )); + } + #[tokio::test] async fn test_alter_table_update_column_default_is_unsupported() { let catalog = Arc::new(MockCatalog::new()); @@ -6687,7 +6883,7 @@ mod tests { #[test] fn test_extract_partition_by_no_clause() { - let (rewritten, keys) = extract_partition_by("CREATE TABLE t (id INT)").unwrap(); + let (rewritten, keys) = extract_partition_by("CREATE TABLE t (id INT)", true).unwrap(); assert_eq!(rewritten, "CREATE TABLE t (id INT)"); assert!(keys.is_empty()); } @@ -6696,6 +6892,7 @@ mod tests { fn test_extract_partition_by_single_column() { let (rewritten, keys) = extract_partition_by( "CREATE TABLE t (id INT, dt STRING) PARTITIONED BY (dt) WITH ('k'='v')", + true, ) .unwrap(); assert_eq!(keys, vec!["dt"]); @@ -6705,35 +6902,65 @@ mod tests { #[test] fn test_extract_partition_by_multiple_columns() { + let (_, keys) = extract_partition_by( + "CREATE TABLE t (a INT, b INT, c INT) PARTITIONED BY (a, b)", + true, + ) + .unwrap(); + assert_eq!(keys, vec!["a", "b"]); + } + + #[test] + fn test_extract_partition_by_normalizes_identifiers() { + let (_, keys) = + extract_partition_by("CREATE TABLE t (PART INT) PARTITIONED BY (PART)", true).unwrap(); + assert_eq!(keys, ["part"]); + + let (_, keys) = extract_partition_by( + "CREATE TABLE t (\"PART\" INT) PARTITIONED BY (\"PART\")", + true, + ) + .unwrap(); + assert_eq!(keys, ["PART"]); + let (_, keys) = - extract_partition_by("CREATE TABLE t (a INT, b INT, c INT) PARTITIONED BY (a, b)") + extract_partition_by("CREATE TABLE t (`PART` INT) PARTITIONED BY (`PART`)", true) .unwrap(); - assert_eq!(keys, vec!["a", "b"]); + assert_eq!(keys, ["PART"]); + + let (_, keys) = + extract_partition_by("CREATE TABLE t (PART INT) PARTITIONED BY (PART)", false).unwrap(); + assert_eq!(keys, ["PART"]); } #[test] fn test_extract_partition_by_mixed_case() { let (_, keys) = - extract_partition_by("CREATE TABLE t (dt INT) Partitioned by (dt)").unwrap(); + extract_partition_by("CREATE TABLE t (dt INT) Partitioned by (dt)", true).unwrap(); assert_eq!(keys, vec!["dt"]); } #[test] fn test_extract_partition_by_rejects_typed_column() { - let err = extract_partition_by("CREATE TABLE t (dt STRING) PARTITIONED BY (dt STRING)") - .unwrap_err(); + let err = extract_partition_by( + "CREATE TABLE t (dt STRING) PARTITIONED BY (dt STRING)", + true, + ) + .unwrap_err(); assert!(err.to_string().contains("should not specify a type")); } #[test] fn test_extract_partition_by_empty_parens() { - let err = extract_partition_by("CREATE TABLE t (id INT) PARTITIONED BY ()").unwrap_err(); + let err = + extract_partition_by("CREATE TABLE t (id INT) PARTITIONED BY ()", true).unwrap_err(); assert!(err.to_string().contains("at least one column")); } #[test] fn test_extract_partition_by_unmatched_paren() { - let err = extract_partition_by("CREATE TABLE t (id INT) PARTITIONED BY (dt").unwrap_err(); + let err = + extract_partition_by("CREATE TABLE t (id INT) PARTITIONED BY (dt", true).unwrap_err(); assert!(err.to_string().contains("Unmatched")); } @@ -6741,7 +6968,7 @@ mod tests { fn test_extract_partition_by_skips_string_literal() { let sql = "CREATE TABLE t (id INT) WITH ('note' = 'PARTITIONED BY (x)') PARTITIONED BY (id)"; - let (rewritten, keys) = extract_partition_by(sql).unwrap(); + let (rewritten, keys) = extract_partition_by(sql, true).unwrap(); assert_eq!(keys, vec!["id"]); assert!(rewritten.contains("WITH")); assert!(rewritten.contains("'PARTITIONED BY (x)'")); @@ -6750,15 +6977,17 @@ mod tests { #[test] fn test_extract_partition_by_skips_line_comment() { let sql = "CREATE TABLE t (id INT) -- PARTITIONED BY (x)\nPARTITIONED BY (id)"; - let (_, keys) = extract_partition_by(sql).unwrap(); + let (_, keys) = extract_partition_by(sql, true).unwrap(); assert_eq!(keys, vec!["id"]); } #[test] fn test_extract_partition_by_double_quoted_identifier() { - let (_, keys) = - extract_partition_by("CREATE TABLE t (\"order\" INT) PARTITIONED BY (\"order\")") - .unwrap(); + let (_, keys) = extract_partition_by( + "CREATE TABLE t (\"order\" INT) PARTITIONED BY (\"order\")", + true, + ) + .unwrap(); assert_eq!(keys, vec!["order"]); } @@ -6767,6 +6996,7 @@ mod tests { let (_, keys) = extract_partition_by( "CREATE TABLE t (\"a\"\"b,c\" INT, `d``e,f` INT) \ PARTITIONED BY (\"a\"\"b,c\", `d``e,f`)", + true, ) .unwrap(); assert_eq!(keys, vec!["a\"b,c", "d`e,f"]); @@ -6774,20 +7004,25 @@ mod tests { #[test] fn test_extract_partition_by_backtick_quoted_identifier() { - let (_, keys) = - extract_partition_by("CREATE TABLE t (`order` INT) PARTITIONED BY (`order`)").unwrap(); + let (_, keys) = extract_partition_by( + "CREATE TABLE t (`order` INT) PARTITIONED BY (`order`)", + true, + ) + .unwrap(); assert_eq!(keys, vec!["order"]); } #[test] fn test_extract_partition_by_no_paren_after_by() { - let err = extract_partition_by("CREATE TABLE t (id INT) PARTITIONED BY dt").unwrap_err(); + let err = + extract_partition_by("CREATE TABLE t (id INT) PARTITIONED BY dt", true).unwrap_err(); assert!(err.to_string().contains("Expected '('")); } #[test] fn test_extract_partition_by_only_partitioned_no_by() { - let (rewritten, keys) = extract_partition_by("CREATE TABLE partitioned (id INT)").unwrap(); + let (rewritten, keys) = + extract_partition_by("CREATE TABLE partitioned (id INT)", true).unwrap(); assert_eq!(rewritten, "CREATE TABLE partitioned (id INT)"); assert!(keys.is_empty()); } @@ -6795,7 +7030,7 @@ mod tests { #[test] fn test_extract_partition_by_skips_block_comment() { let sql = "CREATE TABLE t (id INT) /* PARTITIONED BY (x) */ PARTITIONED BY (id)"; - let (rewritten, keys) = extract_partition_by(sql).unwrap(); + let (rewritten, keys) = extract_partition_by(sql, true).unwrap(); assert_eq!(keys, vec!["id"]); assert!(rewritten.contains("/* PARTITIONED BY (x) */")); } @@ -6966,6 +7201,66 @@ mod tests { (temp_dir, sql_context) } + async fn collect_partition_rows( + sql_context: &SQLContext, + table: &str, + partition_column: &str, + value_column: &str, + ) -> Vec<(String, i32)> { + let batches = sql_context + .sql(&format!( + "SELECT {partition_column}, {value_column} FROM {table} \ + ORDER BY {partition_column}" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + + let mut rows = Vec::new(); + for batch in batches { + let partitions = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + rows.push((partitions.value(row).to_string(), values.value(row))); + } + } + rows + } + + async fn setup_duplicate_partition_table() -> (tempfile::TempDir, SQLContext) { + let (tmp, sql_context) = setup_fs_sql_context().await; + sql_context + .sql( + "CREATE TABLE paimon.test_db.duplicate_partition_keys ( + dt VARCHAR, + value INT + ) PARTITIONED BY (dt)", + ) + .await + .unwrap(); + sql_context + .sql( + "INSERT INTO paimon.test_db.duplicate_partition_keys + VALUES ('a', 1), ('b', 2)", + ) + .await + .unwrap() + .collect() + .await + .unwrap(); + (tmp, sql_context) + } + #[tokio::test] async fn test_dynamic_read_batch_size_overrides_table_option() { let (_tmp, sql_context) = setup_fs_sql_context().await; @@ -7061,7 +7356,7 @@ mod tests { let (_tmp, sql_context) = setup_fs_sql_context().await; sql_context - .sql("CREATE TABLE paimon.test_db.t2 (pt VARCHAR, id INT) PARTITIONED BY (pt)") + .sql("CREATE TABLE paimon.test_db.t2 (PT VARCHAR, ID INT) PARTITIONED BY (PT)") .await .unwrap(); sql_context @@ -7073,7 +7368,7 @@ mod tests { .unwrap(); sql_context - .sql("TRUNCATE TABLE paimon.test_db.t2 PARTITION (pt = 'a')") + .sql("TRUNCATE TABLE paimon.test_db.t2 PARTITION (PT = 'a')") .await .unwrap(); @@ -7109,7 +7404,7 @@ mod tests { let (_tmp, sql_context) = setup_fs_sql_context().await; sql_context - .sql("CREATE TABLE paimon.test_db.t3 (pt VARCHAR, id INT) PARTITIONED BY (pt)") + .sql("CREATE TABLE paimon.test_db.t3 (PT VARCHAR, ID INT) PARTITIONED BY (PT)") .await .unwrap(); sql_context @@ -7121,7 +7416,7 @@ mod tests { .unwrap(); sql_context - .sql("ALTER TABLE paimon.test_db.t3 DROP PARTITION (pt = 'b')") + .sql("ALTER TABLE paimon.test_db.t3 DROP PARTITION (PT = 'b')") .await .unwrap(); @@ -7152,6 +7447,160 @@ mod tests { assert_eq!(rows, vec![("a".to_string(), 1), ("a".to_string(), 2)]); } + #[tokio::test] + async fn test_partition_removal_rejects_normalized_duplicate_keys() { + for sql in [ + "TRUNCATE TABLE paimon.test_db.duplicate_partition_keys + PARTITION (DT = 'a', dt = 'b')", + "ALTER TABLE paimon.test_db.duplicate_partition_keys + DROP PARTITION (DT = 'a', dt = 'b')", + ] { + let (_tmp, sql_context) = setup_duplicate_partition_table().await; + + let err = sql_context.sql(sql).await.unwrap_err(); + assert!( + err.to_string().contains("Duplicate partition column 'dt'"), + "Expected duplicate partition column error, got: {err}" + ); + assert_eq!( + collect_partition_rows( + &sql_context, + "paimon.test_db.duplicate_partition_keys", + "dt", + "value", + ) + .await, + [("a".to_string(), 1), ("b".to_string(), 2)] + ); + } + } + + #[tokio::test] + async fn test_insert_overwrite_rejects_duplicate_partition_keys() { + for sql in [ + "INSERT OVERWRITE paimon.test_db.duplicate_partition_keys + PARTITION (DT = 'a', dt = 'b') VALUES (9)", + "INSERT OVERWRITE paimon.test_db.duplicate_partition_keys + PARTITION (DT = 'a', dt) VALUES (9)", + ] { + let (_tmp, sql_context) = setup_duplicate_partition_table().await; + + let err = sql_context.sql(sql).await.unwrap_err(); + assert!( + err.to_string().contains("Duplicate partition column 'dt'"), + "Expected duplicate partition column error, got: {err}" + ); + assert_eq!( + collect_partition_rows( + &sql_context, + "paimon.test_db.duplicate_partition_keys", + "dt", + "value", + ) + .await, + [("a".to_string(), 1), ("b".to_string(), 2)] + ); + } + } + + #[tokio::test] + async fn test_partition_writes_preserve_quoted_identifiers() { + let (_tmp, sql_context) = setup_fs_sql_context().await; + let table = "paimon.test_db.quoted_partition_writes"; + + sql_context + .sql(&format!( + "CREATE TABLE {table} ( + \"PART\" VARCHAR, + \"VALUE\" INT + ) PARTITIONED BY (\"PART\")" + )) + .await + .unwrap(); + sql_context + .sql(&format!("INSERT INTO {table} VALUES ('x', 1), ('y', 10)")) + .await + .unwrap() + .collect() + .await + .unwrap(); + + sql_context + .sql(&format!( + "INSERT OVERWRITE {table} (\"VALUE\") \ + PARTITION (\"PART\" = 'x') VALUES (2)" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + sql_context + .sql(&format!( + "UPDATE {table} SET \"VALUE\" = 3 WHERE \"PART\" = 'x'" + )) + .await + .unwrap(); + sql_context + .sql(&format!( + "TRUNCATE TABLE {table} PARTITION (\"PART\" = 'y')" + )) + .await + .unwrap(); + + assert_eq!( + collect_partition_rows(&sql_context, table, "\"PART\"", "\"VALUE\"").await, + [("x".to_string(), 3)] + ); + } + + #[tokio::test] + async fn test_partition_writes_respect_disabled_normalization() { + let (_tmp, sql_context) = setup_fs_sql_context().await; + let table = "paimon.test_db.preserved_partition_writes"; + + sql_context + .sql("SET datafusion.sql_parser.enable_ident_normalization = false") + .await + .unwrap(); + sql_context + .sql(&format!( + "CREATE TABLE {table} ( + PART VARCHAR, + VALUE INT + ) PARTITIONED BY (PART)" + )) + .await + .unwrap(); + sql_context + .sql(&format!("INSERT INTO {table} VALUES ('x', 1)")) + .await + .unwrap() + .collect() + .await + .unwrap(); + + sql_context + .sql(&format!( + "INSERT OVERWRITE {table} (VALUE) \ + PARTITION (PART = 'x') VALUES (2)" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + sql_context + .sql(&format!("UPDATE {table} SET VALUE = 3 WHERE PART = 'x'")) + .await + .unwrap(); + + assert_eq!( + collect_partition_rows(&sql_context, table, "PART", "VALUE").await, + [("x".to_string(), 3)] + ); + } + #[tokio::test] async fn test_truncate_table_incomplete_partition_spec() { let (_tmp, sql_context) = setup_fs_sql_context().await; diff --git a/crates/integrations/datafusion/src/update.rs b/crates/integrations/datafusion/src/update.rs index 5362b09ad..e8693ed10 100644 --- a/crates/integrations/datafusion/src/update.rs +++ b/crates/integrations/datafusion/src/update.rs @@ -37,15 +37,16 @@ use crate::error::to_datafusion_error; use crate::merge_into::{ build_partition_set_from_where, extract_tracking_columns, is_delete_conflict, is_row_id_conflict, ok_result, project_update_columns, quote_identifier, - register_cow_target_table, retry_on_conflict, TempTableTracker, + register_cow_target_table, retry_on_conflict, validate_update_columns, TempTableTracker, }; -use crate::sql_context::SQLContext; +use crate::sql_context::{normalize_schema_identifier, SQLContext}; /// Execute an UPDATE statement on a Paimon table. pub(crate) async fn execute_update( ctx: &SQLContext, update: &Update, table: Table, + enable_ident_normalization: bool, ) -> DFResult { if let TableFactor::Table { alias: Some(a), .. } = &update.table.relation { return Err(DataFusionError::Plan(format!( @@ -58,9 +59,9 @@ pub(crate) async fn execute_update( let core_options = CoreOptions::new(schema.options()); if core_options.data_evolution_enabled() { - execute_data_evolution_update(ctx, update, table).await + execute_data_evolution_update(ctx, update, table, enable_ident_normalization).await } else if schema.trimmed_primary_keys().is_empty() { - execute_cow_update(ctx, update, &table).await + execute_cow_update(ctx, update, &table, enable_ident_normalization).await } else { Err(DataFusionError::Plan( "UPDATE on primary-key tables without data-evolution is not supported".to_string(), @@ -77,9 +78,10 @@ async fn execute_data_evolution_update( ctx: &SQLContext, update: &Update, table: Table, + enable_ident_normalization: bool, ) -> DFResult { retry_on_conflict("UPDATE", is_row_id_conflict, || { - execute_update_once(ctx, update, &table) + execute_update_once(ctx, update, &table, enable_ident_normalization) }) .await } @@ -89,29 +91,11 @@ async fn execute_update_once( ctx: &SQLContext, update: &Update, table: &Table, + enable_ident_normalization: bool, ) -> DFResult { // 1. Extract SET assignments - let mut columns = Vec::new(); - let mut exprs = Vec::new(); - for assignment in &update.assignments { - let col_name = match &assignment.target { - AssignmentTarget::ColumnName(name) => name - .0 - .last() - .and_then(|p| p.as_ident()) - .map(|id| id.value.clone()) - .ok_or_else(|| { - DataFusionError::Plan(format!("Invalid column name in SET: {name}")) - })?, - AssignmentTarget::Tuple(_) => { - return Err(DataFusionError::Plan( - "Tuple assignment in UPDATE SET is not supported".to_string(), - )); - } - }; - columns.push(col_name); - exprs.push(assignment.value.to_string()); - } + let (columns, exprs) = extract_set_assignments(update, enable_ident_normalization)?; + validate_update_columns(&columns, table.schema().fields())?; // 2. Create TableUpdate through the table write builder (validates preconditions) let wb = table.new_write_builder(); @@ -176,9 +160,10 @@ async fn execute_cow_update( ctx: &SQLContext, update: &Update, table: &Table, + enable_ident_normalization: bool, ) -> DFResult { retry_on_conflict("CoW UPDATE", is_delete_conflict, || { - execute_cow_update_once(ctx, update, table) + execute_cow_update_once(ctx, update, table, enable_ident_normalization) }) .await } @@ -188,8 +173,10 @@ async fn execute_cow_update_once( ctx: &SQLContext, update: &Update, table: &Table, + enable_ident_normalization: bool, ) -> DFResult { - let (columns, exprs) = extract_set_assignments(update)?; + let (columns, exprs) = extract_set_assignments(update, enable_ident_normalization)?; + validate_update_columns(&columns, table.schema().fields())?; let table_ref = update.table.to_string(); let where_str = update.selection.as_ref().map(|e| e.to_string()); @@ -305,7 +292,10 @@ async fn execute_cow_update_inner( } /// Extract column names and expressions from UPDATE SET assignments. -fn extract_set_assignments(update: &Update) -> DFResult<(Vec, Vec)> { +fn extract_set_assignments( + update: &Update, + enable_ident_normalization: bool, +) -> DFResult<(Vec, Vec)> { let mut columns = Vec::new(); let mut exprs = Vec::new(); for assignment in &update.assignments { @@ -314,7 +304,7 @@ fn extract_set_assignments(update: &Update) -> DFResult<(Vec, Vec Vec<(i32, String, i32)> { let mut rows = Vec::new(); for batch in batches { @@ -432,7 +486,9 @@ mod tests { let update = parse_update("UPDATE paimon.test_db.t_with_where SET name = 'ALICE' WHERE id = 1"); - execute_update(&sql_context, &update, table).await.unwrap(); + execute_update(&sql_context, &update, table, true) + .await + .unwrap(); let batches = sql_context .sql("SELECT id, name, value FROM paimon.test_db.t_with_where ORDER BY id") @@ -457,8 +513,10 @@ mod tests { async fn test_update_without_where() { let (_tmp, sql_context, table) = setup_data_evolution_table("t_without_where").await; - let update = parse_update("UPDATE paimon.test_db.t_without_where SET value = 99"); - execute_update(&sql_context, &update, table).await.unwrap(); + let update = parse_update("UPDATE paimon.test_db.t_without_where SET VALUE = 99"); + execute_update(&sql_context, &update, table, true) + .await + .unwrap(); let batches = sql_context .sql("SELECT id, name, value FROM paimon.test_db.t_without_where ORDER BY id") @@ -486,7 +544,9 @@ mod tests { let update = parse_update( "UPDATE paimon.test_db.t_multi_col SET name = 'updated', value = 0 WHERE id = 2", ); - execute_update(&sql_context, &update, table).await.unwrap(); + execute_update(&sql_context, &update, table, true) + .await + .unwrap(); let batches = sql_context .sql("SELECT id, name, value FROM paimon.test_db.t_multi_col ORDER BY id") @@ -513,7 +573,9 @@ mod tests { let update = parse_update("UPDATE paimon.test_db.t_no_match SET name = 'nobody' WHERE id = 99"); - let result = execute_update(&sql_context, &update, table).await.unwrap(); + let result = execute_update(&sql_context, &update, table, true) + .await + .unwrap(); let batches = result.collect().await.unwrap(); let count = batches[0] .column(0) @@ -538,7 +600,9 @@ mod tests { .unwrap(); let update = parse_update("UPDATE paimon.test_db.t_row_id SET name = 'ALICE' WHERE id = 1"); - execute_update(&sql_context, &update, table).await.unwrap(); + execute_update(&sql_context, &update, table, true) + .await + .unwrap(); // Get row IDs after update let after = sql_context @@ -583,7 +647,7 @@ mod tests { let sql_context = SQLContext::new(); let update = parse_update("UPDATE t SET id = 1"); - let result = execute_update(&sql_context, &update, table).await; + let result = execute_update(&sql_context, &update, table, true).await; assert!(result.is_err()); assert!(result .unwrap_err() @@ -653,10 +717,20 @@ mod tests { async fn test_cow_update_without_where() { let (_tmp, sql_context) = setup_append_only_table("t_cow_no_where").await; - sql_context - .sql("UPDATE paimon.test_db.t_cow_no_where SET value = 99") + let result = sql_context + .sql("UPDATE paimon.test_db.t_cow_no_where SET VALUE = 99") + .await + .unwrap() + .collect() .await .unwrap(); + let count = result[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0); + assert_eq!(count, 3); let rows = query_rows(&sql_context, "paimon.test_db.t_cow_no_where").await; assert_eq!( diff --git a/crates/integrations/datafusion/tests/merge_into_tests.rs b/crates/integrations/datafusion/tests/merge_into_tests.rs index 87fb50bed..4598b9a7b 100644 --- a/crates/integrations/datafusion/tests/merge_into_tests.rs +++ b/crates/integrations/datafusion/tests/merge_into_tests.rs @@ -25,7 +25,7 @@ mod common; use std::sync::Arc; -use arrow_array::{Int32Array, Int64Array}; +use arrow_array::{Array, Int32Array, Int64Array}; use common::string_value; use paimon::catalog::Identifier; use paimon::table::SnapshotManager; @@ -35,6 +35,17 @@ use tempfile::TempDir; // ======================= Helpers ======================= +const MERGE_TABLE_CONFIGS: [(&str, &str); 2] = [ + ("merge_ident_cow", ""), + ( + "merge_ident_data_evolution", + "WITH ( + 'data-evolution.enabled' = 'true', + 'row-tracking.enabled' = 'true' + )", + ), +]; + fn create_test_env() -> (TempDir, Arc) { let temp_dir = TempDir::new().expect("Failed to create temp dir"); let warehouse = format!("file://{}", temp_dir.path().display()); @@ -138,8 +149,190 @@ async fn register_source(sql_context: &SQLContext, sql: &str) { sql_context.sql(sql).await.unwrap().collect().await.unwrap(); } +async fn assert_merge_insert_identifier( + table_column: &str, + insert_column: &str, + query_column: &str, + enable_ident_normalization: bool, +) { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + + if !enable_ident_normalization { + sql_context + .sql("SET datafusion.sql_parser.enable_ident_normalization = false") + .await + .unwrap(); + } + + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + register_source( + &sql_context, + "CREATE TEMPORARY TABLE paimon.test_db.merge_ident_source \ + AS SELECT * FROM (VALUES (1, 42)) AS t(id, source_value)", + ) + .await; + + for (table_name, options) in MERGE_TABLE_CONFIGS { + sql_context + .sql(&format!( + "CREATE TABLE paimon.test_db.{table_name} ( + id INT NOT NULL, + {table_column} INT + ) {options}" + )) + .await + .unwrap(); + + sql_context + .sql(&format!( + "MERGE INTO paimon.test_db.{table_name} t + USING paimon.test_db.merge_ident_source s ON t.id = s.id + WHEN NOT MATCHED THEN + INSERT (id, {insert_column}) VALUES (s.id, s.source_value)" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + + let batches = sql_context + .sql(&format!( + "SELECT {query_column} FROM paimon.test_db.{table_name}" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 1 + ); + + let values = batches[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(!values.is_null(0), "{table_name} wrote NULL"); + assert_eq!(values.value(0), 42, "{table_name} wrote the wrong value"); + } +} + // ======================= Functional Tests ======================= +#[tokio::test] +async fn test_merge_insert_preserves_quoted_identifier() { + assert_merge_insert_identifier("\"MixedCase\"", "\"MixedCase\"", "\"MixedCase\"", true).await; +} + +#[tokio::test] +async fn test_merge_insert_normalizes_unquoted_identifier_by_default() { + assert_merge_insert_identifier("MIXEDCASE", "MIXEDCASE", "mixedcase", true).await; +} + +#[tokio::test] +async fn test_merge_insert_respects_disabled_identifier_normalization() { + assert_merge_insert_identifier("MixedCase", "MixedCase", "\"MixedCase\"", false).await; +} + +#[tokio::test] +async fn test_merge_insert_rejects_unknown_and_duplicate_columns() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + register_source( + &sql_context, + "CREATE TEMPORARY TABLE paimon.test_db.merge_ident_source \ + AS SELECT * FROM (VALUES (1, 42)) AS t(id, source_value)", + ) + .await; + + for (table_name, options) in MERGE_TABLE_CONFIGS { + sql_context + .sql(&format!( + "CREATE TABLE paimon.test_db.{table_name} ( + id INT NOT NULL, + value INT + ) {options}" + )) + .await + .unwrap(); + + for (columns, expected_error) in [ + ("id, missing", "Unknown column 'missing' in MERGE INSERT"), + ("id, ID", "Duplicate column 'id' in MERGE INSERT"), + ] { + let sql = format!( + "MERGE INTO paimon.test_db.{table_name} t + USING paimon.test_db.merge_ident_source s ON t.id = s.id + WHEN NOT MATCHED THEN + INSERT ({columns}) VALUES (s.id, s.source_value)" + ); + assert_merge_error(&sql_context, &sql, expected_error).await; + } + } +} + +#[tokio::test] +async fn test_merge_update_rejects_unknown_and_duplicate_target_columns() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + register_source( + &sql_context, + "CREATE TEMPORARY TABLE paimon.test_db.merge_update_source \ + AS SELECT * FROM (VALUES (1)) AS t(id)", + ) + .await; + + for (table_name, options) in MERGE_TABLE_CONFIGS { + sql_context + .sql(&format!( + "CREATE TABLE paimon.test_db.{table_name} ( + id INT, + value INT + ) {options}" + )) + .await + .unwrap(); + sql_context + .sql(&format!( + "INSERT INTO paimon.test_db.{table_name} (id, value) VALUES (1, 10)" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + + for (assignments, expected_error) in [ + ("\"VALUE\" = 3", "Unknown column 'VALUE' in UPDATE"), + ("value = 3, VALUE = 4", "Duplicate column 'value' in UPDATE"), + ] { + let sql = format!( + "MERGE INTO paimon.test_db.{table_name} t + USING paimon.test_db.merge_update_source s ON t.id = s.id + WHEN MATCHED THEN UPDATE SET {assignments}" + ); + assert_merge_error(&sql_context, &sql, expected_error).await; + } + } +} + #[tokio::test] async fn test_row_id_values_after_insert() { let (_tmp, catalog) = create_test_env(); diff --git a/crates/integrations/datafusion/tests/pk_tables.rs b/crates/integrations/datafusion/tests/pk_tables.rs index 39e259722..0f44c8197 100644 --- a/crates/integrations/datafusion/tests/pk_tables.rs +++ b/crates/integrations/datafusion/tests/pk_tables.rs @@ -961,9 +961,9 @@ async fn test_pk_insert_overwrite_with_partition_clause() { sql_context .sql( "CREATE TABLE paimon.test_db.t_ow_part ( - dt STRING, id INT NOT NULL, name STRING, - PRIMARY KEY (dt, id) - ) PARTITIONED BY (dt) + DT STRING, ID INT NOT NULL, NAME STRING, + PRIMARY KEY (DT, ID) + ) PARTITIONED BY (DT) WITH ('bucket' = '1')", ) .await @@ -986,7 +986,7 @@ async fn test_pk_insert_overwrite_with_partition_clause() { // The SELECT only provides non-partition columns (id, name). sql_context .sql( - "INSERT OVERWRITE paimon.test_db.t_ow_part PARTITION (dt = '2024-01-01') \ + "INSERT OVERWRITE paimon.test_db.t_ow_part PARTITION (DT = '2024-01-01') \ VALUES (10, 'new_alice'), (20, 'new_bob')", ) .await @@ -1236,9 +1236,9 @@ async fn test_pk_insert_overwrite_dynamic_partition_preserves_other_partitions() sql_context .sql( "CREATE TABLE paimon.test_db.t_dyn ( - dt STRING, id INT NOT NULL, name STRING, - PRIMARY KEY (dt, id) - ) PARTITIONED BY (dt) + DT STRING, ID INT NOT NULL, NAME STRING, + PRIMARY KEY (DT, ID) + ) PARTITIONED BY (DT) WITH ('bucket' = '1')", ) .await @@ -1259,7 +1259,7 @@ async fn test_pk_insert_overwrite_dynamic_partition_preserves_other_partitions() // Should only overwrite partitions present in the source data. sql_context .sql( - "INSERT OVERWRITE paimon.test_db.t_dyn PARTITION (dt) \ + "INSERT OVERWRITE paimon.test_db.t_dyn PARTITION (DT) \ VALUES ('2024-01-01', 10, 'new_alice')", ) .await @@ -1354,9 +1354,9 @@ async fn test_pk_insert_overwrite_with_after_columns_reorder() { sql_context .sql( "CREATE TABLE paimon.test_db.t_reorder ( - dt STRING, id INT NOT NULL, name STRING, - PRIMARY KEY (dt, id) - ) PARTITIONED BY (dt) + DT STRING, ID INT NOT NULL, NAME STRING, + PRIMARY KEY (DT, ID) + ) PARTITIONED BY (DT) WITH ('bucket' = '1')", ) .await @@ -1365,7 +1365,7 @@ async fn test_pk_insert_overwrite_with_after_columns_reorder() { // Insert with columns in reversed order: (name, id) instead of schema order (id, name) sql_context .sql( - "INSERT OVERWRITE paimon.test_db.t_reorder (name, id) PARTITION (dt = '2024-01-01') \ + "INSERT OVERWRITE paimon.test_db.t_reorder (NAME, ID) PARTITION (dt = '2024-01-01') \ VALUES ('alice', 1), ('bob', 2)", ) .await diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs b/crates/integrations/datafusion/tests/sql_context_tests.rs index 308b806c0..b1f37d855 100644 --- a/crates/integrations/datafusion/tests/sql_context_tests.rs +++ b/crates/integrations/datafusion/tests/sql_context_tests.rs @@ -1317,6 +1317,46 @@ async fn test_alter_table_update_column_type_and_nullability() { assert!(table.schema().fields()[1].data_type().is_nullable()); } +#[tokio::test] +async fn test_alter_column_preserves_quoted_identifier() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog.clone()).await; + + sql_context + .sql("CREATE SCHEMA paimon.mydb") + .await + .expect("CREATE SCHEMA should succeed"); + sql_context + .sql( + "CREATE TABLE paimon.mydb.quoted_column ( + id INT, + \"MixedCase\" INT + )", + ) + .await + .expect("CREATE TABLE should preserve the quoted column name"); + + sql_context + .sql( + "ALTER TABLE paimon.mydb.quoted_column + ALTER COLUMN \"MixedCase\" TYPE BIGINT", + ) + .await + .expect("ALTER COLUMN should resolve the quoted column name"); + + let table = catalog + .get_table(&Identifier::new("mydb", "quoted_column")) + .await + .unwrap(); + let field = table + .schema() + .fields() + .iter() + .find(|field| field.name() == "MixedCase") + .expect("quoted column should retain its exact name"); + assert!(matches!(field.data_type(), DataType::BigInt(_))); +} + #[tokio::test] async fn test_alter_table_rename() { let (_tmp, catalog) = create_test_env(); diff --git a/docs/src/sql.md b/docs/src/sql.md index 42c78cbb3..d687da090 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -578,6 +578,27 @@ CREATE TABLE IF NOT EXISTS paimon.my_db.users ( ); ``` +Top-level column identifiers in persistent Paimon DDL follow +`datafusion.sql_parser.enable_ident_normalization`. With the default value +`true`, unquoted identifiers are stored in lowercase, while quoted identifiers +preserve their spelling. This applies consistently to column definitions, +primary keys, partition keys, and `ALTER TABLE` column operations. + +```sql +CREATE TABLE paimon.my_db.unquoted_example (ID INT); -- stores `id` +CREATE TABLE paimon.my_db.quoted_example ("ID" INT); -- stores `ID` +``` + +Set the option to `false` to preserve unquoted spelling: + +```sql +SET datafusion.sql_parser.enable_ident_normalization = false; +CREATE TABLE paimon.my_db.preserved_example (ID INT); -- stores `ID` +``` + +Changing the option does not rename existing schema fields. Quote an existing +mixed-case or uppercase field when normalization is enabled. + Unsupported syntax (will return an error): - `CREATE EXTERNAL TABLE` - `LOCATION`