dongjoon-hyun commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4368343232
##########
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:
**1. Removing this timer brings back the `aggTime` gap that #57742 closed on
review.** In #57742, viirya and peter-toth ([Finding
7](https://github.com/apache/spark/pull/57742#discussion_r3734471987)) pointed
out that `doExecute` times only this iterator's constructor, so once
pass-through fires, the rows pulled through `nextPassThroughOutput()` are timed
by no metric. dd0f15465e9 added the drain timer to fix that. Without it, a
near-distinct 10M-row partition that bypasses at the default `minRows = 100000`
reports the build of about 1% of its rows in `time in aggregation build`, and
the projection, `processRow` and `generateOutput` work for the rest goes
uncounted. That is the workload this feature targets, and `docs/web-ui.md`
describes the metric as "the time spent on aggregation".
The point about downstream work is right: timing from the first to the last
row also counts the shuffle write. Summing nanoseconds inside
`nextPassThroughOutput()` only, and adding them to `aggTime` once when
`passThroughHasNext` turns false, would avoid both that and the millisecond
truncation, at the cost of two `System.nanoTime()` calls per row. Could you
either take that trade-off, or keep the timer out, document that `aggTime`
excludes pass-through (see item 7), and say in the description that this
reverts dd0f15465e9 from #57742?
##########
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:
5. Minor: the description says the new name is what the implementation calls
these rows, but the implementation spells it `PassThrough`:
`nextPassThroughOutput`, `handlePassThroughRow`,
`passThroughAggregationBuffer`, `passThroughHasNext` and
`adaptivePassThroughTerm`. With `numPassthroughRows`, a case-sensitive search
for `PassThrough` finds every other pass-through identifier but not this
metric. Could you name it `numPassThroughRows` while it is still unreleased?
##########
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:
**2. In the fused shape, `aggTime` still includes every pass-through row and
the `Final`'s work on them.** This comment, and the description's "`aggTime`
again covers only the build up to the point where pass-through starts", hold
only when an `Exchange` separates `Partial` and `Final`. When the child already
satisfies the `Final`'s distribution (single-partition input,
`df.repartition($"k").groupBy($"k")`, or a join or bucketed scan on the
grouping keys), no `Exchange` is inserted and the two aggregates fuse.
`outputFunc` then feeds the `Final`'s `doConsume`, `shouldStop()` never fires,
and the first timed `doAgg` at L872-874 runs the whole partition: the
frozen-map output, every pass-through row, and the `Final`'s hash-map updates
for all of them. So the same query reports all of its pass-through time when
fused and none when split or interpreted. Could you state this exception in the
comment and the description, or record `aggTime` at the point where
pass-through fires, so that
it stops there in both shapes?
##########
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:
7. Minor: this is about `docs/web-ui.md:382`, which is outside the diff. The
SQL metrics table there still describes `time in aggregation build` as "the
time spent on aggregation" for HashAggregate, and it does not list the
pass-through metric at all. After this PR, a task whose work is mostly
pass-through shows a small aggregation time next to an undocumented row count,
which reads as a cheap aggregate. Could you add the metric to that table and
say there (or in the Adaptive Partial Aggregation section of
`sql-performance-tuning.md`) that pass-through rows are counted but not timed?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -137,7 +137,7 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
df.collect()
val skipped = collect(df.queryExecution.executedPlan) {
case agg: HashAggregateExec if isPartialPhase(agg) =>
- agg.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)
+ agg.metrics.get("numPassthroughRows").map(_.value).getOrElse(0L)
Review Comment:
10. Nit: `agg.metrics.get("numPassthroughRows").map(_.value).getOrElse(0L)`
and the surrounding `collect { case agg: HashAggregateExec if
isPartialPhase(agg) => ... }.sum` are repeated at L140, L203, L224 and L505,
and the key appears as a string literal five times in this suite and three
times in `HashAggregateExec.scala`. That is why this rename had to touch eight
literals, and a typo at any one of them would silently read 0 through
`getOrElse(0L)`, which an `== 0` assertion accepts. Could you add one helper,
e.g. `passThroughRows(df: DataFrame): Long`, and use it at those four sites?
##########
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
Review Comment:
8. Minor: "every passed-through row returns and re-enters here" does not
hold for one-to-many children. As the `needStopCheck` comment at L887-894 says,
`GenerateExec` does not check `shouldStop()` inside its fan-out, so one resumed
`doAgg` passes through the whole batch produced from one input row. Such a
slice can exceed a millisecond, so the removed timer did record some time
there. Could you narrow the sentence to children that check `shouldStop()`
between rows?
##########
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:
4. Minor: "passthrough output rows" breaks the naming of the other count
metrics. 73 of the 85 `createMetric(sparkContext, ...)` call sites in
`sql/core` main code start with "number of", including this operator's "number
of output rows" and "number of sort fallback tasks", and the similar "number of
anchor output rows". It also spells "passthrough" as one word, while the config
docs, `sql-performance-tuning.md` and the code comments say "pass-through".
Since the feature is unreleased, could you use "number of pass-through output
rows" (or "number of passed-through rows")?
##########
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:
**3. The removed timer recorded about nothing because of per-slice
truncation, which is fixable.** Each resumed slice was divided by
`NANOS_PER_MILLIS` on its own, so every sub-millisecond slice added 0, and
`beforeResumedAgg` was close to a no-op since #57742. That is a resolution bug
in the timer rather than a reason to stop timing the resumed build. Summing
`System.nanoTime() - start` into a mutable `long` and adding it to `aggTime`
once, when `childrenConsumed` is set, would time only the `doAgg` work; unlike
the interpreted drain, no downstream work runs inside it. The two
`System.nanoTime()` calls per re-entry would stay, so it is a trade-off. Could
you compare it with this removal in the description, so that dropping the time
is a stated 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
Review Comment:
9. Nit: the description argues that pass-through is a better name than
bypass, but the rename stops at the key, so the same counter is still described
as bypass or skip around it: this comment ("Only the aggregates that can
actually bypass report this"), `adaptiveRowBypassedTerm` guarding
`numPassthroughRows.add(1)` at L1319, and in the suite `AggCounters.skipped`,
`bypassRowsByGroupingKeyCount`, the "bypassed rows" assertion messages and the
test title "(independent of the skip metric)" at
`AdaptivePartialAggregationSuite.scala:1384`. Could you rename at least this
comment and the test-side names that describe this metric?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -200,15 +200,15 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
val counters = AggCounters(
// The metric is only registered on aggregates the feature applies to;
an aggregate without
// it bypassed nothing.
- skipped =
partialAggs.map(_.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)).sum,
+ skipped =
partialAggs.map(_.metrics.get("numPassthroughRows").map(_.value).getOrElse(0L)).sum,
partialOutputRows =
partialAggs.map(_.metrics("numOutputRows").value).sum,
spillBytes = partialAggs.map(_.metrics("spillSize").value).sum,
tasksFallBacked =
partialAggs.map(_.metrics("numTasksFallBacked").value).sum)
checkAgainstReference(df, build)
counters
}
- private def numBypassingRows(build: () => DataFrame): Long =
runAndReadCounters(build).skipped
+ private def numPassthroughRows(build: () => DataFrame): Long =
runAndReadCounters(build).skipped
Review Comment:
6. Minor: no test pins what this PR changes. The suite only reads the
renamed key, and the `aggTime` test in `SQLMetricsSuite.scala:357` runs with
the feature off. So if a later change put a per-row timer back on the resumed
build or in `nextPassThroughOutput()`, every test would still pass. Could you
add a generated-code check, like the `copyFrom` test from #59758, that the
generated code of a split adaptive aggregate calls `System.nanoTime()` only for
the one build timer?
--
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]
