Skip to content

Comet Native Executors: broadcast a small build side in pipelined mode #7408

Open

Description

@andygrove

What is the problem the feature request solves?

In pipelined mode, a join whose build side turns out small still shuffles its whole probe side. TPC-H q17 at SF1000 filters part to 199,925 rows, about 2.8 MB, but Spark estimates the filtered part at 654 MiB, because without column statistics a filter doesn't shrink the estimate. So the static plan joins part and lineitem after shuffling both, and moves all 6 billion lineitem rows across the network to do it.

#7391 fixes this for blocking mode with a rule in Ballista's adaptive planner, which holds the probe side's stage back until the build side is measured, then moves the join into the probe side's stage with the build side broadcast to it. Pipelined mode has no adaptive planning: every task is placed before anything runs.

Ballista's pipelined plan for q17 doesn't broadcast part either, and on a 2-node cluster with 2.5 GbE both engines are bound by the two full lineitem shuffles, about 35 GB per node each way (#7409). Removing one of them halves the network traffic.

Describe the potential solution

Measure small build sides before planning the query, as the driver already runs scalar subqueries before submitting a plan:

  1. When comet-scheduler plans a pipelined query, find the joins that Comet Native Executors: use Ballista's adaptive planning for joins inside Comet fragments #7391's rule can move (inner, built on the right, column keys, no condition, both sides read through exchanges) whose build side Spark estimates within Ballista's maximum estimate multiple of the broadcast threshold, and whose probe side Spark estimates many times larger.
  2. Run each such build side's stages as their own pipelined job, and read the result into the scheduler, as the scheduler reads a pipelined query's result.
  3. If the result is under the broadcast threshold, move the join into the probe side's stage with the result as a broadcast input read from memory, using the same rewrite as Comet Native Executors: use Ballista's adaptive planning for joins inside Comet fragments #7391. Otherwise plan the query as it was.

The cost is running the build side's stages before the rest of the query, and once more when the build side turns out large.

Additional context

Part of #7288. Related to #7391 (option D there).

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