dwsmith1983 commented on code in PR #5880:
URL: https://github.com/apache/datafusion-comet/pull/5880#discussion_r4081314618
##########
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:
> Could this comment say what the test actually does though?
Reworded in df1512682. The comment now says this Iceberg scan is its own
native block with no JVM input, so its metrics publish per batch, and that the
test covers the registration site rather than the listener order, which stays
unguarded until #5265.
--
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]
