feat: add native support for MergeRowsExec (row-level MERGE INTO)

.github/workflows/pr_build_linux.yml

@@ -337,6 +337,7 @@ jobs:
               org.apache.comet.exec.CometAggregateSuite
               org.apache.comet.exec.CometExec3_4PlusSuite
               org.apache.comet.exec.CometExecSuite
+              org.apache.comet.exec.CometMergeRowsSuite
               org.apache.comet.exec.CometGenerateExecSuite
               org.apache.comet.exec.CometWindowExecSuite
               org.apache.comet.exec.CometJoinSuite

.github/workflows/pr_build_macos.yml

@@ -153,6 +153,7 @@ jobs:
               org.apache.comet.exec.CometAggregateSuite
               org.apache.comet.exec.CometExec3_4PlusSuite
               org.apache.comet.exec.CometExecSuite
+              org.apache.comet.exec.CometMergeRowsSuite
               org.apache.comet.exec.CometGenerateExecSuite
               org.apache.comet.exec.CometWindowExecSuite
               org.apache.comet.exec.CometJoinSuite

docs/source/user-guide/latest/compatibility/operators.md

@@ -74,6 +74,31 @@ incorrect result. When any single window expression in a `WindowExec` falls back
 `WindowGroupLimitExec` (window-based limit pushdown) is not yet supported and falls back to Spark
 ([#4837](https://github.com/apache/datafusion-comet/issues/4837)).
 
+## MERGE INTO (MergeRowsExec)
+
+Comet can run `MergeRowsExec` (Spark's row-level `MERGE INTO` dispatch operator, Spark 3.5+)
+natively, but it is disabled by default. Enable it with `spark.comet.exec.mergeRows.enabled=true`.
+
+**Missing per-clause row metrics:** Spark 4.x's `MergeRowsExec` exposes eight metrics --
+`numTargetRowsCopied`, `numTargetRowsInserted`, `numTargetRowsUpdated`, `numTargetRowsDeleted`,
+`numTargetRowsMatchedUpdated`, `numTargetRowsMatchedDeleted`, `numTargetRowsNotMatchedBySourceUpdated`,
+and `numTargetRowsNotMatchedBySourceDeleted` -- breaking down how many rows each `MERGE` clause
+touched. Comet's native operator does not expose these; it only reports the generic `output_rows`,
+`output_batches`, and `elapsed_compute` every native operator reports. EXPLAIN ANALYZE and the
+Spark UI will not show a rows-inserted/updated/deleted breakdown for a native `MERGE`. (Spark
+3.5.x's own `MergeRowsExec` does not have these metrics either -- they were added alongside a
+`Context` field Spark only attaches to `MERGE` clauses starting in 4.x.)
+
+**Cardinality-violation error may differ from Spark's on rare inputs:** Spark validates cardinality
+(rejecting an `ON` condition that matches one target row to multiple source rows,
+`MERGE_CARDINALITY_VIOLATION`) row-at-a-time, interleaved with applying each `MATCHED` clause, so
+whichever failure a given row hits first is the error Spark raises. Comet's native operator is
+vectorized: it validates cardinality for an entire input batch before evaluating any clause. If a
+single batch contains both a cardinality violation and an unrelated clause-evaluation error (for
+example an ANSI divide-by-zero) on different rows, Comet may raise a different error than Spark
+would for the same input, depending on which row each engine reaches first. The query fails either
+way; only the specific error differs.
+
 ## Round-Robin Partitioning
 
 Comet's native shuffle implementation of round-robin partitioning (`df.repartition(n)`) is not compatible with

docs/source/user-guide/latest/operators.md

@@ -116,9 +116,10 @@ omitted from the tables below and may be reconsidered based on demand:
 
 ## Writes
 
-| Operator                 | Status | Notes                                                             |
-| ------------------------ | ------ | ----------------------------------------------------------------- |
-| `DataWritingCommandExec` | ⚠️     | Experimental native Parquet writes, disabled by default (opt-in). |
+| Operator                 | Status | Notes                                                                                                                                                                               |
+| ------------------------ | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `DataWritingCommandExec` | ⚠️     | Experimental native Parquet writes, disabled by default (opt-in).                                                                                                                   |
+| `MergeRowsExec`          | ⚠️     | Row-level `MERGE INTO` dispatch (Spark 3.5+). Disabled by default; opt in with `spark.comet.exec.mergeRows.enabled=true`. See [Operator Compatibility](compatibility/operators.md). |
 
 ## Python and UDF
 

native/common/src/error.rs

@@ -184,6 +184,12 @@ pub enum SparkError {
     #[error("[SCALAR_SUBQUERY_TOO_MANY_ROWS] Scalar subquery returned more than one row.")]
     ScalarSubqueryTooManyRows,
 
+    /// Mirrors Spark's `QueryExecutionErrors.mergeCardinalityViolationError()`, raised by
+    /// `MergeRowsExec.BitmapCardinalityValidator` when a MERGE's ON condition matches a single
+    /// target row against more than one source row.
+    #[error("[MERGE_CARDINALITY_VIOLATION] The ON search condition of the MERGE statement matched a single row from the target table with multiple rows of the source table. This could result in the target row being operated on more than once with an update or delete operation and is not allowed.")]
+    MergeCardinalityViolation,
+
     #[error("{message}")]
     FileNotFound { message: String },
 
@@ -303,6 +309,7 @@ impl SparkError {
             SparkError::InvalidRegexGroupIndex { .. } => "InvalidRegexGroupIndex",
             SparkError::DatatypeCannotOrder { .. } => "DatatypeCannotOrder",
             SparkError::ScalarSubqueryTooManyRows => "ScalarSubqueryTooManyRows",
+            SparkError::MergeCardinalityViolation => "MergeCardinalityViolation",
             SparkError::FileNotFound { .. } => "FileNotFound",
             SparkError::DuplicateFieldCaseInsensitive { .. } => "DuplicateFieldCaseInsensitive",
             SparkError::DuplicateFieldByFieldId { .. } => "DuplicateFieldByFieldId",
@@ -618,7 +625,8 @@ impl SparkError {
             | SparkError::UnexpectedPositiveValue { .. }
             | SparkError::UnexpectedNegativeValue { .. }
             | SparkError::InvalidRegexGroupIndex { .. }
-            | SparkError::ScalarSubqueryTooManyRows => "org/apache/spark/SparkRuntimeException",
+            | SparkError::ScalarSubqueryTooManyRows
+            | SparkError::MergeCardinalityViolation => "org/apache/spark/SparkRuntimeException",
 
             // DateTimeException
             SparkError::InvalidInputInCastToDatetime { .. }
@@ -736,6 +744,9 @@ impl SparkError {
             // Subquery errors
             SparkError::ScalarSubqueryTooManyRows => Some("SCALAR_SUBQUERY_TOO_MANY_ROWS"),
 
+            // MERGE INTO errors
+            SparkError::MergeCardinalityViolation => Some("MERGE_CARDINALITY_VIOLATION"),
+
             // File not found
             SparkError::FileNotFound { .. } => Some("_LEGACY_ERROR_TEMP_2055"),
 

native/core/src/execution/jni_api.rs

@@ -271,6 +271,7 @@ fn op_name(op: &OpStruct) -> &'static str {
         OpStruct::ShuffleScan(_) => "ShuffleScan",
         OpStruct::BroadcastNestedLoopJoin(_) => "BroadcastNestedLoopJoin",
         OpStruct::Sample(_) => "Sample",
+        OpStruct::MergeRows(_) => "MergeRows",
     }
 }