This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new de79604e fix(datafusion): normalize Paimon column identifiers
consistently (#634)
de79604e is described below
commit de79604e3b00098a550042250fb058ea1419e0dc
Author: QuakeWang <[email protected]>
AuthorDate: Mon Aug 3 11:43:28 2026 +0800
fix(datafusion): normalize Paimon column identifiers consistently (#634)
---
crates/integrations/datafusion/src/merge_into.rs | 266 ++++++---
crates/integrations/datafusion/src/sql_context.rs | 631 ++++++++++++++++++---
crates/integrations/datafusion/src/update.rs | 152 +++--
.../datafusion/tests/merge_into_tests.rs | 195 ++++++-
crates/integrations/datafusion/tests/pk_tables.rs | 24 +-
.../datafusion/tests/sql_context_tests.rs | 40 ++
docs/src/sql.md | 21 +
7 files changed, 1115 insertions(+), 214 deletions(-)
diff --git a/crates/integrations/datafusion/src/merge_into.rs
b/crates/integrations/datafusion/src/merge_into.rs
index 97c36079..1080eeae 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<DataFrame> {
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<DataFrame> {
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<CowMergeClauses> {
+fn extract_cow_merge_clauses(
+ merge: &Merge,
+ enable_ident_normalization: bool,
+) -> DFResult<CowMergeClauses> {
let mut matched: Vec<CowMatchedClause> = Vec::new();
let mut inserts: Vec<MergeInsertClause> = Vec::new();
@@ -216,7 +221,12 @@ fn extract_cow_merge_clauses(merge: &Merge) ->
DFResult<CowMergeClauses> {
.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<CowMergeClauses> {
MergeClauseKind::NotMatched | MergeClauseKind::NotMatchedByTarget
=> {
match &clause.action {
MergeAction::Insert(insert_expr) => {
- let columns: Vec<String> =
- 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<CowMergeClauses> {
}
/// 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<DataFrame> {
+async fn execute_cow_merge(
+ ctx: &SQLContext,
+ merge: &Merge,
+ table: Table,
+ enable_ident_normalization: bool,
+) -> DFResult<DataFrame> {
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<DataFrame> {
- 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<String> = Vec::new();
@@ -625,9 +633,14 @@ async fn execute_merge_into_once(
ctx: &SQLContext,
merge: &Merge,
table: &Table,
+ enable_ident_normalization: bool,
) -> DFResult<DataFrame> {
// 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<String, &str> = 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<String>,
}
+fn extract_merge_insert_clause(
+ insert_expr: &MergeInsertExpr,
+ predicate: Option<&SqlExpr>,
+ enable_ident_normalization: bool,
+) -> DFResult<MergeInsertClause> {
+ 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::<DFResult<Vec<_>>>()?;
+
+ 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<String>,
@@ -1035,7 +1128,10 @@ struct ParsedMergeClauses {
}
/// Extract UPDATE and INSERT clauses from the MERGE AST.
-fn extract_merge_clauses(merge: &Merge) -> DFResult<ParsedMergeClauses> {
+fn extract_merge_clauses(
+ merge: &Merge,
+ enable_ident_normalization: bool,
+) -> DFResult<ParsedMergeClauses> {
let mut update: Option<MergeUpdateClause> = None;
let mut delete = false;
let mut inserts: Vec<MergeInsertClause> = Vec::new();
@@ -1063,7 +1159,12 @@ fn extract_merge_clauses(merge: &Merge) ->
DFResult<ParsedMergeClauses> {
.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<ParsedMergeClauses> {
MergeClauseKind::NotMatched | MergeClauseKind::NotMatchedByTarget
=> {
match &clause.action {
MergeAction::Insert(insert_expr) => {
- let columns: Vec<String> =
- 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 1803892b..cd54a08e 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<DataFrame> {
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<dyn Catalog>,
ct: &CreateTable,
partition_keys: Vec<String>,
+ enable_ident_normalization: bool,
) -> DFResult<DataFrame> {
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<String> = 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<String> = 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<DataFrame> {
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<DataFrame> {
+ async fn handle_merge_into(
+ &self,
+ merge: &Merge,
+ enable_ident_normalization: bool,
+ ) -> DFResult<DataFrame> {
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<DataFrame> {
+ async fn handle_update(
+ &self,
+ update: &Update,
+ enable_ident_normalization: bool,
+ ) -> DFResult<DataFrame> {
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<DataFrame> {
@@ -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<DataFrame> {
+ async fn handle_insert_overwrite_partition(
+ &self,
+ insert: &Insert,
+ enable_ident_normalization: bool,
+ ) -> DFResult<DataFrame> {
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::<DFResult<_>>()?,
)
} 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<DataFrame> {
+ async fn handle_truncate_table(
+ &self,
+ truncate: &Truncate,
+ enable_ident_normalization: bool,
+ ) -> DFResult<DataFrame> {
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<DataFrame> {
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<String> {
+fn parse_partition_column(token: &str, enable_ident_normalization: bool) ->
DFResult<String> {
let trimmed = token.trim();
if trimmed.is_empty() {
return Err(DataFusionError::Plan(
@@ -2434,7 +2490,10 @@ fn parse_partition_column(token: &str) ->
DFResult<String> {
}
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<String> {
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<Vec<&str>> {
/// 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<String>)> {
+fn extract_partition_by(
+ sql: &str,
+ enable_ident_normalization: bool,
+) -> DFResult<(String, Vec<String>)> {
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<String>)> {
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<String>)> {
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<SchemaChange> {
+fn column_def_to_add_column(
+ col: &ColumnDef,
+ enable_ident_normalization: bool,
+) -> DFResult<SchemaChange> {
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<SchemaChange> {
}
fn alter_column_to_schema_change(
- column_name: &str,
+ column_name: &Ident,
operation: &AlterColumnOperation,
+ enable_ident_normalization: bool,
) -> DFResult<SchemaChange> {
+ 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<String>
{
})
}
-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<String> {
+fn object_name_to_single_identifier(
+ name: &ObjectName,
+ enable_ident_normalization: bool,
+) -> DFResult<String> {
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<Vec<HashMap<String, Option<Datum>>>> {
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<HashMap<String, Option<Datum>>> {
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::<StringViewArray>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .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 5362b09a..e8693ed1 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<DataFrame> {
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<DataFrame> {
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<DataFrame> {
// 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<DataFrame> {
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<DataFrame> {
- 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<String>,
Vec<String>)> {
+fn extract_set_assignments(
+ update: &Update,
+ enable_ident_normalization: bool,
+) -> DFResult<(Vec<String>, Vec<String>)> {
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<String>, Vec<String
.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}"))
})?,
@@ -401,6 +391,70 @@ mod tests {
}
}
+ #[test]
+ fn test_extract_set_assignments_normalizes_identifiers() {
+ for (sql, enabled, expected) in [
+ ("UPDATE t SET VALUE = 1", true, "value"),
+ ("UPDATE t SET \"VALUE\" = 1", true, "VALUE"),
+ ("UPDATE t SET VALUE = 1", false, "VALUE"),
+ ] {
+ let update = parse_update(sql);
+ let (columns, _) = extract_set_assignments(&update,
enabled).unwrap();
+ assert_eq!(columns, [expected]);
+ }
+ }
+
+ #[tokio::test]
+ async fn test_update_rejects_unknown_and_duplicate_target_columns() {
+ let (_tmp, sql_context, _catalog) = setup_sql_context().await;
+
+ for (table_name, options) in [
+ ("t_update_columns_cow", ""),
+ (
+ "t_update_columns_data_evolution",
+ "WITH (
+ 'data-evolution.enabled' = 'true',
+ 'row-tracking.enabled' = 'true'
+ )",
+ ),
+ ] {
+ 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 err = sql_context
+ .sql(&format!(
+ "UPDATE paimon.test_db.{table_name} SET {assignments}"
+ ))
+ .await
+ .unwrap_err();
+ assert!(
+ err.to_string().contains(expected_error),
+ "Error '{err}' does not contain '{expected_error}'"
+ );
+ }
+ }
+ }
+
fn collect_rows(batches: &[datafusion::arrow::array::RecordBatch]) ->
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::<UInt64Array>()
+ .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 87fb50be..4598b9a7 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<FileSystemCatalog>) {
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::<usize>(),
+ 1
+ );
+
+ let values = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .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 39e25972..0f44c819 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 308b806c..b1f37d85 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 b0da06bf..208e0322 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`
