cloud-fan commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4373735256
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -551,16 +541,10 @@ class TungstenAggregationIterator(
}
} else {
// Adaptive partial aggregation bypassed partial aggregation: emit the
remaining input
- // rows as single-row partial buffers, after the frozen map output
above.
- if (passThroughDrainStart == -1L) {
- passThroughDrainStart = System.nanoTime()
- }
- val out = nextPassThroughOutput()
- if (!passThroughHasNext) {
- aggTime += NANOSECONDS.toMillis(System.nanoTime() -
passThroughDrainStart)
- passThroughDrainStart = -1L
- }
- out
+ // rows as single-row partial buffers, after the frozen map output
above. The drain is
+ // not timed: `next()` is pulled lazily, so timing from the first to
the last pass-through
Review Comment:
I'd keep the timer out. Timing each row costs two `System.nanoTime()` calls,
about 20 ns each on my machine, which is comparable to the work of passing the
row through, and this feature exists to make that path cheap. Summing inside
`nextPassThroughOutput()` avoids counting downstream work and the truncation,
but keeps the per-row cost. So `aggTime` now means the build up to the point
where pass-through starts, plus the map setup, on both paths and in every plan
shape. The docs now say pass-through rows are counted but not timed (item 7),
and the description says this reverts dd0f15465e9 from #57742.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -841,14 +841,14 @@ case class HashAggregateExec(
} else {
""
}
+ // The resumed build is not timed: split by an exchange, every
passed-through row returns and
Review Comment:
Fixed. When pass-through starts, the build timer now stops right after the
map setup, so the fused shape stops timing at the same point as the split one.
It costs one `System.nanoTime()` call per task.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -841,14 +841,14 @@ case class HashAggregateExec(
} else {
""
}
+ // The resumed build is not timed: split by an exchange, every
passed-through row returns and
+ // re-enters here, so a timer would cost two `System.nanoTime()` calls per
row while the
+ // millisecond division truncates each sub-millisecond slice to 0.
Review Comment:
This is the same trade-off as item 1. Summing nanoseconds fixes the
truncation but keeps two calls per re-entry, which is one re-entry per row when
the child checks `shouldStop()` between rows. The description and the code
comment now state the drop as a deliberate choice rather than a side effect of
the truncation.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -79,7 +79,7 @@ case class HashAggregateExec(
// Only the aggregates that can actually bypass report this, so the rest
do not show a
// constant 0 in the SQL UI (see `UnionExec.metrics` for the same
approach).
if (adaptivePartialAggEnabled) {
- Map("numBypassingRows" -> SQLMetrics.createMetric(sparkContext, "number
of bypassing rows"))
+ Map("numPassthroughRows" -> SQLMetrics.createMetric(sparkContext,
"passthrough output rows"))
Review Comment:
Done: "number of pass-through output rows".
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -106,7 +106,8 @@ case class HashAggregateExec(
val aggTime = longMetric("aggTime")
val numTasksFallBacked = longMetric("numTasksFallBacked")
// Registered only when the feature applies, and only read from the
pass-through path.
- val numBypassingRows = if (adaptivePartialAggEnabled)
longMetric("numBypassingRows") else null
+ val numPassthroughRows =
+ if (adaptivePartialAggEnabled) longMetric("numPassthroughRows") else null
Review Comment:
I'd like to keep `numPassthroughRows`. The key isn't user-facing:
`SparkPlanInfo` publishes each metric under its display name, so the UI, event
log, and REST API show only the label, which I've changed to "number of
pass-through output rows" per item 4. Spark also has no one spelling to align
with: `PartitionIdPassthrough` and `ShufflePartitionIdPassThrough` back the
same feature. I've dropped the description's claim that the key follows the
implementation's spelling.
--
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]
