andygrove commented on code in PR #5318:
URL: https://github.com/apache/datafusion-comet/pull/5318#discussion_r4109346060
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -774,16 +774,9 @@ class CometIcebergWriteActionSuite
}
}
- test("native acceleration: ReplaceData (CoW MERGE) falls back (MergeRowsExec
not Comet)") {
- // TODO(comet-merge-rows): native MERGE engagement requires a Comet
equivalent of Iceberg's
- // `MergeRowsExec` (the per-row dispatch operator that assigns
__row_operation codes from
- // MATCHED/NOT MATCHED clauses). Without it, `MergeRowsExec` stays JVM,
the upstream chain
- // breaks Comet-native partway, and `requiresNativeChildren=true` declines
the
- // `IcebergWriteExec -> CometIcebergWriteExec` conversion. Until that
lands, MERGE
- // falls back to the JVM two-op path -- this test pins that contract.
Native `MergeRowsExec`
- // is being added in https://github.com/apache/datafusion-comet/pull/5318;
when that lands
- // this test will start failing and needs to flip to
`assertNativeWriteEngages`.
+ test("native acceleration: ReplaceData (CoW MERGE) runs MergeRows and writer
natively") {
assumeNativeAcceleration()
+ assume(isSpark35Plus && !isSpark41Plus, "native MergeRows requires Spark
3.5 through 4.0")
Review Comment:
This `assume` cancels the test on Spark 4.1, which is the profile the PR and
merge-queue jobs run. It also replaces the test that pinned the fallback on
every profile, and deletes `assertNativeWriteDoesNotEngage`. On 4.1+
`MergeRowsExec` stays on the JVM, so that contract still holds there. Could the
test branch on `isSpark41Plus` and keep asserting on 4.1+ that this MERGE
commits once without engaging `CometIcebergWriteExec`? The comment at line 867
("Unlike the unpartitioned MERGE above...") also needs updating.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -793,16 +786,47 @@ class CometIcebergWriteActionSuite
withSQLConf(CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key ->
"false") {
coalesceInsert("native_cow_merge", Seq((1, "us-east", 10.0), (2,
"us-west", 20.0)))
}
- assertNativeWriteDoesNotEngage("native_cow_merge", Seq(1, 2, 3)) {
- spark.sql("""
- |MERGE INTO cat.db.native_cow_merge t
- |USING (SELECT 2 AS id, 'us-west' AS region, 200.0 AS amount UNION
ALL
- | SELECT 3 AS id, 'eu' AS region, 30.0 AS amount) s
- |ON t.id = s.id
- |WHEN MATCHED THEN UPDATE SET t.amount = s.amount
- |WHEN NOT MATCHED THEN INSERT (id, region, amount) VALUES (s.id,
s.region, s.amount)
- |""".stripMargin)
+
+ var snapshot: Option[WriteSnapshot] = None
Review Comment:
This is the only Iceberg MERGE that runs with the flag on: a two-row
unpartitioned copy-on-write table, checked against hand-written ids. The
description lists NOT MATCHED BY SOURCE, deletes, multiple clauses, NULL
conditions and cardinality as validated against Iceberg, but none of those
shapes are in the suite. Could it compare against a sibling table merged with
Comet off, over a partitioned table large enough to span several tasks, and add
a merge-on-read variant? I ran that matrix locally on 3.5 and 4.0 and it
matched, so this is about keeping it covered.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -257,6 +257,8 @@ object CometConf extends ShimCometConf {
createExecEnabledConfig("localTableScan", defaultValue = false)
val COMET_EXEC_SAMPLE_ENABLED: ConfigEntry[Boolean] =
createExecEnabledConfig("sample", defaultValue = true)
+ val COMET_EXEC_MERGE_ROWS_ENABLED: ConfigEntry[Boolean] =
+ createExecEnabledConfig("mergeRows", defaultValue = false)
Review Comment:
`operators.md` calls itself the complete reference for how Comet handles
each Spark operator, but it has no row for `MergeRowsExec`. Could you add one,
plus a `CometMergeRows` row in `understanding-comet-plans.md`?
`iceberg-writes.md` around line 143 should also say that a copy-on-write MERGE
into an unpartitioned table only reaches the native writer with
`spark.comet.exec.mergeRows.enabled=true`. And since this flag is ignored on
Spark 4.1+, could it pass `notes` saying so, so `configs.md` tells users?
--
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]
