dwsmith1983 commented on code in PR #5880:
URL: https://github.com/apache/datafusion-comet/pull/5880#discussion_r3996944769
##########
spark/src/test/scala/org/apache/spark/sql/comet/CometTaskMetricsSuite.scala:
##########
@@ -128,6 +128,84 @@ class CometTaskMetricsSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
}
+ test("scan input metrics add to the task once per accumulator and keep
existing values") {
+ val nestedBytes = new SQLMetric("nestedBytes", -1L)
+ val nestedRows = new SQLMetric("nestedRows")
+ val nestedPruned = new SQLMetric("nestedPruned")
+ val siblingBytes = new SQLMetric("siblingBytes", -1L)
+ val siblingRows = new SQLMetric("siblingRows")
+ val nestedScan = CometMetricNode(
+ Map(
+ "bytes_scanned" -> nestedBytes,
+ "output_rows" -> nestedRows,
+ "pushdown_rows_pruned" -> nestedPruned))
+ // An operator above the scans contributes output_rows of its own, which
is not input.
+ val outerRows = new SQLMetric("outerRows")
+ outerRows.set(99L)
+ val outerTree = CometMetricNode(
+ Map("output_rows" -> outerRows),
+ Seq(
+ nestedScan,
+ CometMetricNode(Map("bytes_scanned" -> siblingBytes, "output_rows" ->
siblingRows))))
+
+ Seq(None, Some(new IllegalStateException("failed native stage"))).foreach
{ failure =>
+ val ctx = TaskContext.empty()
Review Comment:
> Worth a line in the suite noting that these tests depend on
`markTaskCompleted` running.
Added above the first unit test.
--
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]
