Add MERGE INTO types to datafusion-expr (#20763)

## Which issue does this PR close?

- part of #20746

[EPIC] Complete DML Support (MERGE, INSERT OVERWRITE, TRUNCATE) #19617
As well as task 1 of #20746

## Rationale for this change

Lay the foundation for MERGE INTO support in DataFusion by adding the
logical plan types and their proto serialization. Keeping types separate
from execution lets reviewers reason about the data model independently
of the planner and physical dispatch.

## What changes are included in this PR?

**`datafusion/expr` — new types in `dml.rs`**
- `MergeIntoOp` — carries the `ON` join condition and ordered list of
`WHEN` clauses
- `MergeIntoClause` — a single `WHEN` clause: kind + optional predicate
+ action
- `MergeIntoClauseKind` — `Matched` / `NotMatched` /
`NotMatchedByTarget` / `NotMatchedBySource`; includes
`is_not_matched_by_target()` and `canonical()` helpers because
`NotMatched` and `NotMatchedByTarget` are semantically identical and
must be treated identically downstream
- `MergeIntoAction` — `Update(Vec<(col, expr)>)` / `Insert { columns,
values }` / `Delete`
- `WriteOp::MergeInto(MergeIntoOp)` variant added to the existing
`WriteOp` enum; `WriteOp` is now `#[non_exhaustive]` so future variant
additions are not a SemVer break

**`datafusion/proto-models` — proto schema**
- Extended `DmlNode` with a `MERGE_INTO` type tag and a boxed
`MergeIntoOpNode` payload field
- Added `MergeIntoOpNode`, `MergeIntoClauseNode`, `MergeIntoActionNode`
messages

**`datafusion/proto` — serialization**
- `from_proto`: `parse_write_op(&DmlNode, ...)` reads the payload when
the type tag is `MergeInto`; defensive helpers
`parse_merge_into_op/clause/action` with explicit errors for missing
fields
- `to_proto`: `serialize_merge_into_op/clause/action` helpers; encode
path uses an explicit `match` over all `WriteOp` variants producing
`(dml_type, merge_into)` pair — no silent payload loss
- Cross-crate conversions use `FromProto` (the crate-local trait) rather
than `From` to satisfy the Rust orphan rule after the upstream
`datafusion-proto-models` refactor

**Proto codegen** — after editing `.proto` files, regenerate with:
```bash
PROTOC=/tmp/protoc cargo run --manifest-path datafusion/proto-models/gen/Cargo.toml
```
(Install `protoc` from
https://github.com/protocolbuffers/protobuf/releases if not present; set
`PROTOC` to its path.)

## Are these changes tested?

- `datafusion-expr` unit tests: `WriteOp::MergeInto` display,
`is_not_matched_by_target`, `canonical`
- `datafusion-proto` round-trip test: exercises all four
`MergeIntoClauseKind` variants and all three `MergeIntoAction` variants
through encode → decode
- `datafusion-proto` error-path tests: missing `merge_into` payload,
missing `on` expression, unknown clause kind tag, missing clause action,
missing action oneof

## Are there any user-facing changes?

`WriteOp` gains a `MergeInto` variant and is now `#[non_exhaustive]`.
Existing downstream `match` arms need a wildcard arm added (this is
intentional and expected for a new DML operation).

## Follow-up

A stacking PR that adds the SQL planner, physical planner dispatch, and
`TableProvider::merge_into` hook is available at
wirybeaver#2. If reviewers prefer to
review both together in one pass, I'm happy to include that work here
instead.

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(Box<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(Box::new(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-models/proto/datafusion.proto

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