Skip to content

Comet Native Executors: run a small broadcast side once rather than in every task that reads it #7369

Open

Description

@andygrove

What is the problem the feature request solves?

On a Comet Native Executor cluster (#7289), a broadcast exchange becomes a CoalescePartitionsExec, which Ballista turns into a stage boundary. When the broadcast side has only one partition, such as a scan of a small table, Ballista's adaptive planner drops that coalesce as a no-op. The broadcast side then has no stage of its own and runs inside the stage that reads it, once in each of that stage's tasks. Results are correct, but the work is repeated, which matters for a broadcast side that is cheap to read but expensive to compute.

Describe the potential solution

Keep a stage boundary on every broadcast input, for example by having the scheduler mark broadcast exchanges so Ballista always cuts a stage there, or by pinning a submitted CoalescePartitionsExec as an ExchangeExec in Ballista's PinSubmittedShufflesRule, as it already pins hash and range repartitions.

Additional context

Seen in the scheduler log for TPC-H q2, where one-partition broadcast sides sit directly under the stage that reads them.

Part of #7288.

Willingness to contribute

Yes. I can contribute this feature independently.

Metadata

Metadata

Assignees

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions