From SQL to a GPU plan

DataFusion physical operators run on the CPU and partition input streams. Algeon's adapter lowers supported physical subtrees into native IR, replacing them with GPU execution wrappers.

Walk through a grouped aggregate

Consider orders registered from local Parquet files, with customer_id stored as Int32, amount as Int64 cents and status as UTF-8 strings:

SELECT customer_id, SUM(amount) AS total
FROM orders
WHERE status = 'paid'
GROUP BY customer_id;

Select a node to inspect its mapping. Data flows upward. The two dashed arrows map the partial and final aggregates to one native aggregate. Host Arrow materialization belongs to the wrapper. Batching and scan-repartition wrappers are omitted.

Partial + final aggregates lower to one grouped SUM

For a compatible partial/final pair, lowering recovers SUM(amount) and the customer_id grouping key as one native Aggregate. Its compiled task has kind aggregate. Chunk execution may produce partial states and merge them before downstream output. Those runtime partials do not add a separate compiled task for each chunk.

Observe this boundary. partial_aggregate_count, partial_aggregate_input_rows and final_merge_attempts explain chunk execution. Use task_runtime_rows for the compiled task's observations.

The plan represents a distributed aggregate with a separate row filter. Select an operator to display its native IR representation and observation fields.

Compare the physical tree, native IR and final DataFusion wrapper

The relevant DataFusion operator parameters are shown below. P is the physical plan's hash partition count. Input data flows from the leaf upward.

ProjectionExec: customer_id, sum(orders.amount) AS total
  AggregateExec: mode=FinalPartitioned, group=[customer_id], merge=[SUM state]
    RepartitionExec: Hash(customer_id), partitions=P
      AggregateExec: mode=Partial, group=[customer_id], aggr=[SUM(amount)]
        FilterExec: status = 'paid'
          DataSourceExec: Parquet, projection=[customer_id, amount, status]

Lowering models partial state and final merge explicitly to reconstruct the SQL aggregate. The native IR retains a grouped aggregate plus the row predicate and result projection. Before compilation, the adapter changes the sink to DeferredHostArrow, preserving the host result contract while leaving materialization to the wrapper. The simplified native shape is:

Sink: deferred_host_arrow (materialization by the wrapper)
  Projection: customer_id, sum(orders.amount) AS total
    Aggregate: group=[customer_id], SUM(amount)
      Filter: status = 'paid'
        LocalParquetSource: read_columns=[customer_id, amount, status]

A final aggregate reaches its partial stage through a repartition, a coalesce, or a merge of sorted partitions without a fetch. Each operator merely redistributes or interleaves rows, allowing lowering to bypass it and preserve the partial aggregate as the executable operator under the final schema. DataFusion places the merge below a final aggregate that streams over sorted groups, which happens when a Parquet source declares its sort order. A merge with a fetch drops rows and stays unsupported, as does any other operator below the final stage. The rejection names that operator.

Once this whole-query candidate is selected, the completed DataFusion tree contains the attempt's admission wrapper and the native execution wrapper:

NativeAdmissionExec
  AlgeonNativeExec

AlgeonNativeExec holds the native IR and compiled pipeline internally. The native operators are not DataFusion child nodes. A shorter EXPLAIN tree can therefore represent the same scan, filter and aggregation work.

The distributed aggregate lowering tests cover partial/final reconstruction, and the native IR goldens show scan, filter, aggregate and projection nodes for related queries.

What the GPU compiler produces

The native optimizer rewrites the IR before capability validation and pipeline compilation. Pipeline compilation produces an execution schedule over GPU tasks (source_read, unary, aggregate, join). The schedule records data dependencies, completion barriers, and resident resource lifetimes.

An aggregate barrier prevents consumers from reading incomplete group state. A resource between tasks tracks its producer and consumers, remaining live until all downstream consumers finish. During chunk execution, a compiled aggregate task accumulates multiple partial batches into its working state.

At deferred host sinks, output buffers remain GPU-resident until AlgeonNativeExec materializes host Arrow batches incrementally.

A native join expands into build, probe and materialization tasks, with a build-before-probe barrier. The pipeline compiler owns these task and resource relationships.

Shared resident subplans

When DataFusion inlines a CTE or common subquery across physical units, projection pruning retains only referenced columns in each copy. When two or more native units of one attempt contain an identical closed subplan whose root reduces rows (such as an aggregate or a limit with a fetch), native optimization extracts a single GPU-resident producer.

The first reading unit computes the result, which remains resident on the device under the attempt's grant until all registered units have read it. Streaming chains of scans, filters, projections and joins continue to execute chunk by chunk for each reader. In EXPLAIN output, the extracted producer reports shared_consumers, and shared_resident_extraction diagnostics record per-candidate decisions.

Partitions and result semantics

Native execution replaces DataFusion CPU partition streams with chunk-based execution and dedicated worker lanes. Upstream repartition and coalescing operators bypass into the native pipeline without mapping partition counts to CUDA streams.

Lowering reconstructs operations whose semantics span chunks, such as multi-pass aggregates (AVG retaining sums and counts for final division) and distributed sort and limit merges. A merge of sorted partitions lowers as one native sort by the merge's keys, expression keys included. An InterleaveExec, which DataFusion plans for a UNION ALL whose branches share a hash partitioning, lowers as a union.

A window with ORDER BY keys, or one that numbers rows by position inside partitions, reads its input ordered by its partition keys, then its ORDER BY keys. DataFusion supplies that ordering with a sort and may push the sort below a hash join under the assumption that probe ordering is preserved. Because native joins do not preserve probe order, lowering inserts an explicit sort before the window operator when the input is a join, aggregate, union or GPU-resident relation, and it drops a sort without a fetch below an operator that does not read row order, such as a join input.

The native host-output wrapper exposes one bounded output partition. When native fragments sit under CPU operators, the adapter verifies the parent's schema, ordering and distribution requirements, including co-partitioned join inputs. Incompatible replacements remain on CPU execution.

Connect the plan to an execution

Use the same SQL and session settings when inspecting the example. On the Flight SQL server, EXPLAIN VERBOSE exposes physical-plan details, while EXPLAIN GPU returns native candidate outcomes and coverage of the completed plan:

EXPLAIN GPU
SELECT customer_id, SUM(amount) AS total
FROM orders
WHERE status = 'paid'
GROUP BY customer_id;

EXPLAIN GPU plans its inner query without executing it, so its query ID carries no execution observations. After consuming the query's Flight DoGet result, retrieve GetQueryStats with that execution's query ID. Its [planning] section describes selection; [execution] contains the attempt's observed work. Available [physical_plan] and [native_explain] sections retain the textual plans.

Diagnostic focus	Fields to inspect	Interpretation
Subtree replacement	physical_root, post_native_plan_root, native_plan_fragment_count, host_datafusion_boundary_count	Candidate selection and final executable coverage are separate facts.
Operator task mapping	plan_task_mapping_rows, task_operator_rows, task_runtime_rows	Correlate native plan nodes with compiled tasks and their observations.
Row reduction	row_groups_selected, source_chunks_committed, partial_aggregate_input_rows, output_rows	Row groups, chunks, aggregate input rows and result rows have different scopes.
Aggregation merge execution	partial_aggregate_count, final_merge_attempts, final_merge_levels	Explain the native chunk-and-merge path without inferring it from CPU partitions.
Memory limits	query_grant_bytes, allocator_high_water_bytes, resident_dataframe_peak_bytes	The grant is the allocation cap. Allocator and resident-dataframe peaks measure different memory scopes. The allocator peak covers the attempt's whole allocation domain and is present for failed attempts.
Lifecycle and cleanup	attempt_outcome, attempt_cleanup_disposition, attempt_capacity_disposition	Distinguish the query result from cleanup and the capacity's reuse state.

For task edges, barriers and a timeline, GetQueryRuntimeArtifact returns a control frame followed by JSONL when the sealed artifact is available. The artifact includes native plan nodes, task definitions, resource relationships and runtime events. Correlate (plan_execution_id, plan_node_id) with the task definitions, then use (plan_execution_id, task_id) for task observations. IDs are local to their plan execution. The DataFusion tree and native task graph do not have a one-to-one node mapping.

Retrieve the statistics and runtime artifact with a Flight client

Here, client is the authenticated pyarrow.flight.FlightClient, options contains the same call headers used for the query, and query_id comes from its Flight SQL statement ticket. Consume its DoGet stream before requesting the terminal artifact.

import json
from pyarrow import flight

stats = b"".join(
    bytes(result.body)
    for result in client.do_action(
        flight.Action("GetQueryStats", query_id.encode()), options
    )
).decode()

frames = [
    bytes(result.body)
    for result in client.do_action(
        flight.Action("GetQueryRuntimeArtifact", query_id.encode()), options
    )
]
control = json.loads(frames[0])
if control["runtime_jsonl_follows"]:
    runtime_records = [json.loads(line) for line in frames[1].splitlines()]

The control frame describes artifact availability. Capture is bounded: check incomplete, artifact_truncated and metric_projection_truncated in the statistics, and completeness information in the runtime artifact. An absent observation does not establish that its work took zero time or used zero bytes. Component timers and attempt elapsed time also have different scopes. Their sum is not a kernel-only GPU duration.

The Flight action handlers define retrieval and ownership checks. The runtime JSONL exporter writes the retained topology, events and completeness footer.

Select an execution path

Selection happens before execution. A running native query reports failures to its caller without replaying the query on CPU.

The default mode first attempts whole-query lowering. Unsupported operators, types or semantics leave CPU work in the completed plan; supported subtrees may still become GPU fragments. Explicit graph, vector and ML functions also form GPU fragments. Transfers at CPU/GPU boundaries are part of the cost of such a mixed plan.

A join condition that reads both inputs stays on the join. Each part of it that reads one input and that the cuDF join AST cannot evaluate, such as LIKE, a string function or IS DISTINCT FROM, becomes a column of that input computed before the join, as an expression join key does. Only a part that cannot fail moves, because DataFusion evaluates the condition on key-matched pairs alone. Native Evaluation Semantics states where a part with a checked operation runs.

Decimal division in a keyed inner join predicate uses the native GPU post-join filter. Equality keys first select candidate row pairs. The GPU column evaluator then checks decimal division and preserves CASE branch activation through the shared expression lowering contract. Unmatched rows do not evaluate the residual, and inactive branches cannot trigger division by zero or scaling overflow. Outer, semi, anti and keyless joins retain their existing residual capability restrictions.

PlanningReport records each candidate once as selected, unsupported with the first unmet contract, or supported but kept on DataFusion by a preference rule, with structured reasons. An unsupported candidate also names the element that blocked it, such as one cast between two types, a scalar or aggregate function, or an operator, at the site that rejected it. A missing memory estimate cannot reject a candidate. Allocation limits are enforced during execution.

Parquet source schema

Lowering reads the footer of every distinct object in a Parquet scan, local or remote, and counts byte-range entries of one object once. All footers must hold the same columns in the same order with the same Arrow types; metadata on nested fields is ignored. A column may be required in one file and optional in another when the table schema declares it nullable. Any other difference, including a file that declares a column optional while the table declares it non-null, rejects the scan with native_source_file_schema_mismatch. The source diagnostics then report schema_mapping_disposition as rejected with rejected feature schema_mapping. Column indices, byte metadata and row-group statistics follow the footer's column order, which can differ from DataFusion's file schema. A remote scan of one object whose footer is not cached takes its schema from DataFusion's file schema and gets no footer pruning or byte metadata.

Byte ranges that cover a whole object coalesce on any store. A partial range maps to row groups through the footer, so a remote partial range needs the cached footer described under Object storage. A remote scan without its footers is rejected with native_source_remote_footer_unavailable and rejected feature remote_footer_unavailable. A rejected remote scan still reports the read strategy and credential mode of the read it would make, such as remote_direct and shared_process_env. Both reasons carry a remedy and a source blocking subject.

Uncorrelated scalar subqueries

DataFusion plans an uncorrelated scalar subquery as a ScalarSubqueryExec by default (datafusion.optimizer.enable_physical_uncorrelated_scalar_subquery is true). Native lowering has no counterpart for that operator. When the execution mode lowers relational SQL (native_preferred or native_required), installing the native optimizer sets the option to false, and DataFusion rewrites each such subquery into a join with its one-row aggregate. Installation covers the Flight SQL server session and embedded sessions built with AlgeonGpuBackend::install_on or with_algeon_native. The server installs the optimizer after reading SessionConfig::from_env(), and in those modes DATAFUSION_OPTIMIZER_ENABLE_PHYSICAL_UNCORRELATED_SCALAR_SUBQUERY has no effect. functions_only keeps DataFusion's default.

The join form lowers natively for a scalar aggregate in a projection, an arithmetic expression, a HAVING clause or a filter, and for several subqueries in one query. The subquery may aggregate the scan the outer query reads with a computed argument, as x > (SELECT AVG(x) FROM t) does by averaging CAST(x AS DOUBLE). With datafusion.execution.collect_statistics at its default true, DataFusion answers an unfiltered MIN, MAX or COUNT over a Parquet table from file statistics as literals projected over a PlaceholderRowExec. Native lowering evaluates those literals over a one-row source.

Require GPU execution

NativeRequired rejects a completed plan containing DataFusion CPU operators, and its error names the blocking element of the first unsupported candidate. FunctionsOnly retains CPU relational processing around explicit GPU functions. The default, NativePreferred, permits a CPU or mixed plan. These modes classify DataFusion plan nodes. CPU orchestration and I/O metadata remain part of native execution. The default statistical mode uses GPU kernels; an optional DataFusion-exact mode for variance and standard deviation uses host accumulation inside the native wrapper. Plan coverage alone does not certify that all computation stays on the GPU.

Use GPU coverage validation to inspect the completed plan without running it. EXPLAIN does not claim a GPU grant.