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. Mark WriteOp #[non_exhaustive] so future
variant additions are not a SemVer break for downstream matchers.

Wire MergeInto through datafusion-proto: extend the DmlNode message with
a MERGE_INTO type tag and a (boxed) MergeIntoOpNode payload field, and
add matching serializers in to_proto.rs / from_proto.rs. The conversion
goes through a single fallible helper, parse_write_op(&DmlNode, ...),
which reads the payload when the tag is MergeInto. There is no
From<dml_node::Type> for WriteOp that would have to panic when faced
with a MergeInto tag without payload, and no From<&WriteOp> for
dml_node::Type that would silently lose the payload — both directions
are handled in one place.

Document that MergeIntoClauseKind::NotMatched and NotMatchedByTarget
are semantically equivalent and must be treated identically downstream,
and provide two helpers so consumers don't have to remember:
- is_not_matched_by_target(): predicate covering both spellings
- canonical(): collapses NotMatched into NotMatchedByTarget

Tests:
- datafusion-expr unit tests for the helpers and WriteOp::MergeInto
  display
- datafusion-proto round-trip test exercising all four clause kinds and
  all three action variants
- datafusion-proto error-path tests for each defensive check in
  parse_merge_into_op/clause/action (missing on, missing payload,
  unknown clause kind tag, missing clause action, missing action oneof)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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)]
@@ -227,7 +227,11 @@ impl PartialOrd for DmlStatement {
 /// The type of DML operation to perform.
 ///
 /// See [`DmlStatement`] for more details.
+///
+/// Marked `#[non_exhaustive]` so adding new variants in future releases is
+/// not a SemVer break for downstream matchers.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+#[non_exhaustive]
 pub enum WriteOp {
     /// `INSERT INTO` operation
     Insert(InsertOp),
@@ -239,6 +243,8 @@ pub enum WriteOp {
     Ctas,
     /// `TRUNCATE` operation
     Truncate,
+    /// `MERGE INTO` operation
+    MergeInto(MergeIntoOp),
 }
 
 impl WriteOp {
@@ -250,6 +256,7 @@ impl WriteOp {
             WriteOp::Update => "Update",
             WriteOp::Ctas => "Ctas",
             WriteOp::Truncate => "Truncate",
+            WriteOp::MergeInto(_) => "MergeInto",
         }
     }
 }
@@ -291,10 +298,151 @@ 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,
+}
+
+impl MergeIntoClauseKind {
+    /// True if this clause fires on a source row that has no matching target
+    /// row. Returns `true` for both [`NotMatched`](Self::NotMatched) and
+    /// [`NotMatchedByTarget`](Self::NotMatchedByTarget) (see the type-level
+    /// note explaining why those two variants are semantically identical).
+    ///
+    /// Prefer this predicate over hand-written `matches!` arms so the
+    /// `NotMatched`/`NotMatchedByTarget` equivalence is enforced in one place.
+    pub fn is_not_matched_by_target(&self) -> bool {
+        matches!(self, Self::NotMatched | Self::NotMatchedByTarget)
+    }
+
+    /// Collapse the SQL-spelling variants into the canonical three semantic
+    /// categories: [`Matched`](Self::Matched),
+    /// [`NotMatchedByTarget`](Self::NotMatchedByTarget) (covering both
+    /// "NOT MATCHED" spellings), and
+    /// [`NotMatchedBySource`](Self::NotMatchedBySource).
+    ///
+    /// Use this in downstream `match` expressions when the SQL spelling
+    /// distinction does not matter — e.g. in planners, optimizers, or
+    /// table-provider dispatch.
+    pub fn canonical(self) -> Self {
+        match self {
+            Self::NotMatched => Self::NotMatchedByTarget,
+            other => other,
+        }
+    }
+}
+
+/// 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");
+    }
+
+    #[test]
+    fn merge_into_clause_kind_is_not_matched_by_target() {
+        assert!(!MergeIntoClauseKind::Matched.is_not_matched_by_target());
+        assert!(MergeIntoClauseKind::NotMatched.is_not_matched_by_target());
+        assert!(MergeIntoClauseKind::NotMatchedByTarget.is_not_matched_by_target());
+        assert!(!MergeIntoClauseKind::NotMatchedBySource.is_not_matched_by_target());
+    }
+
+    #[test]
+    fn merge_into_clause_kind_canonical_collapses_not_matched() {
+        assert_eq!(
+            MergeIntoClauseKind::NotMatched.canonical(),
+            MergeIntoClauseKind::NotMatchedByTarget
+        );
+        assert_eq!(
+            MergeIntoClauseKind::NotMatchedByTarget.canonical(),
+            MergeIntoClauseKind::NotMatchedByTarget
+        );
+        assert_eq!(
+            MergeIntoClauseKind::Matched.canonical(),
+            MergeIntoClauseKind::Matched
+        );
+        assert_eq!(
+            MergeIntoClauseKind::NotMatchedBySource.canonical(),
+            MergeIntoClauseKind::NotMatchedBySource
+        );
+    }
+}

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

@@ -300,13 +300,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;