comphead commented on code in PR #6696:
URL: https://github.com/apache/datafusion-comet/pull/6696#discussion_r4488371532
##########
spark/src/test/spark-4.1+/org/apache/comet/exec/CometMergeRowsSuite.scala:
##########
@@ -53,6 +56,62 @@ class CometMergeRowsSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
.setMaster("local[5,2]")
}
+ test("summary-free insert-only MergeRows requires Spark 4.2") {
+ val input = spark.range(3)
+ val child = input.queryExecution.sparkPlan
+ val id = child.output.head
+ val instructions = Seq(
+ MergeRows.Keep(MergeRows.Insert, EqualTo(id, Literal(1L)), Seq(id)),
+ MergeRows.Keep(MergeRows.Insert, Literal(true), Seq(id)))
+ val op = MergeRowsExec(
+ Literal(true),
+ Literal(false),
+ Seq.empty,
+ instructions,
+ Seq.empty,
+ checkCardinality = false,
+ output = Seq(id),
+ child = child)
+
+ assert(ShimCometMergeRows.canRunWithoutMergeSummary(op) == isSpark42Plus)
Review Comment:
This builds the same hand-made two-instruction `MergeRowsExec` as the
near-miss and schema tests in `CometInsertOnlyMergeSuite`, and its 4.2 branch
repeats what the end-to-end 4.2 tests show. Could it keep only the 4.1 negative
case, the `isSpark42Plus` gate and the `withNativeMergeSummary` check, with the
fixture plan in one helper shared by both suites?
##########
spark/src/test/spark-4.2/org/apache/comet/exec/CometInsertOnlyMergeSuite.scala:
##########
@@ -0,0 +1,433 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.exec
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{CometListenerBusUtils, SparkConf}
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal}
+import org.apache.spark.sql.catalyst.plans.logical.MergeRows
+import org.apache.spark.sql.comet.CometMergeRowsExec
+import org.apache.spark.sql.connector.catalog.{Identifier,
InMemoryRowLevelOperationTableCatalog, InMemoryTable}
+import org.apache.spark.sql.connector.write.MergeSummary
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.v2.{InsertOnlyMergeExec,
MergeRowsExec}
+import org.apache.spark.sql.util.QueryExecutionListener
+
+import org.apache.comet.{CometConf, CometExplainInfo}
+import org.apache.comet.serde.OperatorOuterClass.Operator
+import org.apache.comet.serde.Unsupported
+import org.apache.comet.serde.operator.CometMergeRows
+import org.apache.comet.shims.ShimCometMergeRows
+
+/** Spark 4.2 coverage for the dedicated InsertOnlyMergeExec rewrite. */
+class CometInsertOnlyMergeSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
+
+ private val catalog = "insert_only_merge"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set(s"spark.sql.catalog.$catalog",
classOf[InMemoryRowLevelOperationTableCatalog].getName)
+ .set("spark.sql.autoBroadcastJoinThreshold", "-1")
+ .set("spark.sql.adaptive.autoBroadcastJoinThreshold", "-1")
+ .set("spark.sql.shuffle.partitions", "4")
+ }
+
+ private case class MergeResult(plans: Seq[SparkPlan], rows: Seq[String],
summary: MergeSummary)
+
+ private def resetTables(target: String, source: String, sourceRows: String):
Unit = {
+ sql(s"DROP TABLE IF EXISTS $catalog.default.$target")
+ sql(s"DROP TABLE IF EXISTS $catalog.default.$source")
+ sql(s"CREATE TABLE $catalog.default.$target (id INT, amount INT) USING
parquet")
+ sql(s"CREATE TABLE $catalog.default.$source (id INT, amount INT) USING
parquet")
+ sql(s"INSERT INTO $catalog.default.$target VALUES (1, 10), (2, 20)")
+ sql(s"INSERT INTO $catalog.default.$source VALUES $sourceRows")
+ }
+
+ private def lastMergeSummary(table: String): MergeSummary = {
+ val cat = spark.sessionState.catalogManager
+ .catalog(catalog)
+ .asInstanceOf[InMemoryRowLevelOperationTableCatalog]
+ cat
+ .loadTable(Identifier.of(Array("default"), table))
+ .asInstanceOf[InMemoryTable]
+ .commits
+ .last
+ .writeSummary
+ .get
+ .asInstanceOf[MergeSummary]
+ }
+
+ private def runMerge(
+ target: String,
+ mergeSql: String,
+ cometEnabled: Boolean,
+ adaptiveEnabled: Boolean): MergeResult = {
+ 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 =
+ ()
+ }
+
+ spark.listenerManager.register(listener)
+ try {
+ val configs =
+ Seq(
+ CometConf.COMET_EXEC_MERGE_ROWS_ENABLED.key -> "true",
+ "spark.sql.adaptive.enabled" -> adaptiveEnabled.toString,
+ "spark.sql.ansi.enabled" -> "true") ++
+ (if (cometEnabled) Seq.empty[(String, String)]
+ else Seq(CometConf.COMET_ENABLED.key -> "false"))
+ withSQLConf(configs: _*) {
+ sql(mergeSql)
+ }
+ CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
+ } finally {
+ spark.listenerManager.unregister(listener)
+ }
+
+ val rows = sql(s"SELECT id, amount FROM $catalog.default.$target ORDER BY
id, amount")
+ .collect()
+ .map(_.toString)
+ .toSeq
+ MergeResult(captured.map(_.executedPlan).toSeq, rows,
lastMergeSummary(target))
+ }
+
+ private def hasInsertOnlyMerge(plans: Seq[SparkPlan]): Boolean =
+ plans.exists(plan =>
+ find(plan) { case _: InsertOnlyMergeExec => true; case _ => false
}.nonEmpty)
+
+ private def hasCometMergeRows(plans: Seq[SparkPlan]): Boolean =
+ plans.exists(plan =>
+ find(plan) { case _: CometMergeRowsExec => true; case _ => false
}.nonEmpty)
+
+ private def hasSparkMergeRows(plans: Seq[SparkPlan]): Boolean =
+ plans.exists(plan => find(plan) { case _: MergeRowsExec => true; case _ =>
false }.nonEmpty)
+
+ private def assertInsertOnlySummary(summary: MergeSummary, inserted: Long):
Unit = {
+ assert(summary.numTargetRowsInserted() == inserted)
+ assert(summary.numTargetRowsCopied() == 0)
+ assert(summary.numTargetRowsUpdated() == 0)
+ assert(summary.numTargetRowsDeleted() == 0)
+ assert(summary.numTargetRowsMatchedUpdated() == 0)
+ assert(summary.numTargetRowsMatchedDeleted() == 0)
+ assert(summary.numTargetRowsNotMatchedBySourceUpdated() == 0)
+ assert(summary.numTargetRowsNotMatchedBySourceDeleted() == 0)
+ }
+
+ test("multiple NOT MATCHED clauses preserve parity with AQE on and off") {
+ val sourceRows = "(2, 200), (3, 300), (3, 301), (4, 400)"
Review Comment:
Every unmatched source row here matches a clause and all values are
non-NULL, so two shapes specific to this rewrite are not exercised end to end:
an unmatched row that matches no clause (dropped, not counted as an insert) and
a NULL merge key (never matches in the `LeftAnti` join, so it is inserted).
Could `sourceRows` gain rows such as `(NULL, 500)` and `(5, NULL)`? With
`s.amount < 350` and `s.amount >= 350`, `(5, NULL)` makes both predicates NULL
and should be dropped.
--
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]
