Switching away from DriverAdapter

[devavret]

Disclaimer: Bulk of this text is hand-written but parts of it are embellished with AI. I did my own investigation and ideas presented are my own.

I understand this is a wall of text so here's a tl;dr:

We want to move cudf operator selection before LocalPlanner, to have efficient driver counts for CPU fallback, and to not build and throw away cpu operators.

The options are:

1. Use dedicated cudf PlanNodes. This needs the least core infrastructure and allows mixed CPU/GPU execution, but introduces parallel PlanNode types.
2. Select one backend for the whole PlanFragment. This avoids new PlanNodes, but requires broader core changes and makes efficient CPU fallback harder.
3. Store the backend choice per PlanNode. This preserves mixed execution without duplicating most PlanNodes, but still requires general backend-selection infrastructure in Velox.

I currently lean towards dedicated cudf PlanNodes (option 1) as the first step.

Description

Currently, cudf execution starts with a regular Velox plan. LocalPlanner creates CPU operators, and DriverAdapter later replaces compatible runs with cudf operators.

This was useful for bringing up cudf execution without changing the plan, but it has a few problems now:

1. The number of drivers is selected before DriverAdapter considers GPU execution. GPU operators generally want fewer drivers than CPU operators, and the adapter cannot change pipeline parallelism after the driver factories have been created. This also makes CPU fallback inefficient because CPU sections can inherit GPU-sized parallelism. This is described in [cuDF] Decouple CPU and GPU Driver Parallelism to Avoid CPU Underutilization with cuDF DriverAdapter #15820.

2. GPU execution is an implicit runtime decision. We start with a CPU plan, construct CPU operators, and then decide whether to discard and replace some of them. As discussed here: feat: Select output transport per PartitionedOutput node via a pluggable registry #16980 (comment), physical operator selection should ideally be represented in the plan and resolved before operator construction. Besides doing unnecessary work, constructing CPU operators first means their constructors must remain valid even when those operators were never intended to execute.

The goal is to make the CPU/GPU choice before LocalPlanner fixes the pipeline shape and driver count.

There are a few ways we could do this.

Option 1: dedicated cudf PlanNodes

I have been prototyping this approach in the cudf-plan-translator branch.

An external engine can continue producing regular Velox plans initially. Before creating the task, a plan rewriter converts supported parts of the plan to cudf PlanNodes and inserts conversion boundaries where execution changes between CPU and GPU.

auto physicalPlan = CudfPlanRewriter::rewrite(cpuPlan, config);
auto fragment = core::PlanFragment{std::move(physicalPlan)};

For example, a mixed plan could look like:

CPU TableScan
    |
CudfFromVelox
    |
LocalExchange <- Needed to do driver fan-in fan-out
    |
CudfFilterProject
    |
LocalExchange <- Needed to do driver fan-in fan-out
    |
CudfToVelox
    |
unsupported CPU operator

The presence of a CudfFilterProjectNode, CudfAggregationNode, etc. means that support has already been checked by the rewriter. LocalPlanner then uses CudfPlanNodeTranslator to create the corresponding operator directly.

This has the following benefits:

1. Most of the current infrastructure already works. Velox has a custom PlanNodeTranslator interface, and most existing cudf operators only need to accept the dedicated cudf node instead of the corresponding CPU node.

2. LocalPlanner sees the GPU nodes while determining pipeline parallelism. The branch already reports the preferred driver count from cudf PlanNodes, before drivers are created.

   This is necessary but not sufficient for decoupling CPU and GPU parallelism. Nodes in the same pipeline still share a driver count. A CPU/GPU boundary needs a local exchange when the two regions need different widths. The rewriter can make that boundary explicit.

3. It allows mixed CPU and GPU execution within a task. This is useful for fallback and for plans where only some operators are worth running on the GPU.

   This matters in Presto’s single_node_execution_enabled mode. For ordinary queries, this mode commonly produces a single stage containing a single task; there is even a native execution test that expects exactly one stage and one task. If backend selection is made at task granularity, such a query cannot mix CPU and GPU execution.

   This is separate from merely running with one worker. The native test builder has separate settings for one worker and single-node planning.

4. Cudf-specific capabilities can be represented in the physical nodes. We don’t need to pretend that every field and mode supported by a CPU node is also meaningful to the cudf operator and then reject it after CPU operator construction.

There are two gaps in the current Velox extension infrastructure.

Local exchange

LocalPartitionNode is special. LocalPlanner uses the same node to create a LocalPartition sink in each producer pipeline and a LocalExchange source in the consumer pipeline. The custom PlanNode translation hook cannot currently customize this producer-side supplier.

The branch therefore disables the general cudf DriverAdapter but keeps the local-partition adapter enabled. That adapter replaces only the producer-side LocalPartition.

We would need a core extension point for backend-specific pipeline boundary operators or, more narrowly, for the producer-side operator supplier of a local exchange.

Multiple PlanNodes producing one operator

The current custom translator receives one PlanNodePtr and returns one operator. It cannot inspect the remaining nodes in the pipeline or tell LocalPlanner that it consumed more than one node.

Built-in Filter + Project fusion is implemented directly in LocalPlanner: it looks ahead to the next node, creates one FilterProject, and advances past both nodes

velox/velox/exec/LocalPlanner.cpp

Lines 488 to 504 in 14779a1

	for (int32_t i = 0; i < planNodes.size(); ++i) {
	// Id of the Operator being made. This is not the same as 'i'
	// because some PlanNodes may get fused.
	auto id = operators.size();
	auto planNode = planNodes[i];
	if (auto filterNode =
	std::dynamic_pointer_cast<const core::FilterNode>(planNode)) {
	if (i < planNodes.size() - 1) {
	auto next = planNodes[i + 1];
	if (auto projectNode =
	std::dynamic_pointer_cast<const core::ProjectNode>(next)) {
	operators.push_back(
	std::make_unique<FilterProject>(
	id, ctx.get(), filterNode, projectNode));
	i++;
	continue;
	}

.

The current branch handles this particular case by rewriting Filter + Project into one CudfFilterProjectNode. This is sufficient for known fusions, but a general extension could look something like:

struct OperatorTranslation {
  std::unique_ptr<Operator> op;
  uint32_t consumedPlanNodes{1};
};

virtual std::optional<OperatorTranslation> toOperator(
    DriverCtx* ctx,
    int32_t operatorId,
    std::span<const core::PlanNodePtr> planNodes);

LocalPlanner could pass the unconsumed part of the pipeline to the translator and advance by consumedPlanNodes.

We don’t need this immediately for cudf, but it is likely useful for future operator fusion. Wave currently works around the same limitation by inspecting a sequence of already-created CPU operators, creating one WaveDriver, and replacing the complete range through DriverAdapter.

The main disadvantage of this option is the proliferation of PlanNode types. Many cudf nodes will contain nearly the same information as their CPU counterparts and differ primarily in their concrete type.

Option 2: select one execution backend for the PlanFragment

Another option is to continue using the regular CPU PlanNodes and add an execution backend to PlanFragment.

This would also need a backend registry in core.

LocalPlanner would resolve the backend once and store it on each DriverFactory.

Changing only the main operator construction loop is not enough. The selected backend must also be used for:

1. Driver counts, before factory->numDrivers is assigned.
2. Producer-side operator suppliers, including LocalPartition, hash-build, and nested-loop-build operators.
3. Join bridges created by Task.
4. The main operator chain.

For a task-level backend, an unsupported node should fail clearly. Silently creating a CPU operator would make the task mixed-backend again and bring back an implicit fallback decision.

This approach has the benefit that external planners and PlanNode definitions remain unchanged.

The main limitations are:

1. This requires new general infrastructure in core Velox rather than changes mostly contained in experimental/cudf.

2. A task is either CPU or GPU. Unsupported operators require moving part of the query into another task instead of falling back within the current task.

3. Two tasks on the same worker do not use Presto’s normal local exchange. The scheduler still creates a remote split. I could not find a same-process local-exchange path between regular Presto tasks.

   Existing HTTP exchange plus explicit GPU-to-host or host-to-GPU conversion could work but won't be efficient.

   Efficient heterogeneous tasks would require transport selection based on both endpoints. The current UCX implementation is GPU-to-GPU and UcxPartitionedOutput requires CudfVector input. It does have an in-process GPU transfer path, but CPU-to-GPU and GPU-to-CPU communication will need to be supported.

4. Using ordinary PlanNodes means cudf capability validation must live in the backend planner or registry. It is not represented directly by the physical plan.

5. This does not solve the multi-node fusion limitation. A task-level Wave backend would still need a way to consume a range of PlanNodes.

Possible middle ground: backend selection per PlanNode

There is also a middle ground between dedicated node types and one backend for the entire task.

PlanFragment could contain a backend assignment keyed by PlanNode ID:

struct PlanFragment {
  ...
  folly::F14FastMap<PlanNodeId, std::string> executionBackends;
};

This is close to the shape used by #16980. The current PlanFragment has per-node input and output transport maps, rather than one transport choice for the whole task.

A per-node backend map would avoid most dedicated PlanNode types while preserving mixed CPU/GPU execution within a task. It would still need the backend registry and all of the LocalPlanner changes described above. Conversion boundaries would also need to remain explicit, either as PlanNodes or additional boundary metadata.

Preference

Personally, I lean towards dedicated cudf PlanNodes as the first step.

It is already close to working, keeps most of the implementation in experimental/cudf, supports mixed execution within one task, and gives LocalPlanner the information it needs before choosing driver counts. The two core extension points we would eventually need are support for backend-specific local exchange operators and an optional multi-PlanNode translation API.

The per-node backend map is the alternative I would consider if maintaining parallel PlanNode classes becomes a significant problem. I don’t think a single task-wide backend is a good default for cudf fallback because it moves every CPU/GPU boundary to a task exchange.

The main design question is whether physical backend selection should be represented using concrete PlanNode types, or whether Velox wants a general per-node backend registry in core.