Skip to content

Commit fdfbd44

Browse files
committed
Refactor DSL to use nested JSON graph construction
Fixes #42 DSL functions now build stateless nested JSON structures instead of writing to df.nodes during construction. This eliminates orphaned nodes, enables transaction-safe graph building, and simplifies df.explain(). Key changes: - Durofut now embeds children as Box<Durofut> instead of ID references - Durofut no longer includes a node_id field; IDs are generated when writing to the database, making Durofut a pure data structure without database concerns. - df.start() recursively inserts all nodes in a single transaction - df.explain() parses nested JSON directly without temp tables - Add E2E test 26_graph_reuse.sql to verify graph storage and reuse
1 parent e269310 commit fdfbd44

11 files changed

Lines changed: 1319 additions & 413 deletions

USER_GUIDE.md

Lines changed: 22 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,19 @@ Functions are persisted to disk. If PostgreSQL crashes:
138138
- In-progress steps resume from the last checkpoint
139139
- Pending steps execute when the server restarts
140140

141+
### Graph Construction
142+
143+
DSL functions build graph structures **in memory** without touching the database. Only when you call `df.start()` are the nodes written to the database:
144+
145+
```sql
146+
-- This creates a JSON string representing the graph.
147+
SELECT 'SELECT 1' ~> 'SELECT 2';
148+
-- Returns: {"node_type":"THEN","left_node":{"node_type":"SQL","query":"SELECT 1"},"right_node":{"node_type":"SQL","query":"SELECT 2"}}
149+
150+
-- Only df.start() writes to the database
151+
SELECT df.start('SELECT 1' ~> 'SELECT 2');
152+
```
153+
141154
---
142155

143156
## DSL Reference
@@ -1097,15 +1110,15 @@ SQL |=> 'step1': SELECT 1 ✓ Completed
10971110
**2. Dry-Run Preview** - Pass a DSL expression to visualize without executing:
10981111

10991112
```sql
1100-
SELECT df.explain($$
1113+
SELECT df.explain(
11011114
'SELECT 1' |=> 'a'
11021115
~> 'SELECT 2' |=> 'b'
11031116
~> df.if(
11041117
'SELECT $a > 0',
11051118
'SELECT ''yes''',
11061119
'SELECT ''no'''
11071120
)
1108-
$$);
1121+
);
11091122
```
11101123

11111124
Output shows the graph structure:
@@ -1133,7 +1146,7 @@ SQL |=> 'a': SELECT 1
11331146
**ETL Pipeline with Parallel Validation:**
11341147

11351148
```sql
1136-
SELECT df.explain($$
1149+
SELECT df.explain(
11371150
'SELECT * FROM staging WHERE status = ''pending'' LIMIT 1' |=> 'record'
11381151
~> df.if(
11391152
'SELECT $record IS NOT NULL',
@@ -1150,7 +1163,7 @@ SELECT df.explain($$
11501163
),
11511164
'SELECT ''no pending records'''
11521165
)
1153-
$$);
1166+
);
11541167
```
11551168

11561169
Output:
@@ -1177,7 +1190,7 @@ SQL |=> 'record': SELECT * FROM staging WHERE status = 'pending' LIMIT 1
11771190
**Cron Job with Cleanup Loop:**
11781191

11791192
```sql
1180-
SELECT df.explain($$
1193+
SELECT df.explain(
11811194
df.loop(
11821195
df.wait_for_schedule('0 * * * *')
11831196
~> 'DELETE FROM logs WHERE created_at < now() - interval ''7 days''' |=> 'deleted'
@@ -1187,7 +1200,7 @@ SELECT df.explain($$
11871200
'SELECT ''nothing to clean'''
11881201
)
11891202
)
1190-
$$);
1203+
);
11911204
```
11921205

11931206
Output:
@@ -1207,7 +1220,7 @@ LOOP
12071220

12081221
```sql
12091222
-- Visualize the daily-order-archive function before starting it
1210-
SELECT df.explain($$
1223+
SELECT df.explain(
12111224
df.loop(
12121225
df.wait_for_schedule('0 0 * * *')
12131226
~> 'SELECT COUNT(*) as cnt FROM playground.orders
@@ -1224,7 +1237,7 @@ SELECT df.explain($$
12241237
VALUES (''No orders to archive'')'
12251238
)
12261239
)
1227-
$$);
1240+
);
12281241
```
12291242

12301243
Output:
@@ -1655,7 +1668,7 @@ SELECT df.signal('inst_id', 'approval', '{}'); -- send signal
16551668
16561669
-- Visualize
16571670
SELECT df.explain('instance_id'); -- live instance
1658-
SELECT df.explain($$ 'a' ~> 'b' $$); -- dry-run preview
1671+
SELECT df.explain('a' ~> 'b'); -- dry-run preview
16591672
16601673
-- Monitor
16611674
SELECT * FROM df.list_instances();

docs/ARCHITECTURE.md

Lines changed: 98 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -186,42 +186,47 @@ pub extern "C-unwind" fn _PG_init() {
186186

187187
#### Durofut (Durable Future Reference)
188188

189-
A `Durofut` represents a node in the function graph. It's serialized as JSON and passed between DSL functions.
189+
A `Durofut` represents an abstract function graph, sub-graph or leaf node. It's serialized as JSON and passed between DSL functions.
190190

191191
```rust
192192
// src/types.rs
193193
pub struct Durofut {
194-
pub node_id: String, // 8-char hex ID (e.g., "a1b2c3d4")
195-
pub node_type: String, // SQL, THEN, IF, JOIN, LOOP, etc.
196-
pub left_node: Option<String>, // Left child node ID
197-
pub right_node: Option<String>,// Right child node ID
198-
pub query: Option<String>, // SQL query or config JSON
199-
pub result_name: Option<String>, // Named result (from |=> operator)
194+
pub node_type: String, // SQL, THEN, IF, JOIN, LOOP, etc.
195+
pub left_node: Option<Box<Durofut>>, // Embedded left child
196+
pub right_node: Option<Box<Durofut>>, // Embedded right child
197+
pub query: Option<String>, // SQL query or config JSON
198+
pub result_name: Option<String>, // Named result (from |=> operator)
200199
}
201200
```
202201

203202
When serialized to JSON:
204203
```json
205204
{
206-
"node_id": "a1b2c3d4",
207-
"node_type": "SQL",
208-
"query": "SELECT 1"
205+
"node_type": "THEN",
206+
"left_node": {
207+
"node_type": "SQL",
208+
"query": "SELECT 1"
209+
},
210+
"right_node": {
211+
"node_type": "SQL",
212+
"query": "SELECT 2"
213+
}
209214
}
210215
```
211216

212217
#### FunctionNode (Database Representation)
213218

214-
Nodes are persisted in `df.nodes`:
219+
`df.start(<function>)` adds a new row to `df.instances`, then iterates the nodes in the function graph bottom up, persisting each one to the `df.nodes` table along with the instance ID, a new ID for the node, and the IDs of its child nodes, if any.
215220

216221
```sql
217222
CREATE TABLE df.nodes (
218223
id VARCHAR(8) PRIMARY KEY,
219-
instance_id VARCHAR(8), -- Set by df.start() during linking
224+
instance_id VARCHAR(8), -- Set by df.start()
220225
node_type TEXT NOT NULL, -- SQL, THEN, IF, JOIN, LOOP, etc.
221226
query TEXT, -- SQL query or config JSON
222227
result_name TEXT, -- Named result for $variable substitution
223-
left_node VARCHAR(8), -- Left child
224-
right_node VARCHAR(8), -- Right child
228+
left_node VARCHAR(8), -- Left child ID
229+
right_node VARCHAR(8), -- Right child ID
225230
status TEXT DEFAULT 'pending',
226231
result JSONB,
227232
created_at TIMESTAMPTZ DEFAULT now()
@@ -230,24 +235,20 @@ CREATE TABLE df.nodes (
230235

231236
### DSL Functions
232237

233-
Each DSL function (`df.sql`, `df.sleep`, `df.join`, etc.) creates a node and returns its JSON representation.
238+
Each DSL function (`df.sql`, `df.sleep`, `df.join`, etc.) creates a Durofut and returns its JSON representation. All graph construction is stateless.
234239

235240
#### Example: `df.sql()`
236241

237242
```rust
238243
// src/dsl.rs
239244
#[pg_extern(schema = "df")]
240245
pub fn sql(query: &str) -> String {
241-
let durofut = Durofut {
242-
node_id: short_id(), // Generate 8-char hex ID
246+
Durofut {
243247
node_type: "SQL".to_string(),
244-
left_node: None,
245-
right_node: None,
246-
query: Some(query.to_string()), // Store the SQL query
247-
result_name: None,
248-
};
249-
durofut.insert_node(); // INSERT INTO df.nodes
250-
durofut.to_json() // Return JSON for chaining
248+
query: Some(query.to_string()),
249+
..Default::default()
250+
}
251+
.to_json()
251252
}
252253
```
253254

@@ -259,16 +260,13 @@ pub fn then_fn(a: &str, b: &str) -> String {
259260
let a_fut = Durofut::ensure(a); // Auto-wrap plain SQL if needed
260261
let b_fut = Durofut::ensure(b);
261262

262-
let durofut = Durofut {
263-
node_id: short_id(),
263+
Durofut {
264264
node_type: "THEN".to_string(),
265-
left_node: Some(a_fut.node_id), // First step
266-
right_node: Some(b_fut.node_id), // Second step
267-
query: None,
268-
result_name: None,
269-
};
270-
durofut.insert_node();
271-
durofut.to_json()
265+
left_node: Some(Box::new(a_fut)), // Embed first step
266+
right_node: Some(Box::new(b_fut)), // Embed second step
267+
..Default::default()
268+
}
269+
.to_json()
272270
}
273271
```
274272

@@ -283,25 +281,18 @@ impl Durofut {
283281
if Self::is_durofut(s) {
284282
Self::from_json(s) // Already a Durofut
285283
} else {
286-
// Plain SQL string - create a SQL node
287-
let fut = Durofut {
288-
node_id: short_id(),
284+
// Plain SQL string
285+
Durofut {
289286
node_type: "SQL".to_string(),
290287
query: Some(s.to_string()),
291288
..Default::default()
292-
};
293-
fut.insert_node();
294-
fut
289+
}
295290
}
296291
}
297292

298293
pub fn is_durofut(s: &str) -> bool {
299-
// Check if valid JSON with 8-char hex node_id
300-
if let Ok(fut) = serde_json::from_str::<Durofut>(s) {
301-
fut.node_id.len() == 8 && fut.node_id.chars().all(|c| c.is_ascii_hexdigit())
302-
} else {
303-
false
304-
}
294+
// Check if valid JSON that can be deserialized as Durofut
295+
serde_json::from_str::<Durofut>(s).is_ok()
305296
}
306297
}
307298
```
@@ -344,52 +335,81 @@ CREATE OPERATOR !> (FUNCTION = df.if_else_op, ...);
344335
CREATE OPERATOR @> (FUNCTION = df.loop_prefix_op, RIGHTARG = text);
345336
```
346337

347-
### Node Linking
338+
### Node Insertion
348339

349-
When `df.start()` is called, it recursively links all nodes to the instance:
340+
When `df.start()` is called, it recursively inserts all nodes from the nested graph into the database:
350341

351342
```rust
352343
// src/dsl.rs - df.start()
353344
pub fn start(fut: &str, label: Option<&str>) -> String {
354345
let durofut = Durofut::ensure(fut);
355346
let instance_id = short_id();
356347

357-
// Create instance record
358-
Spi::run(&format!(
359-
"INSERT INTO df.instances (id, label, root_node, status)
360-
VALUES ('{}', {}, '{}', 'pending')",
361-
instance_id, label_sql, durofut.node_id
362-
));
363-
364-
// Recursively link all nodes in the graph
365-
fn link_nodes(node_id: &str, instance_id: &str, visited: &mut HashSet<String>) {
366-
if visited.contains(node_id) { return; }
367-
visited.insert(node_id.to_string());
368-
369-
// Set instance_id on this node
370-
Spi::run(&format!(
371-
"UPDATE df.nodes SET instance_id = '{}' WHERE id = '{}'",
372-
instance_id, node_id
373-
));
374-
375-
// Get child nodes and recurse
376-
let left = get_left_node(node_id);
377-
let right = get_right_node(node_id);
378-
let config = get_config(node_id);
348+
// Recursively insert all nodes from the nested graph
349+
// Note: No HashSet needed - nested graphs are trees, not DAGs
350+
fn insert_nodes(node: &Durofut, instance_id: &str) -> String {
351+
let node_id = short_id(); // Generate ID at insertion time
379352

380-
if let Some(l) = left { link_nodes(&l, instance_id, visited); }
381-
if let Some(r) = right { link_nodes(&r, instance_id, visited); }
353+
// Recursively insert children FIRST to get their IDs
354+
let left_id = node.left_node.as_ref().map(|n| insert_nodes(n, instance_id));
355+
let right_id = node.right_node.as_ref().map(|n| insert_nodes(n, instance_id));
382356

383-
// Handle extra nodes in config (e.g., condition_node, extra_nodes)
384-
if let Some(cfg) = config {
385-
if let Some(cond_id) = cfg["condition_node"].as_str() {
386-
link_nodes(cond_id, instance_id, visited);
357+
// Process config JSON to replace embedded Durofuts with IDs
358+
// (for IF condition_node, LOOP condition_node, JOIN3 extra_nodes)
359+
let query_escaped = if let Some(ref query_str) = node.query {
360+
if let Ok(mut config) = serde_json::from_str::<serde_json::Value>(query_str) {
361+
// For IF/LOOP nodes: replace condition_node Durofut with ID
362+
if node.node_type == "IF" || node.node_type == "LOOP" {
363+
if let Some(cond_json) = config.get("condition_node") {
364+
if let Ok(cond_node) = serde_json::from_value::<Durofut>(cond_json.clone()) {
365+
let cond_id = insert_nodes(&cond_node, instance_id);
366+
config["condition_node"] = serde_json::json!(cond_id);
367+
}
368+
}
369+
}
370+
// For JOIN3 nodes: replace extra_nodes Durofuts with IDs
371+
if node.node_type == "JOIN" {
372+
if let Some(extras) = config.get("extra_nodes").and_then(|e| e.as_array()) {
373+
let extra_ids: Vec<String> = extras.iter()
374+
.filter_map(|e| serde_json::from_value::<Durofut>(e.clone()).ok())
375+
.map(|n| insert_nodes(&n, instance_id))
376+
.collect();
377+
if !extra_ids.is_empty() {
378+
config["extra_nodes"] = serde_json::json!(extra_ids);
379+
}
380+
}
381+
}
382+
format!("'{}'", serde_json::to_string(&config).unwrap().replace('\'', "''"))
383+
} else {
384+
format!("'{}'", query_str.replace('\'', "''"))
387385
}
388-
// ... handle extra_nodes for join3, etc.
389-
}
386+
} else {
387+
"NULL".to_string()
388+
};
389+
390+
// Insert this node with all fields
391+
Spi::run(&format!(
392+
"INSERT INTO df.nodes
393+
(id, instance_id, node_type, query, result_name, left_node, right_node)
394+
VALUES ('{}', '{}', '{}', {}, {}, {}, {})",
395+
node_id, instance_id, node.node_type,
396+
query_escaped,
397+
escape_option(&node.result_name),
398+
escape_option(&left_id),
399+
escape_option(&right_id)
400+
));
401+
402+
node_id // Return the generated ID
390403
}
391404

392-
link_nodes(&durofut.node_id, &instance_id, &mut HashSet::new());
405+
let root_node_id = insert_nodes(&durofut, &instance_id);
406+
407+
// Create instance record with the root node ID
408+
Spi::run(&format!(
409+
"INSERT INTO df.instances (id, label, root_node, status)
410+
VALUES ('{}', {}, '{}', 'pending')",
411+
instance_id, label_sql, root_node_id
412+
));
393413

394414
// Capture variables and enqueue to duroxide
395415
let vars = capture_vars(); // SELECT * FROM df.vars

0 commit comments

Comments
 (0)