Skip to content

refactor(pipeline): build the execution DAG in the decomposer instead of its own node - #177

Merged
nadeem4 merged 1 commit into
mainfrom
refactor/dag-in-decomposer
Sep 23, 2026
Merged

nadeem4 merged 1 commit into
mainfrom
refactor/dag-in-decomposer

Conversation

@nadeem4

@nadeem4 nadeem4 commented Sep 23, 2026 •

Copy link
Copy Markdown
Owner

global_planner read nothing from state but decomposer_response
(global_planner/node.py:34-37), so the ExecutionDAG it produced was a pure
function of the decomposer's output — a format conversion with a graph node, a
response model, a state field and a package wrapped around it. It is now
nl2sql/pipeline/nodes/decomposer/dag.py, a function the decomposer calls, and
GraphState carries the ExecutionDAG itself instead of a one-field wrapper.

decomposer -> global_planner -> layer_router      becomes      decomposer -> layer_router

layer_router is untouched — its empty body is the fan-in barrier for parallel
sub-queries.

Rebased onto main after #175 (the ordinal removal) landed.

Removed, and what read it

removed reader
the global_planner node, package and GlobalPlannerResponse graph.py, routes.py, aggregator/node.py, state.py — all now read state.execution_dag
GraphState.global_planner_response replaced by execution_dag: Optional[ExecutionDAG]
LogicalNode.output_schema, and RelationSchema / ColumnSpec with it only the planner, copying a scan's schema onto the combine node
ExecutionDAG.dag_id, .content_hash, and _hash_execution_dag nothing, anywhere
JsonLiteral only the deleted re-export module
all six scan-node attributes keys five had no reader. The sixth, datasource_id, was read at routes.py:74 on a branch that cannot be taken: a scan node's node_id is its sub-query's id, so sub_query_map on the line above always hits. The router now takes the datasource from that SubQuery, with no fallback to invent.

The import cycle

It did appear, exactly as the review predicted, and it is fixed at the root
rather than papered over.

graph_utils imported ExecutionDAG through global_planner.schemas. That
loaded the whole pipeline.nodes package first, which happened to import
plan_cache in the one order that works. Importing nl2sql.execution.dag
directly exposed the real cycle:

  • plan_cache needs PlanModel from nodes.ast_planner.schemas
  • nodes.ast_planner.node needs PlanCache
  • and both nodes/__init__.py and nodes/ast_planner/__init__.py eagerly
    re-exported the node class, so importing the schemas ran the node.

Neither __init__ re-exports a node class now. Nothing in the tree imported
those re-exports (sql_agent.py was the single importer of
nodes.ast_planner.ASTPlannerNode and now names .node), and nothing in the
docs referenced them. A new test in tests/architecture/test_boundaries.py
imports nl2sql.pipeline.plan_cache on its own to keep it that way.

Known limitation (found here, fix deliberately out of scope)

Two identically-worded sub-queries collapse into one DAG node and fail the
run.
The decomposer gives each sub-query a content-addressed id
(_stable_id over intent, metrics, filters, group_by, order_by, limit and
expected_schema). Two sub-queries the model wrote identically therefore get the
same id, so the graph has one node where the combine group expects two: the
edges both land on it, the indegree bookkeeping never balances, and
ExecutionDAG._layered_toposort reports "ExecutionDAG contains a cycle". The
run ends with PLANNER_FAILED and a message that does not describe the actual
problem.

This is pre-existing — the global planner built the same DAG from the same
ids and failed the same way. This PR only makes it visible: the DAG is now
built one node earlier, so the case shows up in the decomposer's tests.
test_ops_after_a_real_combine_stay_post_combine had been feeding exactly this
input and never noticed, because nothing downstream of the decomposer ran in
it; its fixture now uses two genuinely different sub-queries, which is what the
test meant all along.

The real fix belongs in the decomposer — either de-duplicate sub-queries that
share a stable id, or make the id unique per emitted sub-query — and it is
not attempted here. What this PR does do is keep the failure reported
rather than raised: see below.

The DAG's own failure path, kept on purpose

Because of the above, a decomposition can be well-formed and still not describe
a runnable graph. The DAG build therefore keeps its own PLANNER_FAILED path,
now on the decomposer: the decomposition is returned, no execution_dag is,
and the layer router ends the run with that error as its cause — the same
outcome the global planner's error path produced.
test_a_dag_that_cannot_be_built_ends_the_run_with_its_error pins it end to
end through result_from_state, using the duplicate-id case as its trigger.

Timings, traces and the step list

  • tracing/recorder.py: NODE_INPUTS drops global_planner; the aggregator's
    inputs become ("execution_dag", "artifact_refs").
  • nl2sql/pipeline/steps.py (the Pipeline page's list, feat(playground): a Pipeline page showing what each step does #171): the
    global_planner step is gone and the decomposer's sentence says what it said.
    The drift test in tests/unit/test_pipeline_steps.py compares the list to the
    real compiled graphs and agrees again; test_playground_pipeline expects 13
    steps rather than 14.
  • tests/e2e/test_trace_fake_llm.py's node set drops it.
  • Nothing else names the node: git grep global_planner over the tree is empty.

Tests (re-run after the rebase onto merged #175)

  • pytest -m "not integration" -q: 1578 passed, 5 skipped
  • EMBEDDING_PROVIDER=local pytest -m "integration and not llm" -q: 72 passed, 3 skipped, which includes tier 1 at 129/129
  • npm test in web/playground: 106 passed (it reads the step list over /api/settings)
  • mkdocs build --strict: clean

Docs

docs/architecture/{overview,pipeline,determinism,failure_recovery,graph_state,invariants}.md,
docs/architecture/nodes/{index,decomposer_node,engine_aggregator_node}.md,
docs/architecture/subgraphs/main_pipeline_graph.md, docs/index.md, and
mkdocs.yml; docs/architecture/nodes/global_planner_node.md is deleted.
Every Mermaid diagram of the control graph loses the node.

… of its own node

`global_planner` read nothing but `state.decomposer_response`, so the DAG it
built was a pure function of the decomposer's output: a format conversion with
a graph node, a response model, a state field and a package around it. It is
now `decomposer/dag.py`, a function the decomposer calls, and `GraphState`
carries the `ExecutionDAG` itself rather than a one-field wrapper around it.

Most of what the node wrote was never read, and is gone with it:

* `LogicalNode.output_schema` (and `RelationSchema`/`ColumnSpec` with it) --
  the only reader was the planner copying a scan's schema onto the combine node;
* `ExecutionDAG.dag_id` and `.content_hash`, and the hash that produced them --
  no readers anywhere;
* all six scan-node attribute keys. Five had no reader; the sixth,
  `datasource_id`, was read on a branch that cannot be taken, because a scan
  node's `node_id` *is* its sub-query's id and so is always found in the
  router's `sub_query_map` on the line above. The router now looks the
  sub-query up and uses its datasource, with no fallback to invent.

`layer_router` is untouched: its empty body is the fan-in barrier for parallel
sub-queries, and `decomposer -> layer_router` replaces `decomposer ->
global_planner -> layer_router`.

The import cycle the review predicted did appear. `graph_utils` imported
`ExecutionDAG` through `global_planner.schemas`, which loaded the whole
`pipeline.nodes` package first and so happened to import `plan_cache` in the
one order that works. Importing `nl2sql.execution.dag` directly exposed the
real cycle: `plan_cache` needs `PlanModel` from `ast_planner.schemas`, and
`ast_planner.node` needs `PlanCache`, while both packages' `__init__` eagerly
re-exported the node class -- so importing the schemas ran the node. Fixed at
the root: neither `__init__` re-exports a node class any more (nothing imported
those re-exports; `sql_agent.py` now names `.node`), and a test imports
`nl2sql.pipeline.plan_cache` on its own to keep it that way.

One behaviour is preserved deliberately. A decomposition can be well-formed and
still not describe a runnable graph -- two sub-queries the model wrote
identically get the same content-addressed id -- so the DAG build keeps its own
`PLANNER_FAILED` path, now on the decomposer: the decomposition is returned,
no `execution_dag` is, and the layer router ends the run with that error as its
cause. `test_ops_after_a_real_combine_stay_post_combine` fed exactly that case
and never noticed, because nothing downstream of the decomposer ran in it.

`nl2sql/pipeline/steps.py` loses the `global_planner` step and the decomposer's
sentence gains what it said, so the Pipeline page's drift test agrees with the
graph again. The trace recorder's `NODE_INPUTS` drops `global_planner` and the
aggregator now reads `execution_dag`.
@nadeem4
nadeem4 force-pushed the refactor/dag-in-decomposer branch from 11f0b37 to 56e6ad5 Compare September 23, 2026 20:36
@nadeem4
nadeem4 merged commit d6fb1e1 into main Sep 23, 2026
6 checks passed
@nadeem4
nadeem4 deleted the refactor/dag-in-decomposer branch September 23, 2026 20:44
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant