Skip to content

Comet Native Executors: run LIMIT and top-K queries with a per-partition limit before the exchange #7365

Open

Description

@andygrove

What is the problem the feature request solves?

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.

Part of #7288.

Willingness to contribute

Yes. I can contribute this feature independently.

Activity

  1. self-assigned this
    on Nov 1, 2026
  2. andygrove commented on Nov 7, 2026

    @andygrove
    MemberAuthor

    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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

enhancementNew feature or request

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions