Skip to content

Commit f12d8af

Browse files
zzylolclaude
andcommitted
feat(dag-viewer): read planner operators and flat logical DAGs
stage_pipeline now writes operators as {"NonASAP": {"Scan": …}} and the logical stages as flat DAGs (node i is nodes[i], children inside the operator). normalizeStagePipeline converts both to the {kind, …} payloads and explicit edges the viewer draws; documents already in that shape pass through unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW
1 parent 322d8df commit f12d8af

3 files changed

Lines changed: 94 additions & 0 deletions

File tree

‎tools/dag-viewer/app.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,7 @@
135135

136136
// Show one stage document; `source` names it, `story` explains it.
137137
function showDocument(data, source, story) {
138+
data = normalizeStagePipeline(data);
138139
const errors = validateStagePipeline(data);
139140
if (errors.length) throw new Error(errors.slice(0, 3).join('; '));
140141
doc = data;

‎tools/dag-viewer/stages.js‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,85 @@ function stageRootNodeId(root) {
2525
return null;
2626
}
2727

28+
// The planner writes operators as `{"NonASAP": {"Scan": {...}}}` or
29+
// `{"ASAP": {"SummaryAgg": {...}}}`, children as node ids inside the
30+
// operator, and logical stages as flat DAGs (node `i` is `nodes[i]`, no
31+
// edge list). The viewer reads `{kind, ...}` payloads and explicit edges;
32+
// these functions convert, and leave documents already in that shape alone.
33+
const STAGE_CHILD_FIELDS = ['child', 'children', 'left', 'right', 'summary_input', 'outer', 'inner'];
34+
const STAGE_RENAMED_KIND = { Join: 'join_kind', SetOp: 'set_kind', TimeRange: 'range_kind', PromqlSeriesSample: 'sample_kind' };
35+
36+
function stageSnake(name) {
37+
if (name === 'SQLWindowFunc') return 'sql_window_func';
38+
return name.replace(/([a-z0-9])([A-Z])/g, '$1_$2').toLowerCase();
39+
}
40+
41+
function stageOperatorParts(operator) {
42+
const [category, inner] = Object.entries(operator)[0];
43+
const [variant, fields] = typeof inner === 'string' ? [inner, {}] : Object.entries(inner)[0];
44+
const body = fields && typeof fields === 'object' && !Array.isArray(fields) ? { ...fields } : { value: fields };
45+
return [category, variant, body];
46+
}
47+
48+
function stagePayloadOf(operator) {
49+
if (!operator || typeof operator !== 'object' || typeof operator.kind === 'string') return operator;
50+
const [category, variant, body] = stageOperatorParts(operator);
51+
if ('kind' in body) { body[STAGE_RENAMED_KIND[variant] || 'variant_kind'] = body.kind; delete body.kind; }
52+
return category === 'NonASAP'
53+
? { kind: 'relational', operator: { kind: stageSnake(variant), ...body } }
54+
: { kind: stageSnake(variant), ...body };
55+
}
56+
57+
// Node ids an operator reads: its inputs, then scalar-subquery references.
58+
function stageOperatorChildren(operator) {
59+
const [, , body] = stageOperatorParts(operator);
60+
const ids = [];
61+
STAGE_CHILD_FIELDS.forEach((field) => {
62+
const value = body[field];
63+
if (typeof value === 'number') ids.push(value);
64+
else if (Array.isArray(value)) value.forEach((id) => { if (typeof id === 'number') ids.push(id); });
65+
});
66+
const walk = (value) => {
67+
if (!value || typeof value !== 'object') return;
68+
if (typeof value.ScalarSubquery === 'number') ids.push(value.ScalarSubquery);
69+
Object.values(value).forEach(walk);
70+
};
71+
Object.entries(body).forEach(([key, value]) => { if (!STAGE_CHILD_FIELDS.includes(key)) walk(value); });
72+
return ids;
73+
}
74+
75+
function normalizeStageDag(dag) {
76+
if (!dag || !Array.isArray(dag.nodes)) return dag;
77+
if (dag.nodes.every((node) => node && node.operator && !node.payload)) {
78+
const edges = [];
79+
const nodes = dag.nodes.map((node, id) => {
80+
stageOperatorChildren(node.operator).forEach((producer) => edges.push({ producer, consumer: id }));
81+
return {
82+
id,
83+
payload: stagePayloadOf(node.operator),
84+
result_kind: node.result_kind,
85+
output_schema: node.schema,
86+
guarantee: node.guarantee,
87+
timing: node.timing,
88+
};
89+
});
90+
return { ...dag, nodes, edges };
91+
}
92+
return { ...dag, nodes: dag.nodes.map((node) => (node && node.payload ? { ...node, payload: stagePayloadOf(node.payload) } : node)) };
93+
}
94+
95+
function normalizeStagePipeline(doc) {
96+
if (!isStagePipelineDocument(doc)) return doc;
97+
const candidates = (stage) => (stage && Array.isArray(stage.candidates)
98+
? { ...stage, candidates: stage.candidates.map((c) => (c ? { ...c, dag: normalizeStageDag(c.dag) } : c)) }
99+
: stage);
100+
const out = { ...doc };
101+
if (doc.stage0_logical) out.stage0_logical = { ...doc.stage0_logical, dag: normalizeStageDag(doc.stage0_logical.dag) };
102+
if (doc.stage1_logical_asap !== undefined) out.stage1_logical_asap = candidates(doc.stage1_logical_asap);
103+
if (doc.stage2_physical_asap !== undefined) out.stage2_physical_asap = candidates(doc.stage2_physical_asap);
104+
return out;
105+
}
106+
28107
function validateStageDag(dag, where, physical, queryCount) {
29108
const errors = [];
30109
if (!dag || typeof dag !== 'object') return [`${where}: dag is missing`];

‎tools/dag-viewer/test_viewer.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,20 @@ def test_fixture_is_a_valid_document(self):
7575
"""The committed sample passes the viewer's own shape validation."""
7676
self.assertEqual(self.validate(self.doc), [])
7777

78+
def test_planner_document_is_valid_after_normalizing(self):
79+
"""`stage_pipeline` writes operators as `{"NonASAP": {"Scan": ...}}`
80+
and logical stages as flat DAGs; normalized, they validate, and a
81+
flat DAG gets one edge per operator input."""
82+
doc = json.loads((HERE / "examples" / "planner-layering-example1.json").read_text())
83+
normalized = self.js.call("normalizeStagePipeline", doc)
84+
self.assertEqual(self.validate(normalized), [])
85+
stage0 = normalized["stage0_logical"]["dag"]
86+
self.assertEqual(stage0["nodes"][0]["payload"]["kind"], "relational")
87+
self.assertEqual(stage0["nodes"][0]["payload"]["operator"]["kind"], "scan")
88+
self.assertTrue(stage0["edges"])
89+
self.assertTrue(all(e["producer"] < e["consumer"] for e in stage0["edges"]))
90+
self.assertEqual(self.js.call("normalizeStagePipeline", self.doc), self.doc)
91+
7892
def test_partial_documents_are_valid(self):
7993
"""A run that stopped after Stage 1 or Stage 2 still loads."""
8094
through_stage1 = self.copy()

0 commit comments

Comments
 (0)