andygrove commented on code in PR #5880:
URL: https://github.com/apache/datafusion-comet/pull/5880#discussion_r3997383875
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -3738,6 +3738,81 @@ class CometIcebergNativeSuite
}
}
+ test("Iceberg native scan left unconsumed by a limit still reports task
input metrics") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ spark.sql("""
+ CREATE TABLE test_cat.db.task_metrics_limit_test (
+ id INT,
+ value DOUBLE
+ ) USING iceberg
+ """)
+ spark
+ .range(20000)
+ .selectExpr("CAST(id AS INT)", "CAST(id * 1.5 AS DOUBLE) as value")
+ .repartition(4)
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable("test_cat.db.task_metrics_limit_test")
+
+ val bytesReadValues = mutable.ArrayBuffer.empty[Long]
+ val recordsReadValues = mutable.ArrayBuffer.empty[Long]
+ val listener = new SparkListener {
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ val im = taskEnd.taskMetrics.inputMetrics
+ bytesReadValues.synchronized {
+ bytesReadValues += im.bytesRead
+ recordsReadValues += im.recordsRead
+ }
+ }
+ }
+ spark.sparkContext.addSparkListener(listener)
+
+ try {
+ // The limit stops pulling before the scan is exhausted, so the
final metric publish
Review Comment:
Nit: I ran the ordering revert against both suites to see what this test
buys. It passes with the ordering reverted, as you said it would, and so does
`native scan block left unconsumed by a limit ...`. Only the broadcast-join
test fails. The reason is in `jni_api.rs`: a block with no JVM input is spawned
onto the `batch_receiver` channel and calls `update_metrics` on every batch,
while a block that has one takes the busy-poll `stream` path where
`update_metrics` only fires on the update interval. So having a JVM input is
what decides whether the ordering matters, not stopping early.
Both tests earn their keep as coverage for their sites. Could this comment
say what the test actually does though? Right now it reads "the final metric
publish happens in the iterator's completion-time close", which is the
broadcast-join test's mechanism rather than this one. Something like: this
Iceberg scan is its own block with no JVM input, so metrics publish per batch
and this covers the site rather than the order, which stays unguarded until
#5265.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala:
##########
@@ -105,24 +101,34 @@ case class CometMetricNode(metrics: Map[String,
SQLMetric], children: Seq[CometM
})
/**
- * Reports aggregated scan input metrics (bytesRead, recordsRead) to Spark's
task metrics.
- * Aggregates across all scan leaf nodes to handle plans with multiple scans
(e.g., joins). Must
- * be called in a TaskCompletionListener after the iterator is fully
consumed.
+ * Reports the scan leaves' bytes and rows (summed across joins and unions)
to Spark's task
+ * input metrics, which drive the Input column on the UI's Stages and
Executors tabs.
+ *
+ * Must be registered on the task thread before
[[org.apache.comet.CometExecIterator]] so its
+ * completion listener publishes final SQL metrics before this listener
runs. A block with a JVM
+ * input only publishes on the metrics update interval, and a consumer that
stops early, such as
+ * a limit, leaves the final publish to that close.
+ *
+ * Adds to the task's counters instead of replacing them, so bytes that a
fallback Spark scan
Review Comment:
Nit: the reversed-arm test and the description bullet cover this now, but
the scaladoc here still has the unconditional version, "so bytes that a
fallback Spark scan accumulated in the same task survive". That is the claim
the reversed arm disproves, and it is the sentence someone reading this file
will find rather than the PR description.
Could you carry a short version of the caveat up here? Something like: this
survives only when the fallback scan registers its listener after Comet's,
which a `CometSparkToColumnarExec` input always does but a coalesced Spark-scan
partition computed first does not.
--
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]
