unikdahal opened a new pull request, #5318:
URL: https://github.com/apache/datafusion-comet/pull/5318
## Which issue does this PR close?
Part of #5122.
## Rationale for this change
Row-level `MERGE` plans currently fall back to Spark entirely because
`MergeRowsExec` has no
native equivalent, even though the expensive parts of the plan (the
source/target join, filters,
projections) are operators Comet already accelerates. `MergeRowsExec` itself
is close to a
projection: it routes each joined row through matched / not-matched /
not-matched-by-source
instruction lists that are ordinary Catalyst expressions, emitting kept,
inserted, or discarded
rows. That makes it a native operator independent of native writes -- this
PR keeps the join and
merge dispatch logic in one native stage instead of falling back to Spark
for the whole statement.
## What changes are included in this PR?
- New native `MergeRowsExec` operator
(`native/core/src/execution/operators/merge_rows.rs`)
reproducing Spark's dispatch: matched / not-matched /
not-matched-by-source instruction groups,
Keep / Discard / Split semantics (0/1/2 output projections),
first-match-wins clause evaluation,
and Spark's NULL-collapses-to-false predicate semantics.
- Cardinality checking (`MERGE_CARDINALITY_VIOLATION`), mirroring Spark's
`BitmapCardinalityValidator`, surfaced as a real `SparkRuntimeException`
through the structured
`SparkError` / JNI error-conversion path -- not a generic native exception.
- New `MergeRows` / `MergeInstruction` / `MergeOutputRow` protobuf messages.
- `CometOperatorSerde` for `MergeRowsExec`, version-gated for Spark 3.5 and
4.x (the operator
doesn't exist before 3.5).
- New `spark.comet.exec.mergeRows.enabled` config, disabled by default
(experimental/opt-in).
- `output_rows` / `output_batches` / `elapsed_compute` metrics. Does not yet
expose Spark 4.x's
8 per-clause row counters
(numTargetRowsCopied/Inserted/Updated/Deleted/...) -- documented as a
known gap in `compatibility/operators.md`, since Spark 3.5.x doesn't have
those metrics either.
- Docs: operator support table entry + compatibility/caveats write-up.
## How are these changes tested?
- Rust unit tests in `merge_rows.rs`:
matched/not-matched/not-matched-by-source dispatch,
first-match-wins ordering, NULL handling, cardinality violation (including
split across batches
and NULL row ids), Split output row values, out-of-range row-id ordinal
rejection.
- `CometMergeRowsSuite` (Scala): MERGE INTO against Spark's own
`InMemoryRowLevelOperationTableCatalog` (non-Iceberg, pins the Spark-core
contract independent of
any connector), asserting native engagement and exact Comet/Spark result
parity, plus a dedicated
test asserting the cardinality violation surfaces as
`SparkRuntimeException[MERGE_CARDINALITY_VIOLATION]`,
matching Spark's own exception type and error class.
- Manually verified end-to-end against a real Iceberg REST catalog across
MATCHED / NOT MATCHED /
NOT MATCHED BY SOURCE clause combinations (update, delete, insert,
multi-clause first-match-wins,
NULL conditions, cardinality violations), confirming native engagement and
exact result/error
parity with Spark.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
