You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Comet Native Executors: run LIMIT and top-K queries with a per-partition limit before the exchange #7365
On a Comet Native Executor cluster (#7289), a query with LIMIT, including Dataset.show(), fails: CometCollectLimitExec performs its own shuffle to one partition inside doExecuteColumnar, so it is not a native block, and the driver's stage-graph builder refuses it with "it does not run natively".
CometTakeOrderedAndProjectExec has the same shape. The prototype special-cases it as a stage that reads its child through a single-partition exchange and runs Comet's final top-K plan. Unless the child is a CometLocalTopKExec, the child's whole output crosses that one-partition exchange, where Comet on Spark runs a per-partition top-K first.
Describe the potential solution
Plan both operators as explicit native pieces, as Spark plans LocalLimit, then an exchange, then GlobalLimit: a block with a per-partition limit or top-K, a CometShuffleExchangeExec to a single partition, and a block with the final limit or top-K. The generic stage-graph builder then handles them without a special case, and only the first rows of each partition cross the exchange.
Additional context
The special case is takeOrderedStage in CometNativeExecutorsExec.scala. It calls finalNativePlan(inputPartitions = 2) only to select the sorting branch.
A smaller first step fixes the failure without the per-partition limit. Build a stage for CometCollectLimitExec the way takeOrderedStage builds one for top-K: the stage reads the child's whole output through a single-partition exchange, and its block is CometExecUtils.getLimitNativePlan(child.output, limit, offset), Comet's final limit plan. A query that ends in LIMIT, with or without OFFSET, then runs on the cluster, though every row of the child still crosses the exchange. The per-partition limit and dropping the special cases, as described above, can follow.
Activity
andygrove commentedon Nov 7, 2026
A smaller first step fixes the failure without the per-partition limit. Build a stage for
CometCollectLimitExecthe waytakeOrderedStagebuilds one for top-K: the stage reads the child's whole output through a single-partition exchange, and its block isCometExecUtils.getLimitNativePlan(child.output, limit, offset), Comet's final limit plan. A query that ends inLIMIT, with or withoutOFFSET, then runs on the cluster, though every row of the child still crosses the exchange. The per-partition limit and dropping the special cases, as described above, can follow.