feat: add basic MemTable MERGE INTO support

datafusion/core/tests/sql/sql_api.rs

@@ -266,7 +266,8 @@ async fn merge_into_rejects_subqueries_correlated_to_target_alias() {
     .await;
 
     // Source-correlated and uncorrelated subqueries remain supported through
-    // logical optimization.
+    // logical optimization, but basic MemTable MERGE execution does not yet
+    // evaluate subquery predicates.
     for sql in [
         "MERGE INTO target AS t USING source AS s \
          ON EXISTS (SELECT 1 FROM source AS x WHERE x.id = s.id) \
@@ -275,8 +276,12 @@ async fn merge_into_rejects_subqueries_correlated_to_target_alias() {
          ON t.id = ANY (SELECT id FROM source) \
          WHEN MATCHED THEN DELETE",
     ] {
-        assert_merge_physical_error(&ctx, sql, "MERGE INTO not supported for Base table")
-            .await;
+        assert_merge_physical_error(
+            &ctx,
+            sql,
+            "Physical plan does not support logical expression",
+        )
+        .await;
     }
 }
 
@@ -294,14 +299,19 @@ async fn merge_into_requires_boolean_conditions() {
              WHEN MATCHED AND 1 THEN DELETE",
             "MERGE WHEN condition must be boolean type, but got Int64",
         ),
-        (
-            "MERGE INTO target USING source ON NULL \
-             WHEN MATCHED AND NULL THEN DELETE",
-            "MERGE INTO not supported for Base table",
-        ),
     ] {
         assert_merge_physical_error(&ctx, sql, expected).await;
     }
+
+    ctx.sql(
+        "MERGE INTO target USING source ON NULL \
+         WHEN MATCHED AND NULL THEN DELETE",
+    )
+    .await
+    .unwrap()
+    .create_physical_plan()
+    .await
+    .unwrap();
 }
 
 #[tokio::test]

datafusion/expr/src/logical_plan/display.rs

@@ -24,7 +24,7 @@ use crate::{
     Aggregate, DescribeTable, Distinct, DistinctOn, DmlStatement, Expr, Filter, Join,
     Limit, LogicalPlan, Partitioning, Projection, RecursiveQuery, Repartition, Sort,
     Subquery, SubqueryAlias, TableProviderFilterPushDown, TableScan, Unnest, Values,
-    Window, expr_vec_fmt,
+    Window, WriteOp, expr_vec_fmt,
 };
 
 use crate::dml::CopyTo;
@@ -404,11 +404,22 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
                 })
             }
             LogicalPlan::Dml(DmlStatement { table_name, op, .. }) => {
-                json!({
-                    "Node Type": "Projection",
+                let mut object = json!({
+                    "Node Type": "Dml",
                     "Operation": op.name(),
                     "Table Name": table_name.table()
-                })
+                });
+                if let WriteOp::MergeInto(merge_op) = op {
+                    object["On"] = serde_json::Value::String(merge_op.on.to_string());
+                    object["Clauses"] = serde_json::Value::Array(
+                        merge_op
+                            .clauses
+                            .iter()
+                            .map(|clause| serde_json::Value::String(clause.to_string()))
+                            .collect(),
+                    );
+                }
+                object
             }
             LogicalPlan::Copy(CopyTo {
                 input: _,

datafusion/expr/src/logical_plan/dml.rs

@@ -407,6 +407,23 @@ impl MergeIntoOp {
     }
 }
 
+impl Display for MergeIntoOp {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "on=[{}]", self.on)?;
+        if !self.clauses.is_empty() {
+            write!(f, " clauses=[")?;
+            for (i, clause) in self.clauses.iter().enumerate() {
+                if i > 0 {
+                    write!(f, ", ")?;
+                }
+                write!(f, "{clause}")?;
+            }
+            write!(f, "]")?;
+        }
+        Ok(())
+    }
+}
+
 /// A single WHEN clause within a MERGE INTO statement.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
 pub struct MergeIntoClause {
@@ -418,6 +435,16 @@ pub struct MergeIntoClause {
     pub action: MergeIntoAction,
 }
 
+impl Display for MergeIntoClause {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "WHEN {}", self.kind)?;
+        if let Some(predicate) = &self.predicate {
+            write!(f, " AND {predicate}")?;
+        }
+        write!(f, " THEN {}", self.action)
+    }
+}
+
 /// Which rows a MERGE WHEN clause applies to.
 ///
 /// Mirrors `sqlparser::ast::MergeClauseKind` so that the SQL spelling is
@@ -444,6 +471,17 @@ pub enum MergeIntoClauseKind {
     NotMatchedBySource,
 }
 
+impl Display for MergeIntoClauseKind {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Matched => write!(f, "MATCHED"),
+            Self::NotMatched => write!(f, "NOT MATCHED"),
+            Self::NotMatchedByTarget => write!(f, "NOT MATCHED BY TARGET"),
+            Self::NotMatchedBySource => write!(f, "NOT MATCHED BY SOURCE"),
+        }
+    }
+}
+
 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
@@ -488,6 +526,38 @@ pub enum MergeIntoAction {
     Delete,
 }
 
+impl Display for MergeIntoAction {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Update(assignments) => {
+                write!(f, "UPDATE SET ")?;
+                for (i, (column, value)) in assignments.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{column} = {value}")?;
+                }
+                Ok(())
+            }
+            Self::Insert { columns, values } => {
+                write!(f, "INSERT")?;
+                if !columns.is_empty() {
+                    write!(f, " ({})", columns.join(", "))?;
+                }
+                write!(f, " VALUES (")?;
+                for (i, value) in values.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{value}")?;
+                }
+                write!(f, ")")
+            }
+            Self::Delete => write!(f, "DELETE"),
+        }
+    }
+}
+
 fn make_count_schema() -> DFSchemaRef {
     Arc::new(
         Schema::new(vec![Field::new("count", DataType::UInt64, false)])
@@ -516,6 +586,13 @@ mod tests {
         }));
         assert_eq!(op.name(), "MergeInto");
         assert_eq!(format!("{op}"), "MergeInto");
+        let WriteOp::MergeInto(merge_op) = &op else {
+            unreachable!("constructed as MergeInto")
+        };
+        assert_eq!(
+            merge_op.to_string(),
+            "on=[id = source_id] clauses=[WHEN MATCHED AND qty > Int64(0) THEN UPDATE SET qty = source_qty]"
+        );
     }
 
     #[test]

datafusion/expr/src/logical_plan/plan.rs

@@ -2027,7 +2027,11 @@ impl LogicalPlan {
                         Ok(())
                     }
                     LogicalPlan::Dml(DmlStatement { table_name, op, .. }) => {
-                        write!(f, "Dml: op=[{op}] table=[{table_name}]")
+                        write!(f, "Dml: op=[{op}] table=[{table_name}]")?;
+                        if let WriteOp::MergeInto(merge_op) = op {
+                            write!(f, " {merge_op}")?;
+                        }
+                        Ok(())
                     }
                     LogicalPlan::Copy(CopyTo {
                         input: _,

datafusion/sql/tests/sql_integration.rs

@@ -3632,6 +3632,20 @@ fn plan_merge_into_canonicalizes_qualifiers_and_preserves_quoted_columns() {
     };
     assert_eq!(columns, &["id".to_string(), "Age".to_string()]);
     assert_eq!(values[0].to_string(), "s.j2_id");
+
+    let display = plan.display_indent().to_string();
+    assert_contains!(
+        &display,
+        "Dml: op=[MergeInto] table=[person_quoted_cols] on=[person_quoted_cols.id = s.j2_id] clauses=[WHEN MATCHED THEN UPDATE SET First Name = s.j2_string, WHEN NOT MATCHED THEN INSERT (id, Age) VALUES (s.j2_id, Int64(42))]"
+    );
+
+    let json = plan.display_pg_json().to_string();
+    assert_contains!(&json, r#""Node Type": "Dml""#);
+    assert_contains!(&json, r#""On": "person_quoted_cols.id = s.j2_id""#);
+    assert_contains!(
+        &json,
+        r#""WHEN MATCHED THEN UPDATE SET First Name = s.j2_string""#
+    );
 }
 
 #[rstest]