Skip to content

EngineAggregatorNode

Overview

  • Executes the ExecutionDAG over persisted artifacts using the aggregation engine.
  • Exists to combine scan results deterministically into final result sets.
  • Sits after layer_router and before AnswerSynthesizerNode.
  • Class: EngineAggregatorNode
  • Source: packages/nl2sql/src/nl2sql/pipeline/nodes/aggregator/node.py

Responsibilities

  • Load scan artifacts from artifact_refs, lazily, as polars scans.
  • Execute combine and post‑operation nodes in DAG order, as one lazy plan.
  • Save each terminal result as its own artifact, and produce AggregatorResponse with those artifacts and a preview of each.

Position in Execution Graph

Upstream: - layer_router (after all scan artifacts are available)

Downstream: - AnswerSynthesizerNode, when aggregation succeeded - End of run, when it failed (aggregator_route in pipeline/routes.py)

Trigger conditions: - Router dispatches to aggregator when no pending scan nodes remain.

flowchart LR
    Router[layer_router] --> Aggregator[EngineAggregatorNode] --> Synth[AnswerSynthesizerNode]

Inputs

From GraphState:

  • execution_dag (required)
  • artifact_refs (required)

Validation performed:

  • None in node; aggregation service raises on missing artifacts.

Outputs

Mutations to GraphState:

  • aggregator_response (AggregatorResponse)
  • reasoning and errors

Side effects:

  • Reads artifacts from the artifact store.
  • Writes one artifact per terminal result, in the run's folder: <tenant_id>/<request_id>/result_<node_id>.parquet with the default template (see Artifact Store).
  • Writes the run's result manifest beside them, <tenant_id>/<request_id>/manifest.json (see Artifact Store).

Internal Flow (Step-by-Step)

  1. Read execution_dag and artifact_refs.
  2. Check every combine input against its row cap (see Row cap before a combine below). A capped input to a join/compare ends the run here with RESULT_TRUNCATED; a capped input to a union adds a RESULT_TRUNCATED warning and aggregation goes on.
  3. Invoke AggregationService.execute(dag, artifact_refs).
  4. A join/compare combine casts both sides of each key to one type before the join, in polars, non-strictly (Wave A, D30). The types are the ones the decomposer declared (JoinKeyPair.left_dtype, right_dtype): each side is cast first to its own, then both to the type they meet as (nl2sql/aggregation/join_keys.py: text is parsed as the other side's type, two numbers meet as a whole number, a date and a datetime as a datetime, a number and a boolean as a boolean by 0 and 1). A number goes straight to the number type, and two keys declared float whose frames hold integers join as integers: a float cannot tell ids above 2^53 apart. A cast never changes a value, it nulls it: 7.9 or a decimal 7.50 cast to a whole number is null, not 7 or 8. Text is stripped of surrounding spaces, on both sides of a text key too. Timestamps of any unit and zone become naive UTC microseconds; text becomes a date or timestamp only by the ISO formats (2024-01-02, 2024-01-02 13:00[:00[.fff]], the same with T, a trailing Z or a +02:00 offset), never by inference, and a TIME key declared a date or timestamp is null. Text parsed as a number loses its padding, so "007" and "7" both match 7: a key of zero-padded codes should be joined as text, or it may fan out. A value that does not convert becomes null and matches nothing; AggregationService.execute counts them per side (PolarsDuckdbEngine.unconverted_join_keys) and the node adds a JOIN_KEY_UNCONVERTED warning for each, carried to QueryResult.warnings: "3 of 412 support.customer_ref values are not numbers and joined nothing." No dialect is involved: this is compute on result frames.
  5. A join/compare combine resolves each join key against its frame's columns. A key written with a side or sub-query prefix (right.customer, sq_2.customer) resolves to the bare column when only that exists; an unknown key is left for polars to report. A join/compare with fewer than two inputs raises (AGGREGATOR_FAILED) instead of returning the one frame it has; the decomposer refuses such a group before anything runs, so this is the last line of defence.
  6. A post-combine op (PolarsDuckdbEngine.post_op) applies every field it carries, in SQL's order: the reshaping its operation names (aggregate groups with polars' group_by, project selects expected_schema), then its filters (after an aggregate they filter the aggregated rows, as HAVING does), its order_by, and its limit. A filter op with order_by and limit, as the decomposer's prompt used to show, keeps all three.
  7. Before polars is asked for any column, a post-combine op's attributes (group_by, metrics, project's expected_schema, filters, order_by) are checked against the combined frame. An attribute the frame does not have is refused with the columns it does have, and a side-qualified one (right.customer) with the reason — see No anti-join below. The decomposer runs the same check on the decomposition before anything executes (nl2sql/aggregation/columns.py, shared with this engine) and asks the model again once, so here it is the last line of defence.
  8. AggregationService.execute returns each terminal node as a polars LazyFrame: the scans are ArtifactStore.scan_parquet, and the combines and post-combine ops only add to the plan, so no sub-query result is loaded whole before it is written.
  9. Save each terminal result with ArtifactStore.write_lazy, which streams the plan into a parquet artifact, and build AggregatorResponse:
  10. computed_artifacts[node_id]: the saved result's ArtifactRef, with the whole result's row_count and columns (sub_query_id unset).
  11. results: one ResultSummary per saved result (node_id, columns, row_count, and truncated: a sub-query upstream of the node in the DAG was cut at its row cap), also written as the run's manifest.json. QueryResult.result_manifest carries it, and the facade's paging and CSV export read the result by it.
  12. terminal_results[node_id]: the first settings.results.preview_rows rows, read back from that artifact. This is what the answer model is shown; it is not the whole result.
  13. Return success reasoning.
  14. On exception, emit AGGREGATOR_FAILED, on the state's errors and on AggregatorResponse.errors. The graph's aggregator_route then ends the run: the answer synthesizer is not called on the empty result, so the aggregator's error is the run's outcome and no model call is spent explaining {}.

Row cap before a combine

The generator caps every sub-query at its datasource's row_limit, bounded by the settings.results.row_limit ceiling (100,000, also the cap when none is set), unless the plan asks for fewer rows itself, and records that cap as row_cap on its response and on the sub-query's SubgraphOutput. The cap is applied in each sub-query's SQL, before the combine: a join input with 2240 rows under a row_limit of 1000 reaches the aggregator as 1000, and every match the other 1240 held is silently lost. The wrapper marks such a sub-query's SubgraphOutput.truncated, which QueryResult and the answer carry (see AnswerSynthesizerNode).

So before combining, the aggregator looks at each combine input whose result has row_cap rows or more:

  • join / compare: refused. The run ends with RESULT_TRUNCATED (through aggregator_route, so no answer is synthesized), naming the sub-query and the cap. Missing rows mean missing matches, so the combined answer could be wrong, and a wrong answer is worse than none.
  • union: a warning. The rows a union returns are still right, only possibly incomplete, so it runs and a RESULT_TRUNCATED warning with the sub-query's id is carried to QueryResult.warnings.

A result of exactly row_cap rows may be complete; telling the two apart would mean fetching one row more, so it is treated as cut short. A plan's own top-N (LIMIT 5) is the answer, not truncation, and is never flagged. To answer such a question, ask something narrower (aggregate in the sub-query) or raise the datasource's row_limit.

No anti-join: what "bought X but never Y" does here

CombineGroup.operation is one of standalone, compare, join and union. join and compare are both inner joins (how="inner" in PolarsDuckdbEngine.combine), and FilterSpec has no null or existence operator, so there is no way to express "present on the left and absent on the right". This is a capability gap in the plan language, not a bug in the engine.

Asked "Which customers bought jazz tracks but never rock?", the model decomposes it into the two sets and then reaches for a right-hand column to negate against — right.customer. That column does not exist: an inner join keyed on the shared column drops the right-hand copy, and the other right-hand columns are suffixed _right. Every tier 2 run has failed this question, and what reached the caller was polars' own unable to find column "right.customer"; valid columns: ["customer"].

Resolving the prefix away would be worse. right.customer would become customer, the filter would run against the intersection, and the answer would be the customers who bought both genres — wrong, and silently so. So the attribute check refuses the question and says why. The decomposer now makes this check first and asks the model once more with the same message; a second answer with the same problem ends the run in the decomposer with it, before any scan runs. A recording run also met the made-up variant, a filter on bought_rock, which the same check catches. The message reads:

Aggregator failed: Post-combine operation references 'right.customer', which
the combined result does not have. Available columns: customer. A column
qualified by side does not survive the combine: ... it cannot be expressed:
the plan language has no anti-join or set-difference operation ... so the
question is refused rather than answered with the intersection.

The same gap has a second shape: the decomposer sometimes emits the two sub-queries uncombined, and AggregationService.execute then returns two terminal result sets, which the run reports as "expected one result set, got 2" (2026-09-22_27cad30). Both are the same missing operation.

Closing it properly means adding an anti-join (or except) to CombineGroup.operation, the decomposer prompt, the DAG and the engine. That is a feature, not a fix, and is deliberately not done here.

Contracts & Interfaces

Implements a LangGraph node callable:

def __call__(self, state: GraphState) -> Dict[str, Any]

Key contracts:

  • AggregatorResponse
  • ExecutionDAG

Determinism Guarantees

  • Deterministic for a fixed DAG and artifact inputs.
  • Aggregation order follows DAG layers and ordered inputs.

Error Handling

Emits PipelineError with:

  • AGGREGATOR_FAILED
  • JOIN_KEY_UNCONVERTED, a warning: join-key values that did not convert to the key's type
  • RESULT_TRUNCATED, an error when a join/compare input hit its row cap, a warning when a union input did

Logs failures via logger.error.

Retry + Idempotency

  • No internal retry logic.
  • Idempotent for a fixed DAG and artifacts.

Performance Characteristics

  • Scans Parquet artifacts lazily and executes joins/aggregations in polars; the combined result is streamed to its artifact (sink_parquet), and only the preview rows are collected into Python.
  • Cost grows with artifact size and DAG complexity. A join, sort or group-by may still need its inputs in memory as polars columns, never as Python rows.

Observability

  • Logger: aggregator
  • Adds reasoning entries for success and failure.

Configuration

  • Uses PolarsDuckdbEngine.
  • settings.results.preview_rows: rows of each terminal result kept in terminal_results.
  • The combined result is written to the runtime's results store (ctx.artifact_store, <storage.path>/results/), as the sub-query artifacts are.

Extension Points

  • Replace aggregation engine by modifying node to use a different AggregationService.
  • Replace node in build_graph() for alternate aggregation behavior.

Known Limitations

  • Fails if required artifacts are missing.
  • Several post-combine ops on one combine group are siblings, not a chain: each reads the combine's output and is its own terminal result.
  • packages/nl2sql/src/nl2sql/pipeline/nodes/aggregator/node.py
  • packages/nl2sql/src/nl2sql/aggregation/aggregator.py