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 5b2eceb8 feat(table): add prepared MERGE operations to core updates
(#1068)
5b2eceb8 is described below
commit 5b2eceb820d2a4598b3fe71e495350a945b0a58d
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 09:31:56 2026 +0800
feat(table): add prepared MERGE operations to core updates (#1068)
---
bindings/python/README.md | 15 +
.../python/python/pypaimon_rust/datafusion.pyi | 23 +
bindings/python/src/lib.rs | 1 +
bindings/python/src/merge_into.rs | 149 ++++++
bindings/python/src/write.rs | 76 +++
bindings/python/tests/test_update.py | 37 ++
crates/integrations/datafusion/src/lib.rs | 2 +
.../integrations/datafusion/src/merge_condition.rs | 267 ++++++++++
crates/paimon/src/table/mod.rs | 6 +
crates/paimon/src/table/table_merge_into.rs | 588 +++++++++++++++++++++
crates/paimon/src/table/table_merge_into_tests.rs | 477 +++++++++++++++++
crates/paimon/src/table/table_update.rs | 20 +
12 files changed, 1661 insertions(+)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index f436eeef..8cdca711 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -108,6 +108,21 @@ print(f"\nRead: {batches_tt[0].num_rows} rows")
print(batches_tt[0])
```
+### Prepared MERGE operations
+
+Batch and stream update objects expose `merge_into` to prepare data-evolution
+updates, deletion vectors and inserts without publishing a snapshot. Rust core
+owns matching, ordered action selection and file staging. SQL conditions are
+compiled by the Rust DataFusion adapter. PyPaimon's existing `WhenMatched` and
+`WhenNotMatched` APIs normalize into these core clauses; no Python join runs on
+the native path.
+
+Source NULL keys do not match target NULL keys. Multiple source matches are
+rejected before action conditions, except for a sole unconditional DELETE.
+Self-merge uses the target's pinned row IDs. A later action failure retains
files already prepared by earlier actions, even
+when the operation cannot return their messages. Returned messages may only be
+explicitly aborted by a caller that knows publication has been abandoned.
+
### Custom Blob URI sources
Before the first Arrow write, call
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index ed8a1b4a..8e91f096 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -262,6 +262,17 @@ class BatchTableWrite:
...
class BatchTableUpdate:
+ def merge_into(
+ self, source: Optional[pyarrow.Table], *, on: Sequence[Tuple[str,
str]],
+ when_matched: Sequence[Dict[str, Any]], when_not_matched:
Sequence[Dict[str, Any]]
+ ) -> List[CommitMessage]:
+ """Prepare MERGE against one snapshot. None selects self-merge on
_ROW_ID.
+
+ Clauses carry assignments as (column, source/target/literal, value)
+ tuples, a delete flag, and optional SQL condition metadata.
+ Commit returned messages through the matching builder's committer.
+ """
+ ...
def update_by_predicate(
self, predicate: Optional[Dict[str, Any]], assignments: Dict[str, Any],
read_columns: Optional[Sequence[str]] = None
@@ -289,6 +300,18 @@ class TableUpdateByRowId:
def _abort(self) -> None: ...
class StreamTableUpdate:
+ def merge_into(
+ self, source: Optional[pyarrow.Table], *, on: Sequence[Tuple[str,
str]],
+ when_matched: Sequence[Dict[str, Any]], when_not_matched:
Sequence[Dict[str, Any]],
+ commit_identifier: int
+ ) -> List[CommitMessage]:
+ """Prepare MERGE against one snapshot. None selects self-merge on
_ROW_ID.
+
+ Clauses carry assignments as (column, source/target/literal, value)
+ tuples, a delete flag, and optional SQL condition metadata.
+ Commit returned messages through the matching builder's committer.
+ """
+ ...
def update_by_predicate(
self, predicate: Optional[Dict[str, Any]], assignments: Dict[str, Any],
commit_identifier: int, read_columns: Optional[Sequence[str]] = None
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 30446a3f..b4e270f7 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -21,6 +21,7 @@ mod blob;
mod blob_uri_reader;
mod context;
mod error;
+mod merge_into;
mod oss_cpp;
mod predicate;
mod read;
diff --git a/bindings/python/src/merge_into.rs
b/bindings/python/src/merge_into.rs
new file mode 100644
index 00000000..17e8683b
--- /dev/null
+++ b/bindings/python/src/merge_into.rs
@@ -0,0 +1,149 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Language transport and DataFusion expression compilation for core MERGE.
+
+use std::sync::{Arc, Mutex};
+
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use paimon::table::{MergeAssignment, MergeCondition, WhenMatched,
WhenNotMatched};
+use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::*;
+use pyo3::types::PyDict;
+
+use crate::error::to_py_err;
+
+type CallbackError = Arc<Mutex<Option<PyErr>>>;
+
+fn condition(
+ py: Python<'_>,
+ clause: &Bound<'_, PyDict>,
+ schema: SchemaRef,
+) -> PyResult<Option<MergeCondition>> {
+ let Some(value) = clause
+ .get_item("condition")?
+ .filter(|value| !value.is_none())
+ else {
+ return Ok(None);
+ };
+ let value = value.cast::<PyDict>()?;
+ let sql: String = value
+ .get_item("sql")?
+ .ok_or_else(|| PyValueError::new_err("MERGE condition needs SQL"))?
+ .extract()?;
+ py.detach(||
runtime().block_on(paimon_datafusion::compile_merge_condition(sql, schema)))
+ .map(Some)
+ .map_err(to_py_err)
+}
+
+fn assignments(
+ py: Python<'_>,
+ clause: &Bound<'_, PyDict>,
+ fields: &[paimon::spec::DataField],
+ errors: CallbackError,
+) -> PyResult<Vec<(String, MergeAssignment)>> {
+ let values = clause
+ .get_item("assignments")?
+ .ok_or_else(|| PyValueError::new_err("MERGE clause needs
assignments"))?;
+ values
+ .try_iter()?
+ .map(|item| {
+ let (name, kind, value): (String, String, Py<PyAny>) =
item?.extract()?;
+ let assignment = match kind.as_str() {
+ "source" =>
MergeAssignment::SourceColumn(value.bind(py).extract()?),
+ "target" =>
MergeAssignment::TargetColumn(value.bind(py).extract()?),
+ "literal" => {
+ let values = PyDict::new(py);
+ values.set_item(&name, value)?;
+ let assignment =
+ crate::update_assignment::from_python(py, &values,
fields, errors.clone())?
+ .remove(0)
+ .1;
+ MergeAssignment::Value(assignment)
+ }
+ _ => return Err(PyValueError::new_err("Unknown MERGE
assignment kind")),
+ };
+ Ok((name, assignment))
+ })
+ .collect()
+}
+
+pub(crate) fn from_python(
+ py: Python<'_>,
+ table: &paimon::table::Table,
+ source_schema: SchemaRef,
+ matched: &Bound<'_, PyAny>,
+ not_matched: &Bound<'_, PyAny>,
+ errors: CallbackError,
+) -> PyResult<(Vec<WhenMatched>, Vec<WhenNotMatched>)> {
+ let target_schema =
+
paimon::arrow::build_target_arrow_schema(table.schema().fields()).map_err(to_py_err)?;
+ let mut fields = vec![Field::new("t._ROW_ID", DataType::Int64, false)];
+ fields.extend(target_schema.fields().iter().map(|field| {
+ field
+ .as_ref()
+ .clone()
+ .with_name(format!("t.{}", field.name()))
+ }));
+ fields.extend(source_schema.fields().iter().map(|field| {
+ field
+ .as_ref()
+ .clone()
+ .with_name(format!("s.{}", field.name()))
+ }));
+ let schema = Arc::new(Schema::new(fields));
+ let matched = matched
+ .try_iter()?
+ .map(|item| {
+ let item = item?;
+ let clause = item.cast::<PyDict>()?;
+ Ok(WhenMatched {
+ condition: condition(py, clause, schema.clone())?,
+ delete: clause
+ .get_item("delete")?
+ .ok_or_else(|| PyValueError::new_err("MERGE matched clause
needs action"))?
+ .extract()?,
+ assignments: assignments(py, clause, table.schema().fields(),
errors.clone())?,
+ })
+ })
+ .collect::<PyResult<Vec<_>>>()?;
+ let schema = Arc::new(Schema::new(
+ source_schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .as_ref()
+ .clone()
+ .with_name(format!("s.{}", field.name()))
+ })
+ .collect::<Vec<_>>(),
+ ));
+ let not_matched = not_matched
+ .try_iter()?
+ .map(|item| {
+ let item = item?;
+ let clause = item.cast::<PyDict>()?;
+ Ok(WhenNotMatched {
+ condition: condition(py, clause, schema.clone())?,
+ assignments: assignments(py, clause, table.schema().fields(),
errors.clone())?,
+ })
+ })
+ .collect::<PyResult<Vec<_>>>()?;
+ Ok((matched, not_matched))
+}
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index d1de7f8a..7555cebc 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -583,6 +583,53 @@ impl UpdateContext {
})
}
+ fn merge_into(
+ &self,
+ py: Python<'_>,
+ source: Option<&Bound<'_, PyAny>>,
+ on: Vec<(String, String)>,
+ when_matched: &Bound<'_, PyAny>,
+ when_not_matched: &Bound<'_, PyAny>,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ let source = match source {
+ Some(source) =>
paimon::table::MergeSource::Batches(arrow_table_batches(source)?),
+ None => paimon::table::MergeSource::SelfTable,
+ };
+ let source_schema = match &source {
+ paimon::table::MergeSource::Batches(batches) =>
batches[0].schema(),
+ paimon::table::MergeSource::SelfTable => {
+ let schema =
paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
+ .map_err(to_py_err)?;
+ let mut fields = schema.fields().to_vec();
+ fields.push(Arc::new(arrow::datatypes::Field::new(
+ "_ROW_ID",
+ arrow::datatypes::DataType::Int64,
+ false,
+ )));
+ Arc::new(ArrowSchema::new(fields))
+ }
+ };
+ let errors = Arc::new(Mutex::new(None));
+ let (matched, not_matched) = crate::merge_into::from_python(
+ py,
+ &self.table,
+ source_schema,
+ when_matched,
+ when_not_matched,
+ errors.clone(),
+ )?;
+ let result = py
+ .detach(|| runtime().block_on(self.inner.merge_into(source, on,
matched, not_matched)));
+ if let Some(error) = errors.lock().unwrap().take() {
+ return Err(error);
+ }
+ Ok(wrap_messages(
+ result.map_err(to_py_err)?,
+ &self.table_location,
+ &self.commit_user,
+ ))
+ }
+
fn update_by_predicate(
&self,
py: Python<'_>,
@@ -862,6 +909,22 @@ pub struct PyStreamTableUpdate {
#[pymethods]
impl PyStreamTableUpdate {
+ #[pyo3(signature = (source, *, on, when_matched, when_not_matched,
commit_identifier))]
+ fn merge_into(
+ &self,
+ py: Python<'_>,
+ source: Option<&Bound<'_, PyAny>>,
+ on: Vec<(String, String)>,
+ when_matched: &Bound<'_, PyAny>,
+ when_not_matched: &Bound<'_, PyAny>,
+ commit_identifier: i64,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ // The stream committer applies the identifier to the returned
messages.
+ let _ = commit_identifier;
+ self.context
+ .merge_into(py, source, on, when_matched, when_not_matched)
+ }
+
#[pyo3(signature = (predicate, assignments, commit_identifier,
read_columns=None))]
fn update_by_predicate(
&self,
@@ -933,6 +996,19 @@ impl PyStreamTableUpdate {
#[pymethods]
impl PyBatchTableUpdate {
+ #[pyo3(signature = (source, *, on, when_matched, when_not_matched))]
+ fn merge_into(
+ &self,
+ py: Python<'_>,
+ source: Option<&Bound<'_, PyAny>>,
+ on: Vec<(String, String)>,
+ when_matched: &Bound<'_, PyAny>,
+ when_not_matched: &Bound<'_, PyAny>,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ self.context
+ .merge_into(py, source, on, when_matched, when_not_matched)
+ }
+
#[pyo3(signature = (predicate, assignments, read_columns=None))]
fn update_by_predicate(
&self,
diff --git a/bindings/python/tests/test_update.py
b/bindings/python/tests/test_update.py
index 3702f645..7bed73e2 100644
--- a/bindings/python/tests/test_update.py
+++ b/bindings/python/tests/test_update.py
@@ -429,3 +429,40 @@ def
test_update_can_transfer_and_be_collected_on_another_thread(tmp_path, kind,
finally:
if enabled:
gc.enable()
+
+
[email protected]('stream', [False, True])
+def test_prepared_merge_conditions_chunks_and_lazy_literals(tmp_path, stream):
+ context = SQLContext()
+ context.register_catalog('paimon', {'warehouse': str(tmp_path)})
+ context.sql('CREATE SCHEMA paimon.prepared_merge')
+ context.sql("""CREATE TABLE paimon.prepared_merge.t (id INT, value INT)
WITH (
+ 'row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true',
+ 'deletion-vectors.enabled' = 'true')""")
+ # Separate commits force the clause to consume values across target files.
+ context.sql('INSERT INTO paimon.prepared_merge.t (id, value) VALUES (1,
10)')
+ context.sql('INSERT INTO paimon.prepared_merge.t (id, value) VALUES (2,
20)')
+ table = PaimonCatalog({'warehouse':
str(tmp_path)}).get_table('prepared_merge.t')
+ builder =
(table.new_stream_write_builder().with_commit_user('prepared-merge')
+ if stream else table.new_batch_write_builder())
+ source = pa.table({'id': [1, 2], 'value': [11, 22]}, schema=pa.schema([
+ ('id', pa.int32()), ('value', pa.int32())]))
+ update = builder.new_update()
+ kwargs = dict(commit_identifier=42) if stream else {}
+ false_clause = dict(delete=False, condition=dict(sql='"s.value" < 0'),
+ assignments=[('value', 'literal', 'invalid int')])
+ assert update.merge_into(source, on=[('id', 'id')],
when_matched=[false_clause],
+ when_not_matched=[], **kwargs) == []
+ messages = update.merge_into(source, on=[('id', 'id')], when_matched=[
+ dict(delete=False, condition=dict(sql='"s.value" IN (SELECT v FROM
(VALUES (11), (22)) AS m(v))'),
+ assignments=[('value', 'literal', pa.chunked_array([[100],
[200]]))])],
+ when_not_matched=[], **kwargs)
+ # Preparation leaves the visible snapshot unchanged.
+
+ def rows():
+ return pa.Table.from_batches(context.sql(
+ 'SELECT id, value FROM paimon.prepared_merge.t ORDER BY
id')).to_pylist()
+ assert rows() == [dict(id=1, value=10), dict(id=2, value=20)]
+ commit = builder.new_commit()
+ commit.commit(42, messages) if stream else commit.commit(messages)
+ assert rows() == [dict(id=1, value=100), dict(id=2, value=200)]
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 7043303e..cfab74c0 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -52,6 +52,7 @@ mod full_text_search;
mod hybrid_search;
mod lateral_vector_search;
mod memory;
+mod merge_condition;
mod merge_into;
mod partition_count_pushdown;
mod physical_plan;
@@ -88,6 +89,7 @@ pub use error::to_datafusion_error;
#[cfg(feature = "fulltext")]
pub use full_text_search::{register_full_text_search, FullTextSearchFunction};
pub use hybrid_search::{register_hybrid_search, HybridSearchFunction};
+pub use merge_condition::compile_merge_condition;
pub use physical_plan::PaimonTableScan;
pub use relation_planner::PaimonRelationPlanner;
pub use sql_context::{SQLContext, SQLContextBuilder};
diff --git a/crates/integrations/datafusion/src/merge_condition.rs
b/crates/integrations/datafusion/src/merge_condition.rs
new file mode 100644
index 00000000..b1a6363f
--- /dev/null
+++ b/crates/integrations/datafusion/src/merge_condition.rs
@@ -0,0 +1,267 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! DataFusion conditions for the engine-independent core MERGE operation.
+
+use datafusion::arrow::array::{BooleanArray, UInt64Array};
+use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
+use datafusion::logical_expr::{Expr, LogicalPlan};
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::prelude::SessionContext;
+use paimon::table::MergeCondition;
+use std::collections::BTreeSet;
+use std::sync::{Arc, Mutex};
+
+fn invalid(error: impl std::fmt::Display) -> paimon::Error {
+ paimon::Error::DataInvalid {
+ message: format!("Invalid MERGE condition: {error}"),
+ source: None,
+ }
+}
+
+const ROW_INDEX: &str = "__merge_condition_row";
+
+// Full SQL planning handles scalar/IN subqueries. Preserve row identity
because
+// their join rewrites may reorder the input. Core consumes a mask in input
order.
+async fn evaluate_query(sql: &str, batch: RecordBatch) ->
paimon::Result<BooleanArray> {
+ let mut fields = batch.schema().fields().to_vec();
+ fields.push(Arc::new(Field::new(ROW_INDEX, DataType::UInt64, false)));
+ let mut columns = batch.columns().to_vec();
+ columns.push(Arc::new(UInt64Array::from_iter_values(
+ 0..batch.num_rows() as u64,
+ )));
+ let input = RecordBatch::try_new(Arc::new(Schema::new(fields)),
columns).map_err(invalid)?;
+ let context = SessionContext::new();
+ context
+ .register_batch("_merge_batch", input)
+ .map_err(invalid)?;
+ let output = context
+ .sql(&format!("SELECT {ROW_INDEX} FROM _merge_batch WHERE {sql}"))
+ .await
+ .map_err(invalid)?
+ .collect()
+ .await
+ .map_err(invalid)?;
+ let mut mask = vec![false; batch.num_rows()];
+ for output in output {
+ let indices = output
+ .column(0)
+ .as_any()
+ .downcast_ref::<UInt64Array>()
+ .ok_or_else(|| invalid("Missing condition row index"))?;
+ for row in 0..output.num_rows() {
+ let index = indices.value(row) as usize;
+ if index >= mask.len() || mask[index] {
+ return Err(invalid("Condition changed input row cardinality"));
+ }
+ mask[index] = true;
+ }
+ }
+ Ok(BooleanArray::from(mask))
+}
+
+/// Compile a SQL condition over `t.<column>` and `s.<column>` Arrow fields.
+/// Target and source dependencies include correlated subquery references.
+/// Core controls matching and action order; this adapter only evaluates SQL.
+pub async fn compile_merge_condition(
+ sql: String,
+ schema: SchemaRef,
+) -> paimon::Result<MergeCondition> {
+ let context = SessionContext::new();
+ context
+ .register_batch("_merge_batch", RecordBatch::new_empty(schema))
+ .map_err(invalid)?;
+ let frame = context
+ .sql(&format!("SELECT * FROM _merge_batch WHERE {sql}"))
+ .await
+ .map_err(invalid)?;
+ let plan = frame.into_unoptimized_plan();
+ let LogicalPlan::Projection(projection) = plan else {
+ return Err(invalid("Expected a condition expression"));
+ };
+ let LogicalPlan::Filter(filter) = projection.input.as_ref() else {
+ return Err(invalid("Expected a condition expression"));
+ };
+ let mut target_columns = BTreeSet::new();
+ let mut source_columns = BTreeSet::new();
+ // Start at the filter so SELECT * does not turn every field into a
+ // dependency. Subquery outer references are not in Expr::column_refs().
+ LogicalPlan::Filter(filter.clone())
+ .apply_with_subqueries(|plan| {
+ for expression in plan.expressions() {
+ expression.apply(|expression| {
+ if let Expr::Column(column) |
Expr::OuterReferenceColumn(_, column) = expression
+ {
+ if let Some(name) = column.name.strip_prefix("t.") {
+ target_columns.insert(name.to_string());
+ }
+ if let Some(name) = column.name.strip_prefix("s.") {
+ source_columns.insert(name.to_string());
+ }
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })
+ .map_err(invalid)?;
+ // Ordinary expressions compile once. Subqueries need the complete planner;
+ // validate that plan before matching, then execute against each action
batch.
+ let compiled = match context
+ .state()
+ .create_physical_expr(filter.predicate.clone(), filter.input.schema())
+ {
+ Ok(expression) => Some(expression),
+ Err(_) => {
+ let plan = context
+ .state()
+ .optimize(&LogicalPlan::Projection(projection))
+ .map_err(invalid)?;
+ context
+ .state()
+ .create_physical_plan(&plan)
+ .await
+ .map_err(invalid)?;
+ None
+ }
+ };
+ let cached = Arc::new(Mutex::new(None::<(SchemaRef, Arc<dyn
PhysicalExpr>)>));
+ Ok(MergeCondition {
+ target_columns: target_columns.into_iter().collect(),
+ source_columns: source_columns.into_iter().collect(),
+ evaluate: Arc::new(move |batch| {
+ let compiled = compiled.clone();
+ let cached = cached.clone();
+ let sql = sql.clone();
+ Box::pin(async move {
+ let Some(compiled) = compiled else {
+ return evaluate_query(&sql, batch).await;
+ };
+ let mut cached = cached.lock().unwrap();
+ if cached
+ .as_ref()
+ .is_none_or(|(schema, _)| schema != &batch.schema())
+ {
+ let expression =
datafusion::physical_expr::utils::reassign_expr_columns(
+ compiled,
+ batch.schema().as_ref(),
+ )
+ .map_err(invalid)?;
+ *cached = Some((batch.schema(), expression));
+ }
+ let value = cached
+ .as_ref()
+ .unwrap()
+ .1
+ .evaluate(&batch)
+ .map_err(invalid)?
+ .into_array(batch.num_rows())
+ .map_err(invalid)?;
+ value
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .cloned()
+ .ok_or_else(|| invalid("Condition must be BOOLEAN"))
+ })
+ }),
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use datafusion::arrow::array::{ArrayRef, Int32Array};
+
+ fn batch(names: &[&str], values: &[Vec<Option<i32>>]) -> RecordBatch {
+ RecordBatch::try_from_iter(names.iter().zip(values).map(|(name,
values)| {
+ (
+ *name,
+ Arc::new(Int32Array::from(values.clone())) as ArrayRef,
+ )
+ }))
+ .unwrap()
+ }
+
+ // Runs on a current-thread runtime: condition evaluation must await its
+ // plan rather than synchronously blocking on another Tokio runtime.
+ #[tokio::test]
+ async fn projected_and_reordered_conditions_preserve_nulls() {
+ let input = batch(
+ &["t.id", "t.value", "s.value"],
+ &[
+ vec![Some(1), Some(2)],
+ vec![Some(10), Some(20)],
+ vec![Some(11), None],
+ ],
+ );
+ let condition = compile_merge_condition("\"s.value\" >
\"t.value\"".into(), input.schema())
+ .await
+ .unwrap();
+ let projected = input.project(&[2, 1]).unwrap();
+ let result = (condition.evaluate)(projected).await.unwrap();
+ assert_eq!(result.iter().collect::<Vec<_>>(), vec![Some(true), None]);
+ assert_eq!(condition.source_columns, vec!["value"]);
+ }
+
+ #[tokio::test]
+ async fn scalar_and_in_subqueries_preserve_input_order() {
+ let input = batch(&["s.value"], &[vec![Some(3), None, Some(1),
Some(2)]]);
+ for sql in [
+ "\"s.value\" > (SELECT 1)",
+ "\"s.value\" IN (SELECT v FROM (VALUES (2), (3)) AS m(v))",
+ "\"s.value\" > (SELECT MAX(v) FROM (VALUES (0), (1)) AS m(v))",
+ ] {
+ let condition = compile_merge_condition(sql.into(), input.schema())
+ .await
+ .unwrap();
+ let result = (condition.evaluate)(input.clone()).await.unwrap();
+ assert_eq!(
+ result.iter().collect::<Vec<_>>(),
+ vec![Some(true), Some(false), Some(false), Some(true)],
+ "{sql}"
+ );
+ }
+ }
+ #[tokio::test]
+ async fn self_conditions_collect_correlated_subquery_dependencies() {
+ let input = batch(
+ &["s.id", "t.value", "s.unused"],
+ &[
+ vec![Some(1), Some(2)],
+ vec![Some(10), Some(20)],
+ vec![Some(3), Some(4)],
+ ],
+ );
+ let condition = compile_merge_condition(
+ "EXISTS (SELECT 1 WHERE \"s.id\" = 1 AND \"t.value\" > 5)".into(),
+ input.schema(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(condition.source_columns, vec!["id"]);
+ assert_eq!(condition.target_columns, vec!["value"]);
+ let result = (condition.evaluate)(input.project(&[1, 0]).unwrap())
+ .await
+ .unwrap();
+ assert_eq!(
+ result.iter().collect::<Vec<_>>(),
+ vec![Some(true), Some(false)]
+ );
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 168f3d91..24a3cf22 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -142,6 +142,9 @@ mod sorted_global_index_options;
mod source;
mod stats_filter;
pub(crate) mod table_commit;
+mod table_merge_into;
+#[cfg(test)]
+mod table_merge_into_tests;
mod table_read;
mod table_scan;
mod table_update;
@@ -219,6 +222,9 @@ pub use source::{
merge_row_ranges, DataSplit, DataSplitBuilder, DeletionFile,
PartitionBucket, Plan, RowRange,
};
pub use table_commit::TableCommit;
+pub use table_merge_into::{
+ MergeAssignment, MergeCondition, MergeSource, WhenMatched, WhenNotMatched,
+};
pub use table_read::{AuditLogRead, TableRead};
pub use table_scan::{AuditLogScan, TableScan};
pub use table_update::TableUpdate;
diff --git a/crates/paimon/src/table/table_merge_into.rs
b/crates/paimon/src/table/table_merge_into.rs
new file mode 100644
index 00000000..e788867b
--- /dev/null
+++ b/crates/paimon/src/table/table_merge_into.rs
@@ -0,0 +1,588 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Data-evolution MERGE orchestration, independent of an expression engine.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use arrow_array::{Array, BooleanArray, Int64Array, RecordBatch, UInt64Array};
+use arrow_row::{RowConverter, SortField};
+use arrow_schema::{DataType, Field, Schema, SchemaRef};
+use arrow_select::{concat::concat_batches, take::take};
+use futures::{future::BoxFuture, TryStreamExt};
+
+use super::data_evolution_writer::RowIdFileIndex;
+use super::{CommitMessage, DataEvolutionWriter, Table, TableUpdateByRowId,
UpdateAssignment};
+
+const ROW_ID: &str = "_ROW_ID";
+type ConditionFunction =
+ dyn Fn(RecordBatch) -> BoxFuture<'static, crate::Result<BooleanArray>> +
Send + Sync;
+
+/// An engine-compiled condition. NULL evaluates to false when choosing a
clause.
+/// Dependencies control target projection, including source fields for self
MERGE.
+/// Evaluation is asynchronous so expression engines can execute subquery
plans.
+#[derive(Clone)]
+pub struct MergeCondition {
+ pub target_columns: Vec<String>,
+ pub source_columns: Vec<String>,
+ pub evaluate: Arc<ConditionFunction>,
+}
+
+/// SET/INSERT values. Column references are evaluated by core, without
callbacks.
+#[derive(Clone)]
+pub enum MergeAssignment {
+ SourceColumn(String),
+ TargetColumn(String),
+ /// Evaluate once across all selected rows of this clause.
+ /// Matched rows follow target scan order; inserts follow source input
order.
+ Value(UpdateAssignment),
+}
+
+/// Ordered matched clauses, corresponding to WHEN MATCHED UPDATE/DELETE.
+#[derive(Clone)]
+pub struct WhenMatched {
+ pub condition: Option<MergeCondition>,
+ pub delete: bool,
+ pub assignments: Vec<(String, MergeAssignment)>,
+}
+
+/// Ordered WHEN NOT MATCHED INSERT clauses.
+#[derive(Clone)]
+pub struct WhenNotMatched {
+ pub condition: Option<MergeCondition>,
+ pub assignments: Vec<(String, MergeAssignment)>,
+}
+
+/// Already materialized input, or the target itself joined on its row ID.
+pub enum MergeSource {
+ Batches(Vec<RecordBatch>),
+ SelfTable,
+}
+
+fn invalid(message: impl Into<String>) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+fn gather(batch: &RecordBatch, rows: &[usize]) -> crate::Result<RecordBatch> {
+ let indices = UInt64Array::from(rows.iter().map(|&row| row as
u64).collect::<Vec<_>>());
+ let columns = batch
+ .columns()
+ .iter()
+ .map(|array| {
+ take(array.as_ref(), &indices, None).map_err(|error|
invalid(error.to_string()))
+ })
+ .collect::<crate::Result<_>>()?;
+ RecordBatch::try_new(batch.schema(), columns).map_err(|error|
invalid(error.to_string()))
+}
+
+fn aliases(target: Option<&RecordBatch>, source: &RecordBatch) ->
crate::Result<RecordBatch> {
+ let mut fields = Vec::new();
+ let mut columns = Vec::new();
+ if let Some(target) = target {
+ fields.push(Arc::new(Field::new(ROW_ID, DataType::Int64, false)));
+ columns.push(target.column_by_name(ROW_ID).unwrap().clone());
+ for (field, column) in
target.schema().fields().iter().zip(target.columns()) {
+ fields.push(Arc::new(
+ field
+ .as_ref()
+ .clone()
+ .with_name(format!("t.{}", field.name())),
+ ));
+ columns.push(column.clone());
+ }
+ } else {
+ fields.push(Arc::new(Field::new(ROW_ID, DataType::Int64, false)));
+ columns.push(Arc::new(Int64Array::from(vec![0; source.num_rows()])));
+ }
+ for (field, column) in
source.schema().fields().iter().zip(source.columns()) {
+ fields.push(Arc::new(
+ field
+ .as_ref()
+ .clone()
+ .with_name(format!("s.{}", field.name())),
+ ));
+ columns.push(column.clone());
+ }
+ RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)
+ .map_err(|error| invalid(error.to_string()))
+}
+
+async fn select_clause(
+ batch: RecordBatch,
+ condition: &Option<MergeCondition>,
+) -> crate::Result<(RecordBatch, RecordBatch)> {
+ let mask = match condition {
+ Some(condition) => Some((condition.evaluate)(batch.clone()).await?),
+ None => None,
+ };
+ if mask
+ .as_ref()
+ .is_some_and(|mask| mask.len() != batch.num_rows())
+ {
+ return Err(invalid("MERGE condition length must match input row
count"));
+ }
+ let mut selected = Vec::new();
+ let mut remaining = Vec::new();
+ for row in 0..batch.num_rows() {
+ if mask
+ .as_ref()
+ .is_none_or(|mask| !mask.is_null(row) && mask.value(row))
+ {
+ selected.push(row);
+ } else {
+ remaining.push(row);
+ }
+ }
+ Ok((gather(&batch, &selected)?, gather(&batch, &remaining)?))
+}
+
+fn column_assignment(
+ batches: &[RecordBatch],
+ prefix: &str,
+ name: &str,
+) -> crate::Result<UpdateAssignment> {
+ Ok(UpdateAssignment::Array(
+ batches
+ .iter()
+ .map(|batch| {
+ batch
+ .column_by_name(&format!("{prefix}.{name}"))
+ .cloned()
+ .ok_or_else(|| invalid(format!("Missing MERGE {prefix}
column {name}")))
+ })
+ .collect::<crate::Result<_>>()?,
+ ))
+}
+
+fn assigned(
+ batches: &[RecordBatch],
+ assignments: &[(String, MergeAssignment)],
+ columns: &[String],
+ schema: SchemaRef,
+ insert: bool,
+) -> crate::Result<Vec<RecordBatch>> {
+ let mut values = Vec::new();
+ for name in columns {
+ let value = assignments
+ .iter()
+ .find(|(column, _)| column == name)
+ .map(|(_, value)| value);
+ let assignment = match value {
+ Some(MergeAssignment::Value(value)) => value.clone(),
+ Some(MergeAssignment::SourceColumn(column)) =>
column_assignment(batches, "s", column)?,
+ Some(MergeAssignment::TargetColumn(column)) =>
column_assignment(batches, "t", column)?,
+ None if insert =>
UpdateAssignment::Scalar(arrow_array::new_null_array(
+ schema.field_with_name(name).unwrap().data_type(),
+ 1,
+ )),
+ None => column_assignment(batches, "t", name)?,
+ };
+ values.push((name.clone(), assignment));
+ }
+ super::update_assignment::assigned_batches(batches, values, schema)
+}
+
+pub(super) async fn merge_into(
+ table: &Table,
+ commit_user: &str,
+ source: MergeSource,
+ on: Vec<(String, String)>,
+ matched: Vec<WhenMatched>,
+ not_matched: Vec<WhenNotMatched>,
+) -> crate::Result<Vec<CommitMessage>> {
+ if on.is_empty() || matched.is_empty() && not_matched.is_empty() {
+ return Err(invalid("MERGE requires ON keys and at least one action"));
+ }
+ let schema =
crate::arrow::build_target_arrow_schema(table.schema().fields())?;
+ let mut update_columns = Vec::new();
+ let mut projection: HashSet<String> = on.iter().map(|(target, _)|
target.clone()).collect();
+ projection.insert(ROW_ID.into());
+ for clause in &matched {
+ if clause.delete && !clause.assignments.is_empty() {
+ return Err(invalid("MERGE DELETE cannot have assignments"));
+ }
+ for (name, value) in &clause.assignments {
+ schema
+ .field_with_name(name)
+ .map_err(|error| invalid(error.to_string()))?;
+ if table.schema().partition_keys().contains(name) {
+ return Err(invalid("MERGE does not support updating partition
columns"));
+ }
+ if !update_columns.contains(name) {
+ update_columns.push(name.clone());
+ }
+ if let MergeAssignment::TargetColumn(name) = value {
+ projection.insert(name.clone());
+ }
+ if matches!(value,
MergeAssignment::Value(UpdateAssignment::Function(_))) {
+ projection.extend(schema.fields().iter().map(|field|
field.name().clone()));
+ }
+ }
+ if let Some(condition) = &clause.condition {
+ projection.extend(condition.target_columns.clone());
+ }
+ }
+ projection.extend(update_columns.clone());
+ for clause in ¬_matched {
+ if clause
+ .condition
+ .as_ref()
+ .is_some_and(|condition| !condition.target_columns.is_empty())
+ || clause
+ .assignments
+ .iter()
+ .any(|(_, value)| matches!(value,
MergeAssignment::TargetColumn(_)))
+ {
+ return Err(invalid("WHEN NOT MATCHED cannot reference target
columns"));
+ }
+ }
+ for conditions in [
+ matched
+ .iter()
+ .map(|clause| &clause.condition)
+ .collect::<Vec<_>>(),
+ not_matched.iter().map(|clause| &clause.condition).collect(),
+ ] {
+ if conditions
+ .iter()
+ .take(conditions.len().saturating_sub(1))
+ .any(|condition| condition.is_none())
+ {
+ return Err(invalid("Only the last MERGE clause may omit its
condition"));
+ }
+ }
+ let _ = DataEvolutionWriter::new(table, update_columns.clone())?;
+ let snapshot = super::time_travel::resolve_snapshot(table).await?;
+ let scan_table = match &snapshot {
+ Some(snapshot) => table.copy_with_pinned_snapshot(snapshot),
+ None => table.clone(),
+ }
+ .copy_with_options(HashMap::from([
+ ("scalar-index.search-mode".into(), "FULL".into()),
+ ("scan.mode".into(), "default".into()),
+ ]));
+ let self_merge = matches!(source, MergeSource::SelfTable);
+ if self_merge {
+ if on != [(ROW_ID.into(), ROW_ID.into())] || !not_matched.is_empty() {
+ return Err(invalid(
+ "Self MERGE requires ON _ROW_ID and no insert clauses",
+ ));
+ }
+ for clause in &matched {
+ if let Some(condition) = &clause.condition {
+ projection.extend(condition.source_columns.clone());
+ }
+ for (_, assignment) in &clause.assignments {
+ if let MergeAssignment::SourceColumn(name) = assignment {
+ projection.insert(name.clone());
+ }
+ }
+ }
+ }
+ let projection = projection.into_iter().collect::<Vec<_>>();
+ let mut builder = scan_table.new_read_builder();
+
builder.with_projection(&projection.iter().map(String::as_str).collect::<Vec<_>>())?;
+ let plan = if snapshot.is_some() {
+ builder.new_scan().with_scan_all_files().plan().await?
+ } else {
+ super::source::Plan::new(Vec::new())
+ };
+ let index = RowIdFileIndex::from_splits(scan_table.clone(),
plan.splits())?;
+ let source = match source {
+ MergeSource::Batches(batches) => {
+ let first = batches
+ .first()
+ .ok_or_else(|| invalid("MERGE source needs an Arrow schema"))?;
+ if batches.iter().any(|batch| batch.schema() != first.schema()) {
+ return Err(invalid("MERGE source batches must share a
schema"));
+ }
+ let names = first
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.name().clone())
+ .collect::<HashSet<_>>();
+ if names.len() != first.num_columns() {
+ return Err(invalid("MERGE source has duplicate column names"));
+ }
+ Some(
+ concat_batches(&first.schema(), &batches)
+ .map_err(|error| invalid(error.to_string()))?,
+ )
+ }
+ MergeSource::SelfTable => None,
+ };
+ let source_schema = source
+ .as_ref()
+ .map_or_else(|| schema.clone(), RecordBatch::schema);
+ for (assignments, condition, insert) in matched
+ .iter()
+ .map(|clause| (&clause.assignments, &clause.condition, false))
+ .chain(
+ not_matched
+ .iter()
+ .map(|clause| (&clause.assignments, &clause.condition, true)),
+ )
+ {
+ let mut seen = HashSet::new();
+ for (name, value) in assignments {
+ if !seen.insert(name) {
+ return Err(invalid(format!("Duplicate MERGE assignment
{name}")));
+ }
+ schema
+ .field_with_name(name)
+ .map_err(|error| invalid(error.to_string()))?;
+ match value {
+ MergeAssignment::SourceColumn(name) => {
+ if !(self_merge && name == ROW_ID) {
+ source_schema
+ .field_with_name(name)
+ .map_err(|error| invalid(error.to_string()))?;
+ }
+ }
+ MergeAssignment::TargetColumn(name) => {
+ if insert {
+ return Err(invalid("INSERT cannot reference target
columns"));
+ }
+ if name != ROW_ID {
+ schema
+ .field_with_name(name)
+ .map_err(|error| invalid(error.to_string()))?;
+ }
+ }
+ MergeAssignment::Value(_) => {}
+ }
+ }
+ if let Some(condition) = condition {
+ for name in &condition.target_columns {
+ if name != ROW_ID {
+ schema
+ .field_with_name(name)
+ .map_err(|error| invalid(error.to_string()))?;
+ }
+ }
+ }
+ }
+ if matched.iter().any(|clause| clause.delete)
+ &&
!crate::spec::CoreOptions::new(table.schema().options()).deletion_vectors_enabled()
+ {
+ return Err(invalid(
+ "MERGE DELETE requires deletion-vectors.enabled=true",
+ ));
+ }
+ let mut converter = None;
+ let mut source_keys = HashMap::<Vec<u8>, Vec<usize>>::new();
+ let mut seen_source = vec![false; source.as_ref().map_or(0,
RecordBatch::num_rows)];
+ if let Some(source) = &source {
+ let keys = on
+ .iter()
+ .map(|(target, source_name)| {
+ let source_column = source
+ .column_by_name(source_name)
+ .ok_or_else(|| invalid(format!("Missing source ON key
{source_name}")))?;
+ let target_type = if target == ROW_ID {
+ &DataType::Int64
+ } else {
+ schema
+ .field_with_name(target)
+ .map_err(|error| invalid(error.to_string()))?
+ .data_type()
+ };
+ if source_column.data_type() != target_type
+ ||
!super::upsert_key_matcher::supported_key_type(target_type)
+ {
+ return Err(invalid(
+ "MERGE ON key types must match and support row
encoding",
+ ));
+ }
+ Ok(source_column.clone())
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let key_converter = RowConverter::new(
+ keys.iter()
+ .map(|column| SortField::new(column.data_type().clone()))
+ .collect(),
+ )
+ .map_err(|error| invalid(error.to_string()))?;
+ let rows = key_converter
+ .convert_columns(&keys)
+ .map_err(|error| invalid(error.to_string()))?;
+ for row in 0..source.num_rows() {
+ if keys.iter().all(|column| !column.is_null(row)) {
+ source_keys
+ .entry(rows.row(row).as_ref().to_vec())
+ .or_default()
+ .push(row);
+ }
+ }
+ converter = Some(key_converter);
+ }
+ // Spark validates cardinality before evaluating action predicates. A sole
+ // unconditional DELETE may match multiple source rows without ambiguity.
+ let check_cardinality = !matched.is_empty()
+ && !(matched.len() == 1 && matched[0].delete &&
matched[0].condition.is_none());
+ let mut joined = Vec::new();
+ let mut stream = builder.new_read()?.to_arrow(plan.splits())?;
+ while let Some(target) = stream.try_next().await? {
+ if let Some(source) = &source {
+ let keys = on
+ .iter()
+ .map(|(name, _)| target.column_by_name(name).unwrap().clone())
+ .collect::<Vec<_>>();
+ let rows = converter
+ .as_ref()
+ .unwrap()
+ .convert_columns(&keys)
+ .map_err(|error| invalid(error.to_string()))?;
+ let mut target_rows = Vec::new();
+ let mut source_rows = Vec::new();
+ for row in 0..target.num_rows() {
+ if keys.iter().any(|column| column.is_null(row)) {
+ continue;
+ }
+ if let Some(sources) = source_keys.get(rows.row(row).as_ref())
{
+ if check_cardinality && sources.len() > 1 {
+ return Err(invalid(
+ "MERGE matched multiple source rows to the same
target _ROW_ID",
+ ));
+ }
+ for &source in sources {
+ seen_source[source] = true;
+ }
+ target_rows.push(row);
+ source_rows.push(sources[0]);
+ }
+ }
+ if !target_rows.is_empty() {
+ joined.push(aliases(
+ Some(&gather(&target, &target_rows)?),
+ &gather(source, &source_rows)?,
+ )?);
+ }
+ } else if target.num_rows() > 0 {
+ joined.push(aliases(Some(&target), &target)?);
+ }
+ }
+ let mut updates = Vec::new();
+ let mut deletes = Vec::new();
+ let mut pending = joined;
+ for clause in matched {
+ let mut remaining = Vec::new();
+ let mut clause_rows = Vec::new();
+ for batch in pending {
+ let (selected, rest) = select_clause(batch,
&clause.condition).await?;
+ if rest.num_rows() > 0 {
+ remaining.push(rest);
+ }
+ if selected.num_rows() == 0 {
+ continue;
+ }
+ if clause.delete {
+ let ids = selected
+ .column_by_name(ROW_ID)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ deletes.extend(ids.values().iter().copied());
+ } else if !update_columns.is_empty() {
+ clause_rows.push(selected);
+ }
+ }
+ // Array/function assignments cover the complete clause selection, not
+ // each physical target batch. Shared assignment cursors consume chunks
+ // across these batches while preserving row IDs and scan order.
+ if !clause_rows.is_empty() {
+ updates.extend(assigned(
+ &clause_rows,
+ &clause.assignments,
+ &update_columns,
+ schema.clone(),
+ false,
+ )?);
+ }
+ pending = remaining;
+ }
+ let mut inserts = Vec::new();
+ if let Some(source) = source {
+ let unmatched = seen_source
+ .iter()
+ .enumerate()
+ .filter_map(|(row, seen)| (!seen).then_some(row))
+ .collect::<Vec<_>>();
+ let mut pending = aliases(None, &gather(&source, &unmatched)?)?;
+ let columns = schema
+ .fields()
+ .iter()
+ .map(|field| field.name().clone())
+ .collect::<Vec<_>>();
+ for clause in not_matched {
+ if pending.num_rows() == 0 {
+ break;
+ }
+ let (selected, rest) = select_clause(pending,
&clause.condition).await?;
+ pending = rest;
+ if selected.num_rows() > 0 {
+ for batch in assigned(
+ std::slice::from_ref(&selected),
+ &clause.assignments,
+ &columns,
+ schema.clone(),
+ true,
+ )? {
+ inserts.push(
+ batch
+
.project(&(1..batch.num_columns()).collect::<Vec<_>>())
+ .map_err(|error| invalid(error.to_string()))?,
+ );
+ }
+ }
+ }
+ }
+ // Materialize/validate every action before staging. Once prepared,
messages
+ // may already be referenced by publication; a later error must retain
their files.
+ let mut messages = Vec::new();
+ if !updates.is_empty() {
+ let mut updater = TableUpdateByRowId::with_index(table, index)?;
+ messages.extend(updater.update_columns(updates,
update_columns).await?);
+ }
+ if !deletes.is_empty() {
+ let mut writer = super::DataEvolutionDeleteWriter::new(&scan_table)?;
+ writer.add_row_ids(deletes)?;
+ messages.extend(writer.prepare_commit().await?);
+ }
+ if !inserts.is_empty() {
+ let mut writer = table
+ .new_write_builder()
+ .with_commit_user(commit_user.to_string())?
+ .new_write()?;
+ let result = async {
+ for batch in inserts {
+ writer.write_arrow_batch(&batch).await?;
+ }
+ writer.prepare_commit().await
+ }
+ .await;
+ writer.close().await;
+ messages.extend(result?);
+ }
+ Ok(messages)
+}
diff --git a/crates/paimon/src/table/table_merge_into_tests.rs
b/crates/paimon/src/table/table_merge_into_tests.rs
new file mode 100644
index 00000000..4b4ec6db
--- /dev/null
+++ b/crates/paimon/src/table/table_merge_into_tests.rs
@@ -0,0 +1,477 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{
+ MergeAssignment, MergeCondition, MergeSource, Table, TableCommit,
UpdateAssignment,
+ WhenMatched, WhenNotMatched,
+};
+use crate::catalog::Identifier;
+use crate::io::FileIOBuilder;
+use crate::spec::{DataType, IntType, Schema, TableSchema};
+use arrow_array::{Array, ArrayRef, BooleanArray, Int32Array, RecordBatch};
+use futures::TryStreamExt;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+fn table() -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .build()
+ .unwrap();
+ Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("db", "t"),
+ "memory:/merge".into(),
+ TableSchema::new(0, &schema),
+ None,
+ )
+}
+
+fn batch(ids: Vec<Option<i32>>, values: Vec<Option<i32>>) -> RecordBatch {
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(ids)) as ArrayRef),
+ ("value", Arc::new(Int32Array::from(values)) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+async fn seed(table: &Table, data: &RecordBatch) {
+ let mut writer = table
+ .new_write_builder()
+ .with_commit_user("merge")
+ .unwrap()
+ .new_write()
+ .unwrap();
+ writer.write_arrow_batch(data).await.unwrap();
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ writer.close().await;
+}
+
+async fn rows(table: &Table) -> Vec<(Option<i32>, Option<i32>)> {
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let mut stream =
builder.new_read().unwrap().to_arrow(plan.splits()).unwrap();
+ let mut values = Vec::new();
+ while let Some(batch) = stream.try_next().await.unwrap() {
+ let ids = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let data = batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ values.extend(ids.iter().zip(data.iter()));
+ }
+ values.sort();
+ values
+}
+
+fn update(condition: Option<MergeCondition>) -> WhenMatched {
+ WhenMatched {
+ condition,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }
+}
+
+#[tokio::test]
+async fn core_merge_preserves_row_ids_and_handles_sql_null_keys() {
+ let table = table();
+ seed(
+ &table,
+ &batch(
+ vec![Some(1), Some(2), None],
+ vec![Some(10), Some(20), Some(90)],
+ ),
+ )
+ .await;
+ let source = batch(
+ vec![Some(1), Some(3), None],
+ vec![Some(11), Some(30), Some(91)],
+ );
+ let updater = table
+ .new_write_builder()
+ .with_commit_user("merge")
+ .unwrap()
+ .new_update()
+ .unwrap();
+ let messages = updater
+ .merge_into(
+ MergeSource::Batches(vec![source]),
+ vec![("id".into(), "id".into())],
+ vec![update(None)],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: vec![
+ ("id".into(), MergeAssignment::SourceColumn("id".into())),
+ (
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ ),
+ ],
+ }],
+ )
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .filter(|message| !message.new_files.is_empty())
+ .any(|message| message.check_from_snapshot == Some(1)));
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (None, Some(90)),
+ (None, Some(91)),
+ (Some(1), Some(11)),
+ (Some(2), Some(20)),
+ (Some(3), Some(30))
+ ]
+ );
+}
+
+#[tokio::test]
+async fn duplicate_source_fails_before_condition_evaluation_and_staging() {
+ let table = table();
+ seed(&table, &batch(vec![Some(1)], vec![Some(10)])).await;
+ let files = table
+ .file_io()
+ .list_status_recursive(table.location())
+ .await
+ .unwrap()
+ .len();
+ let calls = Arc::new(AtomicUsize::new(0));
+ let evaluated = calls.clone();
+ let condition = MergeCondition {
+ target_columns: Vec::new(),
+ source_columns: Vec::new(),
+ evaluate: Arc::new(move |batch| {
+ evaluated.fetch_add(1, Ordering::Relaxed);
+ Box::pin(async move { Ok(BooleanArray::from(vec![false;
batch.num_rows()])) })
+ }),
+ };
+ let updater = table.new_write_builder().new_update().unwrap();
+ let error = updater
+ .merge_into(
+ MergeSource::Batches(vec![batch(
+ vec![Some(1), Some(1)],
+ vec![Some(11), Some(12)],
+ )]),
+ vec![("id".into(), "id".into())],
+ vec![update(Some(condition))],
+ Vec::new(),
+ )
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("multiple source rows"));
+ assert_eq!(calls.load(Ordering::Relaxed), 0);
+ assert_eq!(
+ table
+ .file_io()
+ .list_status_recursive(table.location())
+ .await
+ .unwrap()
+ .len(),
+ files
+ );
+}
+
+#[tokio::test]
+async fn unconditional_delete_collapses_duplicate_source_matches() {
+ let table = table();
+ seed(
+ &table,
+ &batch(vec![Some(1), Some(2)], vec![Some(10), Some(20)]),
+ )
+ .await;
+ let updater = table.new_write_builder().new_update().unwrap();
+ let messages = updater
+ .merge_into(
+ MergeSource::Batches(vec![batch(
+ vec![Some(1), Some(1)],
+ vec![Some(11), Some(12)],
+ )]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: true,
+ assignments: Vec::new(),
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(rows(&table).await, vec![(Some(2), Some(20))]);
+}
+
+#[tokio::test]
+async fn self_merge_null_condition_falls_through_to_next_clause() {
+ let table = table();
+ seed(
+ &table,
+ &batch(vec![Some(1), Some(2)], vec![Some(10), Some(20)]),
+ )
+ .await;
+ let null_condition = MergeCondition {
+ target_columns: Vec::new(),
+ source_columns: Vec::new(),
+ evaluate: Arc::new(|batch| {
+ Box::pin(async move { Ok(BooleanArray::from(vec![None;
batch.num_rows()])) })
+ }),
+ };
+ let value =
MergeAssignment::Value(UpdateAssignment::Scalar(Arc::new(Int32Array::from(vec![
+ 77,
+ ]))));
+ let updater = table.new_write_builder().new_update().unwrap();
+ let messages = updater
+ .merge_into(
+ MergeSource::SelfTable,
+ vec![("_ROW_ID".into(), "_ROW_ID".into())],
+ vec![
+ update(Some(null_condition)),
+ WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![("value".into(), value)],
+ },
+ ],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![(Some(1), Some(77)), (Some(2), Some(77))]
+ );
+}
+
+#[tokio::test]
+async fn core_merge_preserves_all_assignment_chunks() {
+ let table = table();
+ seed(
+ &table,
+ &batch(vec![Some(1), Some(2)], vec![Some(10), Some(20)]),
+ )
+ .await;
+ let values = MergeAssignment::Value(UpdateAssignment::Array(vec![
+ Arc::new(Int32Array::from(vec![100])),
+ Arc::new(Int32Array::from(vec![200])),
+ ]));
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Batches(vec![batch(vec![Some(1), Some(2)],
vec![Some(0), Some(0)])]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![("value".into(), values)],
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![(Some(1), Some(100)), (Some(2), Some(200))]
+ );
+}
+
+#[tokio::test]
+async fn core_merge_rejects_inconsistent_source_schemas_before_staging() {
+ let table = table();
+ seed(&table, &batch(vec![Some(1)], vec![Some(10)])).await;
+ let source = batch(vec![Some(1)], vec![Some(11)]);
+ let reversed = source.project(&[1, 0]).unwrap();
+ let updater = table.new_write_builder().new_update().unwrap();
+ let error = updater
+ .merge_into(
+ MergeSource::Batches(vec![source, reversed]),
+ vec![("id".into(), "id".into())],
+ vec![update(None)],
+ Vec::new(),
+ )
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("must share a schema"));
+ assert_eq!(rows(&table).await, vec![(Some(1), Some(10))]);
+}
+
+#[tokio::test]
+async fn core_merge_consumes_assignments_across_all_selected_files() {
+ for chunked in [false, true] {
+ let table = table();
+ for id in 1..=3 {
+ seed(&table, &batch(vec![Some(id)], vec![Some(id * 10)])).await;
+ }
+ let values: Vec<ArrayRef> = if chunked {
+ vec![
+ Arc::new(Int32Array::from(vec![100])),
+ Arc::new(Int32Array::from(vec![200])),
+ ]
+ } else {
+ vec![Arc::new(Int32Array::from(vec![100, 200]))]
+ };
+ let condition = MergeCondition {
+ target_columns: Vec::new(),
+ source_columns: vec!["id".into()],
+ evaluate: Arc::new(|batch| {
+ Box::pin(async move {
+ let ids = batch
+ .column_by_name("s.id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ Ok(BooleanArray::from(
+ ids.iter().map(|id| id.map(|id| id <
3)).collect::<Vec<_>>(),
+ ))
+ })
+ }),
+ };
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Batches(vec![batch(
+ vec![Some(1), Some(2), Some(3)],
+ vec![Some(11), Some(22), Some(33)],
+ )]),
+ vec![("id".into(), "id".into())],
+ vec![
+ WhenMatched {
+ condition: Some(condition),
+ delete: false,
+ assignments: vec![
+ ("id".into(),
MergeAssignment::TargetColumn("id".into())),
+ (
+ "value".into(),
+
MergeAssignment::Value(UpdateAssignment::Array(values)),
+ ),
+ ],
+ },
+ update(None),
+ ],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ (Some(1), Some(100)),
+ (Some(2), Some(200)),
+ (Some(3), Some(33))
+ ]
+ );
+ }
+}
+
+#[tokio::test]
+async fn core_merge_evaluates_functions_once_for_the_complete_clause() {
+ let table = table();
+ for id in 1..=2 {
+ seed(&table, &batch(vec![Some(id)], vec![Some(id * 10)])).await;
+ }
+ let calls = Arc::new(AtomicUsize::new(0));
+ let evaluated = calls.clone();
+ let values = UpdateAssignment::Function(Arc::new(move |batches| {
+ evaluated.fetch_add(1, Ordering::Relaxed);
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
2);
+ Ok(batches
+ .iter()
+ .map(|batch| {
+ let ids = batch
+ .column_by_name("s.id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ Arc::new(Int32Array::from(
+ ids.iter()
+ .map(|id| id.map(|id| id * 100))
+ .collect::<Vec<_>>(),
+ )) as ArrayRef
+ })
+ .collect())
+ }));
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Batches(vec![batch(
+ vec![Some(1), Some(2)],
+ vec![Some(11), Some(22)],
+ )]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![("value".into(),
MergeAssignment::Value(values))],
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ assert_eq!(calls.load(Ordering::Relaxed), 1);
+ TableCommit::new(table.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ rows(&table).await,
+ vec![(Some(1), Some(100)), (Some(2), Some(200))]
+ );
+}
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index 480b776f..012fa056 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -218,6 +218,26 @@ impl TableUpdate {
writer.prepare_commit().await
}
+ /// Prepare MERGE actions against one target snapshot. Clauses run in
order,
+ /// SQL NULL keys do not match, and ambiguous source matches are rejected.
+ pub async fn merge_into(
+ &self,
+ source: super::MergeSource,
+ on: Vec<(String, String)>,
+ when_matched: Vec<super::WhenMatched>,
+ when_not_matched: Vec<super::WhenNotMatched>,
+ ) -> crate::Result<Vec<CommitMessage>> {
+ super::table_merge_into::merge_into(
+ &self.table,
+ &self.commit_user,
+ source,
+ on,
+ when_matched,
+ when_not_matched,
+ )
+ .await
+ }
+
/// Delete rows by row ID using deletion vectors.
pub async fn delete_by_row_id(&self, row_ids: Vec<i64>) ->
crate::Result<Vec<CommitMessage>> {
if row_ids.is_empty() {
