andygrove opened a new issue, #7365: URL: https://github.com/apache/datafusion-comet/issues/7365
### 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. -- 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]
