sunchao commented on code in PR #6640:
URL: https://github.com/apache/datafusion-comet/pull/6640#discussion_r4180992529
##########
spark/src/test/spark-4.1+/org/apache/comet/exec/CometMergeRowsSuite.scala:
##########
@@ -47,63 +50,162 @@ class CometMergeRowsSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
.set("spark.sql.autoBroadcastJoinThreshold", "-1")
.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "-1")
.set("spark.sql.shuffle.partitions", "4")
+ .setMaster("local[5,2]")
}
- test("Spark 4.1+ MERGE retains Spark MergeRowsExec for write-summary
compatibility") {
- val target = s"$catalog.default.rowlevel_target"
- val source = s"$catalog.default.rowlevel_source"
-
- sql(s"DROP TABLE IF EXISTS $target")
- sql(s"DROP TABLE IF EXISTS $source")
- sql(s"CREATE TABLE $target (id INT, amount DOUBLE) USING parquet")
- sql(s"CREATE TABLE $source (id INT, amount DOUBLE) USING parquet")
- sql(s"INSERT INTO $target VALUES (1, 10.0), (2, 20.0)")
- sql(s"INSERT INTO $source VALUES (2, 200.0), (3, 300.0)")
-
- val captured = ArrayBuffer[QueryExecution]()
- val listener = new QueryExecutionListener {
- override def onSuccess(funcName: String, qe: QueryExecution, durationNs:
Long): Unit =
- captured += qe
- override def onFailure(funcName: String, qe: QueryExecution, exception:
Exception): Unit =
- ()
+ test("native MergeRows semantic counters match Spark for Keep, Discard and
Split") {
+ withTempPath { path =>
+ spark
+ .range(1, 9)
+ .selectExpr(
+ "CAST(id AS INT) AS id",
+ "id <= 6 AS source_present",
+ "id != 6 AS target_present")
+ .repartition(2)
+ .write
+ .parquet(path.getCanonicalPath)
+ val input =
spark.read.parquet(path.getCanonicalPath).queryExecution.analyzed
+ val id = input.output(0)
+ def is(value: Int) = EqualTo(id, Literal(value))
+ def plan = MergeRows(
+ input.output(1),
+ input.output(2),
+ Seq(
+ MergeRows.Keep(MergeRows.Copy, is(1), Seq(id)),
+ MergeRows.Discard(is(2)),
+ MergeRows.Keep(MergeRows.Update, is(3), Seq(id)),
+ MergeRows.Split(is(4), Seq(id), Seq(id)),
+ MergeRows.Keep(MergeRows.Delete, is(5), Seq(id))),
+ Seq(MergeRows.Keep(MergeRows.Insert, Literal(true), Seq(id))),
+ Seq(MergeRows.Keep(MergeRows.Update, is(7), Seq(id)),
MergeRows.Discard(is(8))),
+ checkCardinality = false,
+ output = Seq(id),
+ child = input)
+ val expected = Map(
+ "numTargetRowsCopied" -> 1L,
+ "numTargetRowsDeleted" -> 3L,
+ "numTargetRowsUpdated" -> 3L,
+ "numTargetRowsInserted" -> 1L,
+ "numTargetRowsMatchedUpdated" -> 2L,
+ "numTargetRowsMatchedDeleted" -> 2L,
+ "numTargetRowsNotMatchedBySourceUpdated" -> 1L,
+ "numTargetRowsNotMatchedBySourceDeleted" -> 1L)
+
+ Seq(true, false).foreach { adaptiveEnabled =>
+ withSQLConf("spark.sql.adaptive.enabled" -> adaptiveEnabled.toString) {
+ val baseline = withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ val df = datasetOfRows(spark,
ShimCometMergeRows.withNativeMergeSummary(plan))
+ val rows = df.collect().map(_.toString).sorted.toSeq
+ val merge =
find(df.queryExecution.executedPlan)(_.isInstanceOf[MergeRowsExec]).get
+ assert(expected.map { case (key, _) =>
+ key -> MergeRowsMetricsShim.value(merge.metrics(key))
+ } == expected)
+ rows
+ }
+ withSQLConf(CometConf.COMET_EXEC_MERGE_ROWS_ENABLED.key -> "true") {
+ val df = datasetOfRows(spark,
ShimCometMergeRows.withNativeMergeSummary(plan))
+ assert(df.collect().map(_.toString).sorted.toSeq == baseline)
+ val merge =
find(df.queryExecution.executedPlan)(_.isInstanceOf[CometMergeRowsExec])
+ .getOrElse(
+ fail(s"native MergeRows did not engage:
${df.queryExecution.executedPlan}"))
+ assert(expected.map { case (key, _) =>
+ key -> MergeRowsMetricsShim.value(merge.metrics(key))
+ } == expected)
+ }
+ }
+ }
+ }
+ }
+
+ test("Spark 4.2 semantic metrics retain only successful task attempt
updates") {
+ assume(isSpark42Plus, "last-attempt MERGE metrics require Spark 4.2")
+ val metrics = MergeRowsMetricsShim.metrics(spark.sparkContext)
+ val node = CometMetricNode(metrics, Seq.empty)
+ val names = metrics.keys.toSeq
+ def runAttempt(): Unit = {
+ val attempts = spark.sparkContext
+ .parallelize(0 until 4, 4)
+ .mapPartitions { input =>
+ val count = input.size.toLong
+ val attempt = TaskContext.get().attemptNumber()
+ if (attempt == 0) {
+ names.foreach(node.set(_, 100L))
+ throw new IllegalStateException("injected failed MERGE metric
attempt")
+ }
+ // JNI publishes cumulative snapshots, so repeated sets must not add
the old value.
+ names.foreach(node.set(_, count + 10L))
Review Comment:
[P2] Use monotonic cumulative snapshots in the retry test. With Spark 4.2,
each successful partition has `count == 1`, but these calls report 11 followed
by 1. `CometMetricNode.set` adds only positive deltas and retains the highest
reported value, so each successful attempt contributes 11. The summary is
therefore 44, and the assertion expecting 4 fails before the second RDD
executes. This breaks the registered Spark 4.2 suite. Please repeat `count`, or
use increasing snapshots ending at `count`, while retaining the failed-attempt
injection.
Evidence: A bounded standalone Spark 4.2.0 reproduction compiled the PR’s
`MergeRowsMetricsShim` and used the exact `CometMetricNode.set` method with the
new test’s retry/RDD body. It failed with `numTargetRowsCopied actual=44
expected=4` and `AssertionError: incorrect last-attempt numTargetRowsCopied`.
Changing only the first successful-attempt snapshot from `count + 10L` to
`count` made all eight counters equal 4 for both RDD executions. The full
ScalaTest suite was not run.
--
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]
