David Mollitor created SPARK-59581:
--------------------------------------
Summary: Avoid copying every input row in TakeOrderedAndProject
top-N
Key: SPARK-59581
URL: https://issues.apache.org/jira/browse/SPARK-59581
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 4.1.0
Reporter: David Mollitor
h2. Summary
{{TakeOrderedAndProjectExec}} implements top-N (`ORDER BY ... LIMIT n`). For
unsorted input it selects the top-K with
{{Utils.takeOrdered(iter.map(_.copy()), limit)(ord)}}
({{sql/core/.../execution/limit.scala}}). The {{iter.map(_.copy())}} copies
*every* input row, even though only {{limit}} rows are ever retained.
The copy is forced by the selector:
{{org.apache.spark.util.collection.Utils.takeOrdered}} wraps
Guava's {{Ordering.leastOf}} / {{TopKSelector}}, which buffers element
*references* while selecting. Since the child's whole-stage-codegen iterator
yields a single reused {{UnsafeRow}}, every row must be copied to a detached
instance before being handed to the selector. For N input rows we allocate N
copies to keep {{limit}} (e.g. 20M copies to keep 100).
h2. Analysis
JFR profiling of a top-N query ({{spark.range(20000000).selectExpr("id","id %
1000 as k").orderBy("k").limit(100)}}) showed {{UnsafeRow.copy()}} at ~93.7% of
sampled allocation and ~15.5% of CPU. The top-K selection itself
({{TopKSelector.offer}}, ordering compare) was ~2% each -- the operator's cost
is almost entirely copying rows that are immediately discarded.
h2. Change
Replace the copy-everything-then-select approach with a bounded top-K that
copies a row only when it is actually retained.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
