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 18fc5797 fix(datafusion): map column-less MERGE INSERT VALUES by
position (#984)
18fc5797 is described below
commit 18fc579705bbed61a2f82b589c5e51f5c9397b7c
Author: jackylee <[email protected]>
AuthorDate: Sat Oct 3 18:07:21 2026 +0800
fix(datafusion): map column-less MERGE INSERT VALUES by position (#984)
---
crates/integrations/datafusion/src/merge_into.rs | 187 ++++++++++++++++++++---
1 file changed, 165 insertions(+), 22 deletions(-)
diff --git a/crates/integrations/datafusion/src/merge_into.rs
b/crates/integrations/datafusion/src/merge_into.rs
index 0bb9a3b2..15e8c82e 100644
--- a/crates/integrations/datafusion/src/merge_into.rs
+++ b/crates/integrations/datafusion/src/merge_into.rs
@@ -907,7 +907,7 @@ async fn build_insert_batches_inner(
format!(" WHERE {}", conditions.join(" AND "))
};
- let select_clause = insert_select_clause(ins, table_fields);
+ let select_clause = insert_select_clause(ins, table_fields)?;
let sql = format!("SELECT {select_clause} FROM {tmp_name} AS
{s_alias}{where_clause}");
let batches = ctx.ctx().sql(&sql).await?.collect().await?;
@@ -995,32 +995,56 @@ fn strip_non_source_columns(
///
/// When the INSERT specifies explicit columns (`INSERT (col2, col1) VALUES
(expr2, expr1)`),
/// the output must be reordered to match the table schema so that
`write_arrow_batch`
-/// (which reads columns by positional index) maps them correctly.
-fn insert_select_clause(ins: &MergeInsertClause, table_fields: &[DataField])
-> String {
+/// (which reads columns by positional index) maps them correctly. When the
INSERT omits
+/// the column list (`INSERT VALUES (expr1, expr2, ...)`), the values map to
the table's
+/// columns by position, following standard SQL, where the MERGE INSERT column
list is
+/// optional (unlike Spark, whose grammar requires an explicit list or `INSERT
*`).
+fn insert_select_clause(ins: &MergeInsertClause, table_fields: &[DataField])
-> DFResult<String> {
if ins.columns.is_empty() && ins.value_exprs.is_empty() {
- "*".to_string()
- } else {
- // Build column_name -> expression mapping from the INSERT clause
- let col_expr_map: HashMap<&str, &str> = ins
- .columns
- .iter()
- .zip(ins.value_exprs.iter())
- .map(|(col, expr)| (col.as_str(), expr.as_str()))
- .collect();
+ return Ok("*".to_string());
+ }
- // Emit SELECT in table schema order
- table_fields
+ if ins.columns.is_empty() {
+ // No explicit column list: the values fill the table's columns by
+ // position. Without this branch the empty column->expr map below would
+ // emit `NULL` for every column and silently drop the source values,
+ // inserting all-NULL rows (or failing on a non-null column).
+ if ins.value_exprs.len() != table_fields.len() {
+ return Err(DataFusionError::Plan(format!(
+ "MERGE INSERT has {} value(s) but the table has {} column(s); \
+ list the target columns explicitly or provide one value per
column",
+ ins.value_exprs.len(),
+ table_fields.len()
+ )));
+ }
+ return Ok(table_fields
.iter()
- .map(|field| {
- 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())),
- }
- })
+ .zip(ins.value_exprs.iter())
+ .map(|(field, expr)| format!("{expr} AS {}",
quote_identifier(field.name())))
.collect::<Vec<_>>()
- .join(", ")
+ .join(", "));
}
+
+ // Build column_name -> expression mapping from the INSERT clause
+ let col_expr_map: HashMap<&str, &str> = ins
+ .columns
+ .iter()
+ .zip(ins.value_exprs.iter())
+ .map(|(col, expr)| (col.as_str(), expr.as_str()))
+ .collect();
+
+ // Emit SELECT in table schema order
+ Ok(table_fields
+ .iter()
+ .map(|field| {
+ 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())),
+ }
+ })
+ .collect::<Vec<_>>()
+ .join(", "))
}
/// Parsed WHEN NOT MATCHED THEN INSERT clause.
@@ -1078,6 +1102,24 @@ fn validate_merge_insert_columns(
) -> DFResult<()> {
for insert in inserts {
validate_target_columns(&insert.columns, table_fields, "MERGE
INSERT")?;
+ // A column-less positional INSERT maps its VALUES to the table
columns by
+ // position, so the value count must match the column count. Validate
it
+ // here, upfront: this runs in both CoW and data-evolution modes
before any
+ // batch is built, whereas `insert_select_clause` only runs when
unmatched
+ // rows exist — an all-matched MERGE would otherwise skip the arity
check
+ // and commit. `INSERT *`/`INSERT ROW` (no columns and no values) is
left
+ // untouched.
+ if insert.columns.is_empty()
+ && !insert.value_exprs.is_empty()
+ && insert.value_exprs.len() != table_fields.len()
+ {
+ return Err(DataFusionError::Plan(format!(
+ "MERGE INSERT has {} value(s) but the table has {} column(s); \
+ list the target columns explicitly or provide one value per
column",
+ insert.value_exprs.len(),
+ table_fields.len()
+ )));
+ }
}
Ok(())
@@ -2191,6 +2233,107 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_cow_merge_insert_not_matched_without_columns() {
+ let (_tmp, sql_context, table) =
setup_append_only_table("t_cow_ins_nocols").await;
+
+ sql_context
+ .sql("CREATE TABLE paimon.test_db.source (id INT, name VARCHAR,
value INT)")
+ .await
+ .unwrap();
+ sql_context
+ .sql("INSERT INTO paimon.test_db.source VALUES (4, 'dave', 40),
(5, 'eve', 50)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ // No column list: the VALUES map to the table's columns by position.
+ // Before the fix this dropped the source values and inserted all-NULL
rows.
+ let merge = parse_merge(
+ "MERGE INTO paimon.test_db.t_cow_ins_nocols t USING
paimon.test_db.source s \
+ ON t.id = s.id \
+ WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.name, s.value)",
+ );
+ execute_merge_into(&sql_context, &merge, table, true)
+ .await
+ .unwrap();
+
+ let batches = sql_context
+ .sql("SELECT id, name, value FROM paimon.test_db.t_cow_ins_nocols
ORDER BY id")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let rows = collect_rows(&batches);
+ assert_eq!(
+ rows,
+ vec![
+ (1, "alice".to_string(), 10),
+ (2, "bob".to_string(), 20),
+ (3, "charlie".to_string(), 30),
+ (4, "dave".to_string(), 40),
+ (5, "eve".to_string(), 50),
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_cow_merge_insert_arity_validated_even_when_all_matched() {
+ let (_tmp, sql_context, table) =
setup_append_only_table("t_cow_ins_arity").await;
+
+ sql_context
+ .sql("CREATE TABLE paimon.test_db.source (id INT, name VARCHAR,
value INT)")
+ .await
+ .unwrap();
+ // id = 1 matches the target, so the NOT MATCHED INSERT branch is never
+ // built — the arity check must still run upfront.
+ sql_context
+ .sql("INSERT INTO paimon.test_db.source VALUES (1, 'ALICE', 99)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ // The column-less INSERT lists 2 values for a 3-column table. Even
with no
+ // row to insert, the invalid arity must be rejected before execution
so the
+ // matched UPDATE is not applied.
+ let merge = parse_merge(
+ "MERGE INTO paimon.test_db.t_cow_ins_arity t USING
paimon.test_db.source s \
+ ON t.id = s.id \
+ WHEN MATCHED THEN UPDATE SET value = s.value \
+ WHEN NOT MATCHED THEN INSERT VALUES (s.id, s.name)",
+ );
+ let err = execute_merge_into(&sql_context, &merge, table, true)
+ .await
+ .unwrap_err();
+ assert!(
+ err.to_string().contains("value(s) but the table has"),
+ "got {err}"
+ );
+
+ // The target is unchanged: the matched UPDATE did not apply.
+ let batches = sql_context
+ .sql("SELECT id, name, value FROM paimon.test_db.t_cow_ins_arity
ORDER BY id")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ collect_rows(&batches),
+ vec![
+ (1, "alice".to_string(), 10),
+ (2, "bob".to_string(), 20),
+ (3, "charlie".to_string(), 30),
+ ]
+ );
+ }
+
#[tokio::test]
async fn test_cow_merge_update_and_insert() {
let (_tmp, sql_context, table) =
setup_append_only_table("t_cow_upsert").await;
