Add MERGE INTO types to datafusion-expr

Add new types to support MERGE INTO DML statements:
- MergeIntoOp: Carries ON condition and WHEN clauses
- MergeIntoClause: Single WHEN clause (kind, predicate, action)
- MergeIntoClauseKind: Matched/NotMatched/NotMatchedByTarget/NotMatchedBySource
- MergeIntoAction: Update/Insert/Delete actions

Add WriteOp::MergeInto variant to WriteOp enum.

Wire MergeInto through datafusion-proto: extend the DmlNode message with
a MERGE_INTO type tag and a MergeIntoOpNode payload field, plus matching
serializers in to_proto.rs and from_proto.rs. The tag-only conversion
From<dml_node::Type> for WriteOp cannot reconstruct the MergeInto payload,
so callers must use the new fallible parse_write_op helper.

Document that MergeIntoClauseKind::NotMatched and NotMatchedByTarget are
semantically equivalent and must be treated identically downstream.

Add unit tests in datafusion-expr and proto round-trip tests in
datafusion-proto.

Author: wirybeaver

datafusion/expr/src/logical_plan/dml.rs

@@ -25,7 +25,7 @@ use arrow::datatypes::{DataType, Field, Schema};
 use datafusion_common::file_options::file_type::FileType;
 use datafusion_common::{DFSchemaRef, TableReference};
 
-use crate::{LogicalPlan, TableSource};
+use crate::{Expr, LogicalPlan, TableSource};
 
 /// Operator that copies the contents of a database to file(s)
 #[derive(Clone)]
@@ -239,6 +239,8 @@ pub enum WriteOp {
     Ctas,
     /// `TRUNCATE` operation
     Truncate,
+    /// `MERGE INTO` operation
+    MergeInto(MergeIntoOp),
 }
 
 impl WriteOp {
@@ -250,6 +252,7 @@ impl WriteOp {
             WriteOp::Update => "Update",
             WriteOp::Ctas => "Ctas",
             WriteOp::Truncate => "Truncate",
+            WriteOp::MergeInto(_) => "MergeInto",
         }
     }
 }
@@ -291,10 +294,94 @@ impl Display for InsertOp {
     }
 }
 
+/// Describes a MERGE INTO operation's parameters.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+pub struct MergeIntoOp {
+    /// The join condition from `ON <expr>`.
+    pub on: Expr,
+    /// The WHEN clauses, in the order they appeared in the SQL.
+    pub clauses: Vec<MergeIntoClause>,
+}
+
+/// A single WHEN clause within a MERGE INTO statement.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+pub struct MergeIntoClause {
+    /// Whether this fires on matched or unmatched rows.
+    pub kind: MergeIntoClauseKind,
+    /// Optional additional predicate (`AND <expr>`).
+    pub predicate: Option<Expr>,
+    /// The action to take.
+    pub action: MergeIntoAction,
+}
+
+/// Which rows a MERGE WHEN clause applies to.
+///
+/// Mirrors `sqlparser::ast::MergeClauseKind` so that the SQL spelling is
+/// preserved through the logical plan.
+///
+/// **Note on `NotMatched` vs `NotMatchedByTarget`:** these two variants are
+/// semantically identical — both describe a source row that has no matching
+/// target row. `NotMatched` is the SQL standard short form (used by
+/// Snowflake, Postgres, SQL Server); `NotMatchedByTarget` is BigQuery's
+/// explicit form added for symmetry with `NotMatchedBySource`. Downstream
+/// consumers (planners, table providers, optimizers) MUST treat the two
+/// variants identically.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
+pub enum MergeIntoClauseKind {
+    /// `WHEN MATCHED`
+    Matched,
+    /// `WHEN NOT MATCHED` — see type-level note for the equivalence with
+    /// [`NotMatchedByTarget`](Self::NotMatchedByTarget).
+    NotMatched,
+    /// `WHEN NOT MATCHED BY TARGET` — see type-level note for the
+    /// equivalence with [`NotMatched`](Self::NotMatched).
+    NotMatchedByTarget,
+    /// `WHEN NOT MATCHED BY SOURCE`
+    NotMatchedBySource,
+}
+
+/// The action for a single WHEN clause.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+pub enum MergeIntoAction {
+    /// `UPDATE SET col1 = expr1, col2 = expr2, ...`, stored as
+    /// `(column_name, value_expr)` pairs.
+    Update(Vec<(String, Expr)>),
+    /// `INSERT (col1, col2, ...) VALUES (expr1, expr2, ...)`. `columns` may
+    /// be empty, meaning all columns.
+    Insert {
+        columns: Vec<String>,
+        values: Vec<Expr>,
+    },
+    Delete,
+}
+
 fn make_count_schema() -> DFSchemaRef {
     Arc::new(
         Schema::new(vec![Field::new("count", DataType::UInt64, false)])
             .try_into()
             .unwrap(),
     )
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{col, lit};
+
+    #[test]
+    fn write_op_merge_into_name_and_display() {
+        let op = WriteOp::MergeInto(MergeIntoOp {
+            on: col("id").eq(col("source_id")),
+            clauses: vec![MergeIntoClause {
+                kind: MergeIntoClauseKind::Matched,
+                predicate: Some(col("qty").gt(lit(0_i64))),
+                action: MergeIntoAction::Update(vec![(
+                    "qty".to_string(),
+                    col("source_qty"),
+                )]),
+            }],
+        });
+        assert_eq!(op.name(), "MergeInto");
+        assert_eq!(format!("{op}"), "MergeInto");
+    }
+}

datafusion/expr/src/logical_plan/mod.rs

@@ -36,7 +36,10 @@ pub use ddl::{
     CreateFunctionBody, CreateIndex, CreateMemoryTable, CreateView, DdlStatement,
     DropCatalogSchema, DropFunction, DropTable, DropView, OperateFunctionArg,
 };
-pub use dml::{DmlStatement, WriteOp};
+pub use dml::{
+    DmlStatement, MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp,
+    WriteOp,
+};
 pub use plan::{
     Aggregate, Analyze, ColumnUnnestList, DescribeTable, Distinct, DistinctOn,
     EmptyRelation, Explain, ExplainOption, Extension, FetchType, Filter, Join,

datafusion/proto/proto/datafusion.proto

@@ -299,13 +299,63 @@ message DmlNode{
     INSERT_OVERWRITE = 4;
     INSERT_REPLACE = 5;
     TRUNCATE = 6;
+    MERGE_INTO = 7;
   }
   Type dml_type = 1;
   LogicalPlanNode input = 2;
   TableReference table_name = 3;
   LogicalPlanNode target = 5;
+  // Populated only when dml_type == MERGE_INTO.
+  MergeIntoOpNode merge_into = 6;
 }
 
+// Carries the ON condition and WHEN clauses of a MERGE INTO operation.
+message MergeIntoOpNode {
+  LogicalExprNode on = 1;
+  repeated MergeIntoClauseNode clauses = 2;
+}
+
+// A single WHEN clause within a MERGE INTO statement.
+message MergeIntoClauseNode {
+  enum Kind {
+    MATCHED = 0;
+    NOT_MATCHED = 1;
+    NOT_MATCHED_BY_TARGET = 2;
+    NOT_MATCHED_BY_SOURCE = 3;
+  }
+  Kind kind = 1;
+  // Optional `AND <expr>` predicate. Absent when the clause has no predicate.
+  LogicalExprNode predicate = 2;
+  MergeIntoActionNode action = 3;
+}
+
+// The action for a single WHEN clause.
+message MergeIntoActionNode {
+  oneof action {
+    MergeUpdateAction update = 1;
+    MergeInsertAction insert = 2;
+    MergeDeleteAction delete = 3;
+  }
+}
+
+message MergeUpdateAction {
+  repeated MergeAssignment assignments = 1;
+}
+
+message MergeAssignment {
+  string column = 1;
+  LogicalExprNode value = 2;
+}
+
+message MergeInsertAction {
+  // May be empty (meaning all columns).
+  repeated string columns = 1;
+  // One expression per inserted column.
+  repeated LogicalExprNode values = 2;
+}
+
+message MergeDeleteAction {}
+
 message UnnestNode {
   LogicalPlanNode input = 1;
   repeated datafusion_common.Column exec_columns = 2;