This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7629d9ff0295 [SPARK-57979][SQL] Extract the SortExec metric-recording
epilogue into a shared helper
7629d9ff0295 is described below
commit 7629d9ff0295a8625dfccb9963b4d43b4e481793
Author: Gengliang Wang <[email protected]>
AuthorDate: Wed Jul 15 22:13:35 2026 -0700
[SPARK-57979][SQL] Extract the SortExec metric-recording epilogue into a
shared helper
### What changes were proposed in this pull request?
`SortExec` records the same three sort metrics -- sort time, peak memory
and spill size -- in two
places, using identical logic:
- the interpreted path in `SortExec.doExecute`, and
- the generated code in `SortExec.doProduce`, which re-emits the four
metric-update statements into
every SortExec whole-stage-codegen stage.
This PR extracts that type-independent metric bookkeeping into a shared
helper
`SortExec.recordSortMetrics(sorter, metrics, spillSizeBefore, sortTime,
peakMemory, spillSize)` and
calls it from both paths. The generated `if (needToSort) { ... }` block
shrinks from the four inline
metric statements to a single static call:
Before (generated code, per SortExec stage):
```java
if (sort_needToSort_0) {
long sort_spillSizeBefore_0 = sort_metrics_0.memoryBytesSpilled();
sort_addToSorter_0(partitionIndex);
sort_sortedIter_0 = sort_sorter_0.sort();
((SQLMetric) references[3] /* sortTime
*/).add(sort_sorter_0.getSortTimeNanos() / 1000000);
((SQLMetric) references[1] /* peakMemory
*/).add(sort_sorter_0.getPeakMemoryUsage());
((SQLMetric) references[2] /* spillSize */).add(
sort_metrics_0.memoryBytesSpilled() - sort_spillSizeBefore_0);
sort_metrics_0.incPeakExecutionMemory(sort_sorter_0.getPeakMemoryUsage());
sort_needToSort_0 = false;
}
```
After:
```java
if (sort_needToSort_0) {
long sort_spillSizeBefore_0 = sort_metrics_0.memoryBytesSpilled();
sort_addToSorter_0(partitionIndex);
sort_sortedIter_0 = sort_sorter_0.sort();
org.apache.spark.sql.execution.SortExec.recordSortMetrics(
sort_sorter_0, sort_metrics_0, sort_spillSizeBefore_0,
((SQLMetric) references[3] /* sortTime */),
((SQLMetric) references[1] /* peakMemory */),
((SQLMetric) references[2] /* spillSize */));
sort_needToSort_0 = false;
}
```
This follows the same "extract type-independent generated machinery into a
compiled helper" pattern
as SPARK-57909 (`ColumnarToRowExec.advanceBatch`), and additionally
de-duplicates the interpreted
path, which had the same four lines.
Measured with `WholeStageCodegenSizeBenchmark` (added under SPARK-57915;
plans all 135 TPC-DS
queries, 717 of the 2247 whole-stage-codegen stages are SortExec), current
`master` vs. this change:
| metric | master | this PR | delta |
| --- | --- | --- | --- |
| max method bytecode, summed over stages | 841,185 | 821,349 | -2.4% |
| constant pool, summed over stages | 432,881 | 427,454 | -1.3% |
| source code size (chars) | 21,846,001 | 21,763,355 | -0.4% |
| inner classes | 258 | 258 | unchanged |
| codegen fallbacks | 0 | 0 | unchanged |
The reduction is in compiled bytecode and constant-pool entries, not only
source text -- these are
the metrics that gate the 64KB method / constant-pool limits and HotSpot's
8KB JIT threshold.
`spillSizeBefore` is still captured in the generated code before
`addToSorter` runs (row insertion
can spill), and passed into the helper, so the reported `spillSize`
continues to reflect only this
operator's contribution.
### Why are the changes needed?
This is part of the umbrella SPARK-56908 (reduce the size of code generated
by whole-stage codegen).
Moving the type-independent metric bookkeeping out of the generated code
shrinks every SortExec
stage: the four metric statements (each a `references[]` cast plus a method
call, so real bytecode
and constant-pool method-references that Janino cannot fold away) collapse
to one call, compiled once
per JVM instead of re-emitted per stage. Planning all 135 TPC-DS queries
produces 717 SortExec
whole-stage-codegen stages, so the reduction applies broadly. It also
removes a copy of the metric
logic, so the two paths can no longer drift.
### Does this PR introduce _any_ user-facing change?
No. The generated code and the interpreted path compute exactly the same
metrics as before; this is a
pure code-organization change with no behavioral difference.
### How was this patch tested?
Existing tests, run with whole-stage codegen both on and off:
- `SortSuite` (the sort operator's correctness and metrics), and
- the sort-related cases in `WholeStageCodegenSuite`.
No behavior changes, so no new tests are added; the change is exercised by
the existing SortExec
coverage on both the interpreted and generated paths.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57058 from gengliangwang/SPARK-57979-sort-epilogue.
Authored-by: Gengliang Wang <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
---
.../org/apache/spark/sql/execution/SortExec.scala | 38 ++++++++++++++++------
1 file changed, 28 insertions(+), 10 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
index 351b3c009419..74afa396264e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
@@ -26,8 +26,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
CodeGenerator, ExprCode}
import org.apache.spark.sql.catalyst.plans.physical._
-import org.apache.spark.sql.catalyst.util.DateTimeConstants.NANOS_PER_MILLIS
-import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
/**
* Performs (external) sorting.
@@ -120,10 +119,7 @@ case class SortExec(
// figure out how many bytes we spilled for this operator.
val spillSizeBefore = metrics.memoryBytesSpilled
val sortedIterator = sorter.sort(iter.asInstanceOf[Iterator[UnsafeRow]])
- sortTime += NANOSECONDS.toMillis(sorter.getSortTimeNanos)
- peakMemory += sorter.getPeakMemoryUsage
- spillSize += metrics.memoryBytesSpilled - spillSizeBefore
- metrics.incPeakExecutionMemory(sorter.getPeakMemoryUsage)
+ SortExec.recordSortMetrics(sorter, metrics, spillSizeBefore, sortTime,
peakMemory, spillSize)
sortedIterator
}
@@ -174,10 +170,8 @@ case class SortExec(
| long $spillSizeBefore = $metrics.memoryBytesSpilled();
| $addToSorterFuncName(partitionIndex);
| $sortedIterator = $sorterVariable.sort();
- | $sortTime.add($sorterVariable.getSortTimeNanos() /
$NANOS_PER_MILLIS);
- | $peakMemory.add($sorterVariable.getPeakMemoryUsage());
- | $spillSize.add($metrics.memoryBytesSpilled() - $spillSizeBefore);
- |
$metrics.incPeakExecutionMemory($sorterVariable.getPeakMemoryUsage());
+ | org.apache.spark.sql.execution.SortExec.recordSortMetrics(
+ | $sorterVariable, $metrics, $spillSizeBefore, $sortTime,
$peakMemory, $spillSize);
| $needToSort = false;
| }
|
@@ -212,3 +206,27 @@ case class SortExec(
override protected def withNewChildInternal(newChild: SparkPlan): SortExec =
copy(child = newChild)
}
+
+object SortExec {
+ /**
+ * Records the sort metrics (sort time, peak memory and spill size) for a
completed sort. This
+ * is called both by the interpreted path ([[SortExec.doExecute]]) and by
the generated code of
+ * [[SortExec]] (through the static forwarder), so the type-independent
metric bookkeeping is
+ * compiled once per JVM instead of being re-emitted into every SortExec
stage's generated code.
+ *
+ * `spillSizeBefore` is the task's spilled-bytes count captured before the
rows were fed to the
+ * sorter, so that `spillSize` reflects only this operator's contribution.
+ */
+ def recordSortMetrics(
+ sorter: UnsafeExternalRowSorter,
+ metrics: TaskMetrics,
+ spillSizeBefore: Long,
+ sortTime: SQLMetric,
+ peakMemory: SQLMetric,
+ spillSize: SQLMetric): Unit = {
+ sortTime += NANOSECONDS.toMillis(sorter.getSortTimeNanos)
+ peakMemory += sorter.getPeakMemoryUsage
+ spillSize += metrics.memoryBytesSpilled - spillSizeBefore
+ metrics.incPeakExecutionMemory(sorter.getPeakMemoryUsage)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
