You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/16 17:38:41 UTC
(datafusion) branch gh-readonly-queue/main/pr-20763-d5f03d9f9396c140d5954f08622878b1a40c4951 created (now 408dad3dc5)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-20763-d5f03d9f9396c140d5954f08622878b1a40c4951
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 408dad3dc5 Add MERGE INTO types to datafusion-expr (#20763)
This branch includes the following new commits:
new 408dad3dc5 Add MERGE INTO types to datafusion-expr (#20763)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: Add MERGE INTO types to datafusion-expr (#20763)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-20763-d5f03d9f9396c140d5954f08622878b1a40c4951
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 408dad3dc5733f1b438cf5515af3d8c0da4b178c
Author: Xuanyi Li <xu...@uber.com>
AuthorDate: Tue Jun 16 10:38:18 2026 -0700
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
https://github.com/wirybeaver/datafusion/pull/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) <no...@anthropic.com>
---
datafusion/expr/src/logical_plan/dml.rs | 150 +++-
datafusion/expr/src/logical_plan/mod.rs | 5 +-
datafusion/proto-models/proto/datafusion.proto | 50 ++
datafusion/proto-models/src/generated/pbjson.rs | 833 +++++++++++++++++++++
datafusion/proto-models/src/generated/prost.rs | 111 +++
datafusion/proto/src/logical_plan/from_proto.rs | 137 +++-
datafusion/proto/src/logical_plan/mod.rs | 34 +-
datafusion/proto/src/logical_plan/to_proto.rs | 97 ++-
.../proto/tests/cases/roundtrip_logical_plan.rs | 187 ++++-
9 files changed, 1571 insertions(+), 33 deletions(-)
diff --git a/datafusion/expr/src/logical_plan/dml.rs b/datafusion/expr/src/logical_plan/dml.rs
index b668cbfe2c..5b6403e6e2 100644
--- a/datafusion/expr/src/logical_plan/dml.rs
+++ b/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,6 +298,96 @@ 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)])
@@ -298,3 +395,54 @@ fn make_count_schema() -> DFSchemaRef {
.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
+ );
+ }
+}
diff --git a/datafusion/expr/src/logical_plan/mod.rs b/datafusion/expr/src/logical_plan/mod.rs
index e0e51d7e47..4766c3f333 100644
--- a/datafusion/expr/src/logical_plan/mod.rs
+++ b/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,
diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto
index 322395ab37..f13494cf43 100644
--- a/datafusion/proto-models/proto/datafusion.proto
+++ b/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;
diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs
index 1eb9de00fb..2ba5e25054 100644
--- a/datafusion/proto-models/src/generated/pbjson.rs
+++ b/datafusion/proto-models/src/generated/pbjson.rs
@@ -5487,6 +5487,9 @@ impl serde::Serialize for DmlNode {
if self.target.is_some() {
len += 1;
}
+ if self.merge_into.is_some() {
+ len += 1;
+ }
let mut struct_ser = serializer.serialize_struct("datafusion.DmlNode", len)?;
if self.dml_type != 0 {
let v = dml_node::Type::try_from(self.dml_type)
@@ -5502,6 +5505,9 @@ impl serde::Serialize for DmlNode {
if let Some(v) = self.target.as_ref() {
struct_ser.serialize_field("target", v)?;
}
+ if let Some(v) = self.merge_into.as_ref() {
+ struct_ser.serialize_field("mergeInto", v)?;
+ }
struct_ser.end()
}
}
@@ -5518,6 +5524,8 @@ impl<'de> serde::Deserialize<'de> for DmlNode {
"table_name",
"tableName",
"target",
+ "merge_into",
+ "mergeInto",
];
#[allow(clippy::enum_variant_names)]
@@ -5526,6 +5534,7 @@ impl<'de> serde::Deserialize<'de> for DmlNode {
Input,
TableName,
Target,
+ MergeInto,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -5551,6 +5560,7 @@ impl<'de> serde::Deserialize<'de> for DmlNode {
"input" => Ok(GeneratedField::Input),
"tableName" | "table_name" => Ok(GeneratedField::TableName),
"target" => Ok(GeneratedField::Target),
+ "mergeInto" | "merge_into" => Ok(GeneratedField::MergeInto),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
}
}
@@ -5574,6 +5584,7 @@ impl<'de> serde::Deserialize<'de> for DmlNode {
let mut input__ = None;
let mut table_name__ = None;
let mut target__ = None;
+ let mut merge_into__ = None;
while let Some(k) = map_.next_key()? {
match k {
GeneratedField::DmlType => {
@@ -5600,6 +5611,12 @@ impl<'de> serde::Deserialize<'de> for DmlNode {
}
target__ = map_.next_value()?;
}
+ GeneratedField::MergeInto => {
+ if merge_into__.is_some() {
+ return Err(serde::de::Error::duplicate_field("mergeInto"));
+ }
+ merge_into__ = map_.next_value()?;
+ }
}
}
Ok(DmlNode {
@@ -5607,6 +5624,7 @@ impl<'de> serde::Deserialize<'de> for DmlNode {
input: input__,
table_name: table_name__,
target: target__,
+ merge_into: merge_into__,
})
}
}
@@ -5627,6 +5645,7 @@ impl serde::Serialize for dml_node::Type {
Self::InsertOverwrite => "INSERT_OVERWRITE",
Self::InsertReplace => "INSERT_REPLACE",
Self::Truncate => "TRUNCATE",
+ Self::MergeInto => "MERGE_INTO",
};
serializer.serialize_str(variant)
}
@@ -5645,6 +5664,7 @@ impl<'de> serde::Deserialize<'de> for dml_node::Type {
"INSERT_OVERWRITE",
"INSERT_REPLACE",
"TRUNCATE",
+ "MERGE_INTO",
];
struct GeneratedVisitor;
@@ -5692,6 +5712,7 @@ impl<'de> serde::Deserialize<'de> for dml_node::Type {
"INSERT_OVERWRITE" => Ok(dml_node::Type::InsertOverwrite),
"INSERT_REPLACE" => Ok(dml_node::Type::InsertReplace),
"TRUNCATE" => Ok(dml_node::Type::Truncate),
+ "MERGE_INTO" => Ok(dml_node::Type::MergeInto),
_ => Err(serde::de::Error::unknown_variant(value, FIELDS)),
}
}
@@ -14033,6 +14054,818 @@ impl<'de> serde::Deserialize<'de> for MemoryScanExecNode {
deserializer.deserialize_struct("datafusion.MemoryScanExecNode", FIELDS, GeneratedVisitor)
}
}
+impl serde::Serialize for MergeAssignment {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if !self.column.is_empty() {
+ len += 1;
+ }
+ if self.value.is_some() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.MergeAssignment", len)?;
+ if !self.column.is_empty() {
+ struct_ser.serialize_field("column", &self.column)?;
+ }
+ if let Some(v) = self.value.as_ref() {
+ struct_ser.serialize_field("value", v)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeAssignment {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "column",
+ "value",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Column,
+ Value,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "column" => Ok(GeneratedField::Column),
+ "value" => Ok(GeneratedField::Value),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeAssignment;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeAssignment")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeAssignment, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut column__ = None;
+ let mut value__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Column => {
+ if column__.is_some() {
+ return Err(serde::de::Error::duplicate_field("column"));
+ }
+ column__ = Some(map_.next_value()?);
+ }
+ GeneratedField::Value => {
+ if value__.is_some() {
+ return Err(serde::de::Error::duplicate_field("value"));
+ }
+ value__ = map_.next_value()?;
+ }
+ }
+ }
+ Ok(MergeAssignment {
+ column: column__.unwrap_or_default(),
+ value: value__,
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeAssignment", FIELDS, GeneratedVisitor)
+ }
+}
+impl serde::Serialize for MergeDeleteAction {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let len = 0;
+ let struct_ser = serializer.serialize_struct("datafusion.MergeDeleteAction", len)?;
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeDeleteAction {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ Err(serde::de::Error::unknown_field(value, FIELDS))
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeDeleteAction;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeDeleteAction")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeDeleteAction, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ while map_.next_key::<GeneratedField>()?.is_some() {
+ let _ = map_.next_value::<serde::de::IgnoredAny>()?;
+ }
+ Ok(MergeDeleteAction {
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeDeleteAction", FIELDS, GeneratedVisitor)
+ }
+}
+impl serde::Serialize for MergeInsertAction {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if !self.columns.is_empty() {
+ len += 1;
+ }
+ if !self.values.is_empty() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.MergeInsertAction", len)?;
+ if !self.columns.is_empty() {
+ struct_ser.serialize_field("columns", &self.columns)?;
+ }
+ if !self.values.is_empty() {
+ struct_ser.serialize_field("values", &self.values)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeInsertAction {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "columns",
+ "values",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Columns,
+ Values,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "columns" => Ok(GeneratedField::Columns),
+ "values" => Ok(GeneratedField::Values),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeInsertAction;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeInsertAction")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeInsertAction, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut columns__ = None;
+ let mut values__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Columns => {
+ if columns__.is_some() {
+ return Err(serde::de::Error::duplicate_field("columns"));
+ }
+ columns__ = Some(map_.next_value()?);
+ }
+ GeneratedField::Values => {
+ if values__.is_some() {
+ return Err(serde::de::Error::duplicate_field("values"));
+ }
+ values__ = Some(map_.next_value()?);
+ }
+ }
+ }
+ Ok(MergeInsertAction {
+ columns: columns__.unwrap_or_default(),
+ values: values__.unwrap_or_default(),
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeInsertAction", FIELDS, GeneratedVisitor)
+ }
+}
+impl serde::Serialize for MergeIntoActionNode {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if self.action.is_some() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.MergeIntoActionNode", len)?;
+ if let Some(v) = self.action.as_ref() {
+ match v {
+ merge_into_action_node::Action::Update(v) => {
+ struct_ser.serialize_field("update", v)?;
+ }
+ merge_into_action_node::Action::Insert(v) => {
+ struct_ser.serialize_field("insert", v)?;
+ }
+ merge_into_action_node::Action::Delete(v) => {
+ struct_ser.serialize_field("delete", v)?;
+ }
+ }
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeIntoActionNode {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "update",
+ "insert",
+ "delete",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Update,
+ Insert,
+ Delete,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "update" => Ok(GeneratedField::Update),
+ "insert" => Ok(GeneratedField::Insert),
+ "delete" => Ok(GeneratedField::Delete),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeIntoActionNode;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeIntoActionNode")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeIntoActionNode, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut action__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Update => {
+ if action__.is_some() {
+ return Err(serde::de::Error::duplicate_field("update"));
+ }
+ action__ = map_.next_value::<::std::option::Option<_>>()?.map(merge_into_action_node::Action::Update)
+;
+ }
+ GeneratedField::Insert => {
+ if action__.is_some() {
+ return Err(serde::de::Error::duplicate_field("insert"));
+ }
+ action__ = map_.next_value::<::std::option::Option<_>>()?.map(merge_into_action_node::Action::Insert)
+;
+ }
+ GeneratedField::Delete => {
+ if action__.is_some() {
+ return Err(serde::de::Error::duplicate_field("delete"));
+ }
+ action__ = map_.next_value::<::std::option::Option<_>>()?.map(merge_into_action_node::Action::Delete)
+;
+ }
+ }
+ }
+ Ok(MergeIntoActionNode {
+ action: action__,
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeIntoActionNode", FIELDS, GeneratedVisitor)
+ }
+}
+impl serde::Serialize for MergeIntoClauseNode {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if self.kind != 0 {
+ len += 1;
+ }
+ if self.predicate.is_some() {
+ len += 1;
+ }
+ if self.action.is_some() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.MergeIntoClauseNode", len)?;
+ if self.kind != 0 {
+ let v = merge_into_clause_node::Kind::try_from(self.kind)
+ .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.kind)))?;
+ struct_ser.serialize_field("kind", &v)?;
+ }
+ if let Some(v) = self.predicate.as_ref() {
+ struct_ser.serialize_field("predicate", v)?;
+ }
+ if let Some(v) = self.action.as_ref() {
+ struct_ser.serialize_field("action", v)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeIntoClauseNode {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "kind",
+ "predicate",
+ "action",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Kind,
+ Predicate,
+ Action,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "kind" => Ok(GeneratedField::Kind),
+ "predicate" => Ok(GeneratedField::Predicate),
+ "action" => Ok(GeneratedField::Action),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeIntoClauseNode;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeIntoClauseNode")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeIntoClauseNode, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut kind__ = None;
+ let mut predicate__ = None;
+ let mut action__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Kind => {
+ if kind__.is_some() {
+ return Err(serde::de::Error::duplicate_field("kind"));
+ }
+ kind__ = Some(map_.next_value::<merge_into_clause_node::Kind>()? as i32);
+ }
+ GeneratedField::Predicate => {
+ if predicate__.is_some() {
+ return Err(serde::de::Error::duplicate_field("predicate"));
+ }
+ predicate__ = map_.next_value()?;
+ }
+ GeneratedField::Action => {
+ if action__.is_some() {
+ return Err(serde::de::Error::duplicate_field("action"));
+ }
+ action__ = map_.next_value()?;
+ }
+ }
+ }
+ Ok(MergeIntoClauseNode {
+ kind: kind__.unwrap_or_default(),
+ predicate: predicate__,
+ action: action__,
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeIntoClauseNode", FIELDS, GeneratedVisitor)
+ }
+}
+impl serde::Serialize for merge_into_clause_node::Kind {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ let variant = match self {
+ Self::Matched => "MATCHED",
+ Self::NotMatched => "NOT_MATCHED",
+ Self::NotMatchedByTarget => "NOT_MATCHED_BY_TARGET",
+ Self::NotMatchedBySource => "NOT_MATCHED_BY_SOURCE",
+ };
+ serializer.serialize_str(variant)
+ }
+}
+impl<'de> serde::Deserialize<'de> for merge_into_clause_node::Kind {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "MATCHED",
+ "NOT_MATCHED",
+ "NOT_MATCHED_BY_TARGET",
+ "NOT_MATCHED_BY_SOURCE",
+ ];
+
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = merge_into_clause_node::Kind;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E>
+ where
+ E: serde::de::Error,
+ {
+ i32::try_from(v)
+ .ok()
+ .and_then(|x| x.try_into().ok())
+ .ok_or_else(|| {
+ serde::de::Error::invalid_value(serde::de::Unexpected::Signed(v), &self)
+ })
+ }
+
+ fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E>
+ where
+ E: serde::de::Error,
+ {
+ i32::try_from(v)
+ .ok()
+ .and_then(|x| x.try_into().ok())
+ .ok_or_else(|| {
+ serde::de::Error::invalid_value(serde::de::Unexpected::Unsigned(v), &self)
+ })
+ }
+
+ fn visit_str<E>(self, value: &str) -> std::result::Result<Self::Value, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "MATCHED" => Ok(merge_into_clause_node::Kind::Matched),
+ "NOT_MATCHED" => Ok(merge_into_clause_node::Kind::NotMatched),
+ "NOT_MATCHED_BY_TARGET" => Ok(merge_into_clause_node::Kind::NotMatchedByTarget),
+ "NOT_MATCHED_BY_SOURCE" => Ok(merge_into_clause_node::Kind::NotMatchedBySource),
+ _ => Err(serde::de::Error::unknown_variant(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_any(GeneratedVisitor)
+ }
+}
+impl serde::Serialize for MergeIntoOpNode {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if self.on.is_some() {
+ len += 1;
+ }
+ if !self.clauses.is_empty() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.MergeIntoOpNode", len)?;
+ if let Some(v) = self.on.as_ref() {
+ struct_ser.serialize_field("on", v)?;
+ }
+ if !self.clauses.is_empty() {
+ struct_ser.serialize_field("clauses", &self.clauses)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeIntoOpNode {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "on",
+ "clauses",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ On,
+ Clauses,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "on" => Ok(GeneratedField::On),
+ "clauses" => Ok(GeneratedField::Clauses),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeIntoOpNode;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeIntoOpNode")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeIntoOpNode, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut on__ = None;
+ let mut clauses__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::On => {
+ if on__.is_some() {
+ return Err(serde::de::Error::duplicate_field("on"));
+ }
+ on__ = map_.next_value()?;
+ }
+ GeneratedField::Clauses => {
+ if clauses__.is_some() {
+ return Err(serde::de::Error::duplicate_field("clauses"));
+ }
+ clauses__ = Some(map_.next_value()?);
+ }
+ }
+ }
+ Ok(MergeIntoOpNode {
+ on: on__,
+ clauses: clauses__.unwrap_or_default(),
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeIntoOpNode", FIELDS, GeneratedVisitor)
+ }
+}
+impl serde::Serialize for MergeUpdateAction {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if !self.assignments.is_empty() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.MergeUpdateAction", len)?;
+ if !self.assignments.is_empty() {
+ struct_ser.serialize_field("assignments", &self.assignments)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for MergeUpdateAction {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "assignments",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Assignments,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "assignments" => Ok(GeneratedField::Assignments),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = MergeUpdateAction;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.MergeUpdateAction")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<MergeUpdateAction, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut assignments__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Assignments => {
+ if assignments__.is_some() {
+ return Err(serde::de::Error::duplicate_field("assignments"));
+ }
+ assignments__ = Some(map_.next_value()?);
+ }
+ }
+ }
+ Ok(MergeUpdateAction {
+ assignments: assignments__.unwrap_or_default(),
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.MergeUpdateAction", FIELDS, GeneratedVisitor)
+ }
+}
impl serde::Serialize for NamedStructField {
#[allow(deprecated)]
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs
index 3ac04a6164..e8fa4599e1 100644
--- a/datafusion/proto-models/src/generated/prost.rs
+++ b/datafusion/proto-models/src/generated/prost.rs
@@ -453,6 +453,9 @@ pub struct DmlNode {
pub table_name: ::core::option::Option<TableReference>,
#[prost(message, optional, boxed, tag = "5")]
pub target: ::core::option::Option<::prost::alloc::boxed::Box<LogicalPlanNode>>,
+ /// Populated only when dml_type == MERGE_INTO.
+ #[prost(message, optional, boxed, tag = "6")]
+ pub merge_into: ::core::option::Option<::prost::alloc::boxed::Box<MergeIntoOpNode>>,
}
/// Nested message and enum types in `DmlNode`.
pub mod dml_node {
@@ -476,6 +479,7 @@ pub mod dml_node {
InsertOverwrite = 4,
InsertReplace = 5,
Truncate = 6,
+ MergeInto = 7,
}
impl Type {
/// String value of the enum field names used in the ProtoBuf definition.
@@ -491,6 +495,7 @@ pub mod dml_node {
Self::InsertOverwrite => "INSERT_OVERWRITE",
Self::InsertReplace => "INSERT_REPLACE",
Self::Truncate => "TRUNCATE",
+ Self::MergeInto => "MERGE_INTO",
}
}
/// Creates an enum from field names used in the ProtoBuf definition.
@@ -503,11 +508,117 @@ pub mod dml_node {
"INSERT_OVERWRITE" => Some(Self::InsertOverwrite),
"INSERT_REPLACE" => Some(Self::InsertReplace),
"TRUNCATE" => Some(Self::Truncate),
+ "MERGE_INTO" => Some(Self::MergeInto),
_ => None,
}
}
}
}
+/// Carries the ON condition and WHEN clauses of a MERGE INTO operation.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MergeIntoOpNode {
+ #[prost(message, optional, boxed, tag = "1")]
+ pub on: ::core::option::Option<::prost::alloc::boxed::Box<LogicalExprNode>>,
+ #[prost(message, repeated, tag = "2")]
+ pub clauses: ::prost::alloc::vec::Vec<MergeIntoClauseNode>,
+}
+/// A single WHEN clause within a MERGE INTO statement.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MergeIntoClauseNode {
+ #[prost(enumeration = "merge_into_clause_node::Kind", tag = "1")]
+ pub kind: i32,
+ /// Optional `AND <expr>` predicate. Absent when the clause has no predicate.
+ #[prost(message, optional, tag = "2")]
+ pub predicate: ::core::option::Option<LogicalExprNode>,
+ #[prost(message, optional, tag = "3")]
+ pub action: ::core::option::Option<MergeIntoActionNode>,
+}
+/// Nested message and enum types in `MergeIntoClauseNode`.
+pub mod merge_into_clause_node {
+ #[derive(
+ Clone,
+ Copy,
+ Debug,
+ PartialEq,
+ Eq,
+ Hash,
+ PartialOrd,
+ Ord,
+ ::prost::Enumeration
+ )]
+ #[repr(i32)]
+ pub enum Kind {
+ Matched = 0,
+ NotMatched = 1,
+ NotMatchedByTarget = 2,
+ NotMatchedBySource = 3,
+ }
+ impl Kind {
+ /// String value of the enum field names used in the ProtoBuf definition.
+ ///
+ /// The values are not transformed in any way and thus are considered stable
+ /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+ pub fn as_str_name(&self) -> &'static str {
+ match self {
+ Self::Matched => "MATCHED",
+ Self::NotMatched => "NOT_MATCHED",
+ Self::NotMatchedByTarget => "NOT_MATCHED_BY_TARGET",
+ Self::NotMatchedBySource => "NOT_MATCHED_BY_SOURCE",
+ }
+ }
+ /// Creates an enum from field names used in the ProtoBuf definition.
+ pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+ match value {
+ "MATCHED" => Some(Self::Matched),
+ "NOT_MATCHED" => Some(Self::NotMatched),
+ "NOT_MATCHED_BY_TARGET" => Some(Self::NotMatchedByTarget),
+ "NOT_MATCHED_BY_SOURCE" => Some(Self::NotMatchedBySource),
+ _ => None,
+ }
+ }
+ }
+}
+/// The action for a single WHEN clause.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MergeIntoActionNode {
+ #[prost(oneof = "merge_into_action_node::Action", tags = "1, 2, 3")]
+ pub action: ::core::option::Option<merge_into_action_node::Action>,
+}
+/// Nested message and enum types in `MergeIntoActionNode`.
+pub mod merge_into_action_node {
+ #[derive(Clone, PartialEq, ::prost::Oneof)]
+ pub enum Action {
+ #[prost(message, tag = "1")]
+ Update(super::MergeUpdateAction),
+ #[prost(message, tag = "2")]
+ Insert(super::MergeInsertAction),
+ #[prost(message, tag = "3")]
+ Delete(super::MergeDeleteAction),
+ }
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MergeUpdateAction {
+ #[prost(message, repeated, tag = "1")]
+ pub assignments: ::prost::alloc::vec::Vec<MergeAssignment>,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MergeAssignment {
+ #[prost(string, tag = "1")]
+ pub column: ::prost::alloc::string::String,
+ #[prost(message, optional, tag = "2")]
+ pub value: ::core::option::Option<LogicalExprNode>,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MergeInsertAction {
+ /// May be empty (meaning all columns).
+ #[prost(string, repeated, tag = "1")]
+ pub columns: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
+ /// One expression per inserted column.
+ #[prost(message, repeated, tag = "2")]
+ pub values: ::prost::alloc::vec::Vec<LogicalExprNode>,
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct MergeDeleteAction {}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct UnnestNode {
#[prost(message, optional, boxed, tag = "1")]
diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs
index c68b83964f..b79b21b359 100644
--- a/datafusion/proto/src/logical_plan/from_proto.rs
+++ b/datafusion/proto/src/logical_plan/from_proto.rs
@@ -25,7 +25,9 @@ use datafusion_common::{
};
use datafusion_execution::TaskContext;
use datafusion_execution::registry::FunctionRegistry;
-use datafusion_expr::dml::InsertOp;
+use datafusion_expr::dml::{
+ InsertOp, MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp,
+};
use datafusion_expr::expr::{Alias, NullTreatment, Placeholder, Sort};
use datafusion_expr::expr::{Unnest, WildcardOptions};
use datafusion_expr::logical_plan::Subquery;
@@ -239,22 +241,133 @@ impl FromProto<protobuf::NullEquality> for NullEquality {
}
}
-impl FromProto<protobuf::dml_node::Type> for WriteOp {
- fn from_proto(t: protobuf::dml_node::Type) -> Self {
- match t {
- protobuf::dml_node::Type::Update => WriteOp::Update,
- protobuf::dml_node::Type::Delete => WriteOp::Delete,
- protobuf::dml_node::Type::InsertAppend => WriteOp::Insert(InsertOp::Append),
- protobuf::dml_node::Type::InsertOverwrite => {
- WriteOp::Insert(InsertOp::Overwrite)
+impl FromProto<protobuf::merge_into_clause_node::Kind> for MergeIntoClauseKind {
+ fn from_proto(k: protobuf::merge_into_clause_node::Kind) -> Self {
+ match k {
+ protobuf::merge_into_clause_node::Kind::Matched => {
+ MergeIntoClauseKind::Matched
+ }
+ protobuf::merge_into_clause_node::Kind::NotMatched => {
+ MergeIntoClauseKind::NotMatched
+ }
+ protobuf::merge_into_clause_node::Kind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ protobuf::merge_into_clause_node::Kind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
}
- protobuf::dml_node::Type::InsertReplace => WriteOp::Insert(InsertOp::Replace),
- protobuf::dml_node::Type::Ctas => WriteOp::Ctas,
- protobuf::dml_node::Type::Truncate => WriteOp::Truncate,
}
}
}
+/// Reconstruct a [`WriteOp`] from a [`protobuf::DmlNode`], reading the
+/// `merge_into` payload when the type tag is `MergeInto`.
+pub fn parse_write_op(
+ node: &protobuf::DmlNode,
+ ctx: &TaskContext,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<WriteOp, Error> {
+ let typ = node.dml_type();
+ Ok(match typ {
+ protobuf::dml_node::Type::Update => WriteOp::Update,
+ protobuf::dml_node::Type::Delete => WriteOp::Delete,
+ protobuf::dml_node::Type::InsertAppend => WriteOp::Insert(InsertOp::Append),
+ protobuf::dml_node::Type::InsertOverwrite => WriteOp::Insert(InsertOp::Overwrite),
+ protobuf::dml_node::Type::InsertReplace => WriteOp::Insert(InsertOp::Replace),
+ protobuf::dml_node::Type::Ctas => WriteOp::Ctas,
+ protobuf::dml_node::Type::Truncate => WriteOp::Truncate,
+ protobuf::dml_node::Type::MergeInto => {
+ let merge_into = node.merge_into.as_deref().ok_or_else(|| {
+ Error::General(
+ "DmlNode with MERGE_INTO type is missing the merge_into payload"
+ .to_string(),
+ )
+ })?;
+ WriteOp::MergeInto(Box::new(parse_merge_into_op(merge_into, ctx, codec)?))
+ }
+ })
+}
+
+fn parse_merge_into_op(
+ op: &protobuf::MergeIntoOpNode,
+ ctx: &TaskContext,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<MergeIntoOp, Error> {
+ let on = op.on.as_ref().ok_or_else(|| {
+ Error::General("MergeIntoOpNode is missing required `on` expression".to_string())
+ })?;
+ let on = parse_expr(on, ctx, codec)?;
+ let clauses = op
+ .clauses
+ .iter()
+ .map(|c| parse_merge_into_clause(c, ctx, codec))
+ .collect::<Result<Vec<_>, Error>>()?;
+ Ok(MergeIntoOp { on, clauses })
+}
+
+fn parse_merge_into_clause(
+ clause: &protobuf::MergeIntoClauseNode,
+ ctx: &TaskContext,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<MergeIntoClause, Error> {
+ let kind = protobuf::merge_into_clause_node::Kind::try_from(clause.kind)
+ .map_err(|_| {
+ Error::General(format!(
+ "MergeIntoClauseNode has unknown kind tag {}",
+ clause.kind
+ ))
+ })
+ .map(MergeIntoClauseKind::from_proto)?;
+ let predicate = clause
+ .predicate
+ .as_ref()
+ .map(|e| parse_expr(e, ctx, codec))
+ .transpose()?;
+ let action = clause.action.as_ref().ok_or_else(|| {
+ Error::General("MergeIntoClauseNode is missing required `action`".to_string())
+ })?;
+ let action = parse_merge_into_action(action, ctx, codec)?;
+ Ok(MergeIntoClause {
+ kind,
+ predicate,
+ action,
+ })
+}
+
+fn parse_merge_into_action(
+ action: &protobuf::MergeIntoActionNode,
+ ctx: &TaskContext,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<MergeIntoAction, Error> {
+ use protobuf::merge_into_action_node::Action;
+ let action = action.action.as_ref().ok_or_else(|| {
+ Error::General("MergeIntoActionNode is missing the `action` oneof".to_string())
+ })?;
+ Ok(match action {
+ Action::Update(update) => {
+ let assignments = update
+ .assignments
+ .iter()
+ .map(|a| {
+ let value = a.value.as_ref().ok_or_else(|| {
+ Error::General(format!(
+ "MergeAssignment for column `{}` is missing its value",
+ a.column
+ ))
+ })?;
+ Ok((a.column.clone(), parse_expr(value, ctx, codec)?))
+ })
+ .collect::<Result<Vec<_>, Error>>()?;
+ MergeIntoAction::Update(assignments)
+ }
+ Action::Insert(insert) => MergeIntoAction::Insert {
+ columns: insert.columns.clone(),
+ values: parse_exprs(&insert.values, ctx, codec)?,
+ },
+ Action::Delete(_) => MergeIntoAction::Delete,
+ })
+}
+
impl FromProto<protobuf::NullTreatment> for NullTreatment {
fn from_proto(t: protobuf::NullTreatment) -> Self {
match t {
diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs
index 35c2e76d88..0e73898c67 100644
--- a/datafusion/proto/src/logical_plan/mod.rs
+++ b/datafusion/proto/src/logical_plan/mod.rs
@@ -58,6 +58,7 @@ use datafusion_datasource_json::file_format::{
};
#[cfg(feature = "parquet")]
use datafusion_datasource_parquet::file_format::{ParquetFormat, ParquetFormatFactory};
+use datafusion_expr::dml::InsertOp;
use datafusion_expr::{
AggregateUDF, DmlStatement, FetchType, HigherOrderUDF, RecursiveQuery, SkipType,
TableSource, Unnest, WriteOp,
@@ -1248,10 +1249,12 @@ impl AsLogicalPlan for LogicalPlanNode {
.build()
}
LogicalPlanType::Dml(dml_node) => {
+ let write_op =
+ from_proto::parse_write_op(dml_node, ctx, extension_codec)?;
Ok(LogicalPlan::Dml(datafusion_expr::DmlStatement::new(
from_table_reference(dml_node.table_name.as_ref(), "DML ")?,
to_table_source(&dml_node.target, ctx, extension_codec)?,
- WriteOp::from_proto(dml_node.dml_type()),
+ write_op,
Arc::new(into_logical_plan!(dml_node.input, ctx, extension_codec)?),
)))
}
@@ -2085,7 +2088,33 @@ impl AsLogicalPlan for LogicalPlanNode {
}) => {
let input =
LogicalPlanNode::try_from_logical_plan(input, extension_codec)?;
- let dml_type = dml_node::Type::from_proto(op);
+ let (dml_type, merge_into) = match op {
+ WriteOp::Insert(InsertOp::Append) => {
+ (dml_node::Type::InsertAppend, None)
+ }
+ WriteOp::Insert(InsertOp::Overwrite) => {
+ (dml_node::Type::InsertOverwrite, None)
+ }
+ WriteOp::Insert(InsertOp::Replace) => {
+ (dml_node::Type::InsertReplace, None)
+ }
+ WriteOp::Delete => (dml_node::Type::Delete, None),
+ WriteOp::Update => (dml_node::Type::Update, None),
+ WriteOp::Ctas => (dml_node::Type::Ctas, None),
+ WriteOp::Truncate => (dml_node::Type::Truncate, None),
+ WriteOp::MergeInto(merge_op) => (
+ dml_node::Type::MergeInto,
+ Some(Box::new(to_proto::serialize_merge_into_op(
+ merge_op,
+ extension_codec,
+ )?)),
+ ),
+ other => {
+ return Err(proto_error(format!(
+ "WriteOp variant has no DmlNode encoding: {other}"
+ )));
+ }
+ };
Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::Dml(Box::new(DmlNode {
input: Some(Box::new(input)),
@@ -2098,6 +2127,7 @@ impl AsLogicalPlan for LogicalPlanNode {
table_name.clone(),
)),
dml_type: dml_type.into(),
+ merge_into,
}))),
})
}
diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs
index 71a6bd824a..516aca4094 100644
--- a/datafusion/proto/src/logical_plan/to_proto.rs
+++ b/datafusion/proto/src/logical_plan/to_proto.rs
@@ -22,8 +22,9 @@
use std::collections::HashMap;
use datafusion_common::{NullEquality, TableReference, UnnestOptions};
-use datafusion_expr::WriteOp;
-use datafusion_expr::dml::InsertOp;
+use datafusion_expr::dml::{
+ MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp,
+};
use datafusion_expr::expr::{
self, AggregateFunctionParams, Alias, Between, BinaryExpr, Cast, GroupingSet, InList,
Like, NullTreatment, Placeholder, ScalarFunction, Unnest,
@@ -753,22 +754,92 @@ impl FromProto<NullEquality> for protobuf::NullEquality {
}
}
-impl FromProto<&WriteOp> for protobuf::dml_node::Type {
- fn from_proto(t: &WriteOp) -> Self {
- match t {
- WriteOp::Insert(InsertOp::Append) => protobuf::dml_node::Type::InsertAppend,
- WriteOp::Insert(InsertOp::Overwrite) => {
- protobuf::dml_node::Type::InsertOverwrite
+impl FromProto<MergeIntoClauseKind> for protobuf::merge_into_clause_node::Kind {
+ fn from_proto(k: MergeIntoClauseKind) -> Self {
+ match k {
+ MergeIntoClauseKind::Matched => {
+ protobuf::merge_into_clause_node::Kind::Matched
+ }
+ MergeIntoClauseKind::NotMatched => {
+ protobuf::merge_into_clause_node::Kind::NotMatched
+ }
+ MergeIntoClauseKind::NotMatchedByTarget => {
+ protobuf::merge_into_clause_node::Kind::NotMatchedByTarget
+ }
+ MergeIntoClauseKind::NotMatchedBySource => {
+ protobuf::merge_into_clause_node::Kind::NotMatchedBySource
}
- WriteOp::Insert(InsertOp::Replace) => protobuf::dml_node::Type::InsertReplace,
- WriteOp::Delete => protobuf::dml_node::Type::Delete,
- WriteOp::Update => protobuf::dml_node::Type::Update,
- WriteOp::Ctas => protobuf::dml_node::Type::Ctas,
- WriteOp::Truncate => protobuf::dml_node::Type::Truncate,
}
}
}
+pub fn serialize_merge_into_op(
+ op: &MergeIntoOp,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<protobuf::MergeIntoOpNode, Error> {
+ Ok(protobuf::MergeIntoOpNode {
+ on: Some(Box::new(serialize_expr(&op.on, codec)?)),
+ clauses: op
+ .clauses
+ .iter()
+ .map(|c| serialize_merge_into_clause(c, codec))
+ .collect::<Result<Vec<_>, Error>>()?,
+ })
+}
+
+fn serialize_merge_into_clause(
+ clause: &MergeIntoClause,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<protobuf::MergeIntoClauseNode, Error> {
+ let kind = protobuf::merge_into_clause_node::Kind::from_proto(clause.kind);
+ let predicate = clause
+ .predicate
+ .as_ref()
+ .map(|e| serialize_expr(e, codec))
+ .transpose()?;
+ Ok(protobuf::MergeIntoClauseNode {
+ kind: kind.into(),
+ predicate,
+ action: Some(serialize_merge_into_action(&clause.action, codec)?),
+ })
+}
+
+fn serialize_merge_into_action(
+ action: &MergeIntoAction,
+ codec: &dyn LogicalExtensionCodec,
+) -> Result<protobuf::MergeIntoActionNode, Error> {
+ let action = match action {
+ MergeIntoAction::Update(assignments) => {
+ let assignments = assignments
+ .iter()
+ .map(|(column, value)| {
+ Ok(protobuf::MergeAssignment {
+ column: column.clone(),
+ value: Some(serialize_expr(value, codec)?),
+ })
+ })
+ .collect::<Result<Vec<_>, Error>>()?;
+ protobuf::merge_into_action_node::Action::Update(
+ protobuf::MergeUpdateAction { assignments },
+ )
+ }
+ MergeIntoAction::Insert { columns, values } => {
+ protobuf::merge_into_action_node::Action::Insert(
+ protobuf::MergeInsertAction {
+ columns: columns.clone(),
+ values: serialize_exprs(values, codec)?,
+ },
+ )
+ }
+ MergeIntoAction::Delete => protobuf::merge_into_action_node::Action::Delete(
+ protobuf::MergeDeleteAction {},
+ ),
+ };
+ Ok(protobuf::MergeIntoActionNode {
+ action: Some(action),
+ })
+}
+
impl FromProto<NullTreatment> for protobuf::NullTreatment {
fn from_proto(t: NullTreatment) -> Self {
match t {
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index 7f1d0a666f..9d8e5c2b1e 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -78,6 +78,9 @@ use datafusion_common::{
};
use datafusion_execution::TaskContext;
use datafusion_expr::dml::CopyTo;
+use datafusion_expr::dml::{
+ MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp,
+};
use datafusion_expr::expr::{
self, Between, BinaryExpr, Case, Cast, GroupingSet, InList, Like, NullTreatment,
ScalarFunction, Unnest, WildcardOptions,
@@ -86,10 +89,11 @@ use datafusion_expr::logical_plan::{
ExplainOption, Extension, UserDefinedLogicalNodeCore,
};
use datafusion_expr::{
- Accumulator, AggregateUDF, ColumnarValue, ExprFunctionExt, ExprSchemable,
- LimitEffect, Literal, LogicalPlan, LogicalPlanBuilder, Operator, PartitionEvaluator,
- ScalarUDF, Signature, TryCast, Volatility, WindowFrame, WindowFrameBound,
- WindowFrameUnits, WindowFunctionDefinition, WindowUDF, WindowUDFImpl,
+ Accumulator, AggregateUDF, ColumnarValue, DmlStatement, ExprFunctionExt,
+ ExprSchemable, LimitEffect, Literal, LogicalPlan, LogicalPlanBuilder, Operator,
+ PartitionEvaluator, ScalarUDF, Signature, TryCast, Volatility, WindowFrame,
+ WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition, WindowUDF,
+ WindowUDFImpl, WriteOp,
};
use datafusion_functions_aggregate::average::avg_udaf;
use datafusion_functions_aggregate::expr_fn::{
@@ -524,6 +528,181 @@ async fn roundtrip_logical_plan_dml() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn roundtrip_logical_plan_dml_merge_into() -> Result<()> {
+ let ctx = SessionContext::new();
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Int64, true),
+ Field::new("b", DataType::Decimal128(15, 2), true),
+ ]);
+ ctx.register_csv(
+ "t1",
+ "tests/testdata/test.csv",
+ CsvReadOptions::default().schema(&schema),
+ )
+ .await?;
+
+ let scan = ctx.table("t1").await?.into_optimized_plan()?;
+ let target = match &scan {
+ LogicalPlan::TableScan(t) => Arc::clone(&t.source),
+ other => panic!("expected TableScan, got {other:?}"),
+ };
+
+ let merge = WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: col("a").eq(lit(1_i64)),
+ clauses: vec![
+ MergeIntoClause {
+ kind: MergeIntoClauseKind::Matched,
+ predicate: Some(col("b").gt(lit(ScalarValue::Decimal128(
+ Some(0),
+ 15,
+ 2,
+ )))),
+ action: MergeIntoAction::Update(vec![("b".to_string(), col("b"))]),
+ },
+ MergeIntoClause {
+ kind: MergeIntoClauseKind::NotMatched,
+ predicate: None,
+ action: MergeIntoAction::Insert {
+ columns: vec!["a".to_string(), "b".to_string()],
+ values: vec![col("a"), col("b")],
+ },
+ },
+ MergeIntoClause {
+ kind: MergeIntoClauseKind::NotMatchedByTarget,
+ predicate: None,
+ action: MergeIntoAction::Insert {
+ columns: vec![],
+ values: vec![col("a"), col("b")],
+ },
+ },
+ MergeIntoClause {
+ kind: MergeIntoClauseKind::NotMatchedBySource,
+ predicate: Some(col("a").eq(lit(2_i64))),
+ action: MergeIntoAction::Delete,
+ },
+ ],
+ }));
+
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ "t1".into(),
+ target,
+ merge,
+ Arc::new(scan),
+ ));
+
+ let bytes = logical_plan_to_bytes(&plan)?;
+ let round_trip = logical_plan_from_bytes(&bytes, &ctx.task_ctx())?;
+ assert_eq!(format!("{plan}"), format!("{round_trip}"));
+ Ok(())
+}
+
+#[test]
+fn parse_write_op_merge_into_without_payload_errors() {
+ let ctx = SessionContext::new();
+ let codec = DefaultLogicalExtensionCodec {};
+ let node = protobuf::DmlNode {
+ dml_type: protobuf::dml_node::Type::MergeInto.into(),
+ ..Default::default()
+ };
+ let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec)
+ .expect_err("MergeInto tag without payload must fail");
+ assert!(
+ err.to_string().contains("merge_into"),
+ "unexpected error: {err}"
+ );
+}
+
+/// Build a `DmlNode` whose `merge_into` payload is exactly the supplied
+/// `MergeIntoOpNode`. Used by the error-path tests below.
+fn dml_node_with_merge_payload(payload: protobuf::MergeIntoOpNode) -> protobuf::DmlNode {
+ protobuf::DmlNode {
+ dml_type: protobuf::dml_node::Type::MergeInto.into(),
+ merge_into: Some(Box::new(payload)),
+ ..Default::default()
+ }
+}
+
+#[test]
+fn parse_merge_into_op_missing_on_errors() {
+ let ctx = SessionContext::new();
+ let codec = DefaultLogicalExtensionCodec {};
+ let node = dml_node_with_merge_payload(protobuf::MergeIntoOpNode {
+ on: None,
+ clauses: vec![],
+ });
+ let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec)
+ .expect_err("missing `on` must fail");
+ assert!(err.to_string().contains("`on`"), "unexpected error: {err}");
+}
+
+#[test]
+fn parse_merge_into_clause_unknown_kind_errors() {
+ let ctx = SessionContext::new();
+ let codec = DefaultLogicalExtensionCodec {};
+ let on = serialize_expr(&lit(true), &codec).unwrap();
+ let node = dml_node_with_merge_payload(protobuf::MergeIntoOpNode {
+ on: Some(Box::new(on)),
+ clauses: vec![protobuf::MergeIntoClauseNode {
+ kind: 999, // unknown enum tag
+ predicate: None,
+ action: Some(protobuf::MergeIntoActionNode {
+ action: Some(protobuf::merge_into_action_node::Action::Delete(
+ protobuf::MergeDeleteAction {},
+ )),
+ }),
+ }],
+ });
+ let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec)
+ .expect_err("unknown clause kind tag must fail");
+ assert!(
+ err.to_string().contains("unknown kind tag"),
+ "unexpected error: {err}"
+ );
+}
+
+#[test]
+fn parse_merge_into_clause_missing_action_errors() {
+ let ctx = SessionContext::new();
+ let codec = DefaultLogicalExtensionCodec {};
+ let on = serialize_expr(&lit(true), &codec).unwrap();
+ let node = dml_node_with_merge_payload(protobuf::MergeIntoOpNode {
+ on: Some(Box::new(on)),
+ clauses: vec![protobuf::MergeIntoClauseNode {
+ kind: protobuf::merge_into_clause_node::Kind::Matched.into(),
+ predicate: None,
+ action: None,
+ }],
+ });
+ let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec)
+ .expect_err("missing clause `action` must fail");
+ assert!(
+ err.to_string().contains("missing required `action`"),
+ "unexpected error: {err}"
+ );
+}
+
+#[test]
+fn parse_merge_into_action_missing_oneof_errors() {
+ let ctx = SessionContext::new();
+ let codec = DefaultLogicalExtensionCodec {};
+ let on = serialize_expr(&lit(true), &codec).unwrap();
+ let node = dml_node_with_merge_payload(protobuf::MergeIntoOpNode {
+ on: Some(Box::new(on)),
+ clauses: vec![protobuf::MergeIntoClauseNode {
+ kind: protobuf::merge_into_clause_node::Kind::Matched.into(),
+ predicate: None,
+ action: Some(protobuf::MergeIntoActionNode { action: None }),
+ }],
+ });
+ let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec)
+ .expect_err("missing action oneof must fail");
+ assert!(
+ err.to_string().contains("missing the `action` oneof"),
+ "unexpected error: {err}"
+ );
+}
+
#[tokio::test]
async fn roundtrip_logical_plan_copy_to_sql_options() -> Result<()> {
let ctx = SessionContext::new();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org