refactor(pipeline): build the execution DAG in the decomposer instead of its own node - #177
Merged
Merged
Conversation
… 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
force-pushed
the
refactor/dag-in-decomposer
branch
from
September 23, 2026 20:36
11f0b37 to
56e6ad5
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
global_plannerread nothing from state butdecomposer_response(
global_planner/node.py:34-37), so theExecutionDAGit produced was a purefunction 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, andGraphStatecarries theExecutionDAGitself instead of a one-field wrapper.layer_routeris untouched — its empty body is the fan-in barrier for parallelsub-queries.
Rebased onto
mainafter #175 (the ordinal removal) landed.Removed, and what read it
global_plannernode, package andGlobalPlannerResponsegraph.py,routes.py,aggregator/node.py,state.py— all now readstate.execution_dagGraphState.global_planner_responseexecution_dag: Optional[ExecutionDAG]LogicalNode.output_schema, andRelationSchema/ColumnSpecwith itExecutionDAG.dag_id,.content_hash, and_hash_execution_dagJsonLiteralattributeskeysdatasource_id, was read atroutes.py:74on a branch that cannot be taken: a scan node'snode_idis its sub-query's id, sosub_query_mapon the line above always hits. The router now takes the datasource from thatSubQuery, 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_utilsimportedExecutionDAGthroughglobal_planner.schemas. Thatloaded the whole
pipeline.nodespackage first, which happened to importplan_cachein the one order that works. Importingnl2sql.execution.dagdirectly exposed the real cycle:
plan_cacheneedsPlanModelfromnodes.ast_planner.schemasnodes.ast_planner.nodeneedsPlanCachenodes/__init__.pyandnodes/ast_planner/__init__.pyeagerlyre-exported the node class, so importing the schemas ran the node.
Neither
__init__re-exports a node class now. Nothing in the tree importedthose re-exports (
sql_agent.pywas the single importer ofnodes.ast_planner.ASTPlannerNodeand now names.node), and nothing in thedocs referenced them. A new test in
tests/architecture/test_boundaries.pyimports
nl2sql.pipeline.plan_cacheon 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_idover intent, metrics, filters, group_by, order_by, limit andexpected_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_toposortreports"ExecutionDAG contains a cycle". Therun ends with
PLANNER_FAILEDand a message that does not describe the actualproblem.
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_combinehad been feeding exactly thisinput 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_FAILEDpath,now on the decomposer: the decomposition is returned, no
execution_dagis,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_errorpins it end toend through
result_from_state, using the duplicate-id case as its trigger.Timings, traces and the step list
tracing/recorder.py:NODE_INPUTSdropsglobal_planner; the aggregator'sinputs 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): theglobal_plannerstep is gone and the decomposer's sentence says what it said.The drift test in
tests/unit/test_pipeline_steps.pycompares the list to thereal compiled graphs and agrees again;
test_playground_pipelineexpects 13steps rather than 14.
tests/e2e/test_trace_fake_llm.py's node set drops it.git grep global_plannerover the tree is empty.Tests (re-run after the rebase onto merged #175)
pytest -m "not integration" -q: 1578 passed, 5 skippedEMBEDDING_PROVIDER=local pytest -m "integration and not llm" -q: 72 passed, 3 skipped, which includes tier 1 at 129/129npm testinweb/playground: 106 passed (it reads the step list over/api/settings)mkdocs build --strict: cleanDocs
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, andmkdocs.yml;docs/architecture/nodes/global_planner_node.mdis deleted.Every Mermaid diagram of the control graph loses the node.