Delta Query Architecture: Session API + Backend Abstraction for DD and nanoarrow

[justinjoy]

Delta Query Architecture Design

Status: Proposal (Consensus from Architect + Analyst + Planner + Critic review)
Date: 2026-03-03

---

1. Executive Summary

wirelog currently uses a one-shot batch execution model: each wl_dd_execute_cb() call constructs a new timely dataflow, loads all EDB data via new_collection_from(), runs to completion, and tears down. No state survives between executions.

This document proposes adding delta (incremental) query support where:

• Users can update input data (add/remove tuples) after initial execution
• The system propagates only the changes through the dataflow graph
• Only changed results are reported back (insertions and retractions)

The design targets both the current DD backend and a future nanoarrow (C11-native) backend, maximizing code reuse at the C API layer while keeping backend-specific implementations separate.

---

2. Current Architecture (Evidence-Based)

2.1 One-Shot Execution Lifecycle

From wirelog/cli/driver.c:132-219:

Parse -> Optimize -> DD Plan -> Marshal FFI -> Create Worker -> Load Facts -> Execute -> Destroy All

Every call to wl_run_pipeline() creates a fresh worker, loads all facts, executes, and destroys everything.

2.2 Worker is a Data Bag, Not a Compute Session

From rust/wirelog-dd/src/ffi.rs:

pub struct WlDdWorker {
    pub(crate) num_workers: u32,
    pub(crate) edb_data: HashMap<String, Vec<Vec<i64>>>,
    pub(crate) edb_ncols: HashMap<String, u32>,
}

WlDdWorker stores EDB data as owned vectors and passes them to timely::execute(). It holds no timely worker handle, no input sessions, and no probe.

2.3 Static Collections (No Update Mechanism)

From rust/wirelog-dd/src/dataflow.rs:136:

let (_, coll) = scope.new_collection_from(rows.clone());

The _ discards the InputSession handle. new_collection_from() creates an immutable snapshot. There is no mechanism to inject updates after initial loading.

2.4 Sequential Stratum Execution

From rust/wirelog-dd/src/dataflow.rs:72-92: strata are executed sequentially, each in its own worker.dataflow() scope, with results collected in a HashMap<String, Vec<Row>> between them. This architecture prevents cross-stratum change propagation in a persistent session.

---

3. Design Decisions (Consensus)

The following decisions were reached through architect/analyst/planner proposals reviewed by a critic agent.

3.1 Plan Naming: Keep wl_dd_*

Decision: Do NOT rename wl_dd_plan_t to wl_exec_plan_t.

Rationale: The plan structure encodes DD-specific operators (VARIABLE, MAP, FILTER, JOIN, ANTIJOIN, REDUCE, CONCAT, CONSOLIDATE, SEMIJOIN). A future nanoarrow backend would use different operators (sort-merge join, hash join). The plan is NOT backend-agnostic; the backend abstraction should operate at the session/execution level, not the plan level. This avoids touching 8+ files for zero functional gain.

3.2 Batch Mode: Keep Separate

Decision: The existing batch API (wl_dd_execute_cb) stays as-is. Delta is an addition, not a replacement. Batch is NOT reimplemented on top of sessions.

Rationale:

• The batch path is proven correct across 410 tests and 15 benchmarks
• new_collection_from() is simpler and faster than InputSession for all-at-once execution
• No regression risk to existing users

3.3 Set Semantics for Delta Output

Decision: Delta callback diff values are +1 or -1 only (never +2). The session internally uses distinct()/consolidate() to enforce set semantics.

Rationale: wirelog is a Datalog engine; Datalog has set semantics. Bag semantics would be a language-level change.

3.4 Session Thread Safety

Decision: Sessions are NOT thread-safe. All calls to a given session must come from the same thread.

Rationale: Matches C convention (no hidden locking), keeps implementation simple. If concurrent access is needed later, the caller wraps in a mutex.

3.5 String Interning

Decision: C side pre-interns. session_insert accepts int64_t* only. The session does not hold a reference to wl_intern_t.

Rationale: Keeps the FFI boundary clean (no C-to-Rust-to-C callbacks), matches the existing wl_dd_load_edb contract.

3.6 Multi-Worker Sessions

Decision: MVP is single-worker only. num_workers > 1 returns an error in Phase 1.

Rationale: Timely's multi-worker mode introduces inter-thread communication complexity orthogonal to core delta logic. Single-worker first, optimize later.

3.7 Recursive Aggregation Scope

Decision: Defer to Phase 1D design spike. Min/max are monotone and can be moved inside iterate(). Count/sum in recursive contexts may be explicitly unsupported in delta mode.

Rationale: Recursive aggregation is only well-defined for monotone aggregates over lattices. This is a known Datalog limitation.

---

4. C API Design

4.1 Backend vtable (wirelog/backend.h)

typedef struct wl_session wl_session_t;  /* opaque */

/**
 * Delta callback: invoked once per changed output tuple during step().
 * @diff: +1 for insertion, -1 for retraction.
 */
typedef void (*wl_on_delta_fn)(
    const char *relation,
    const int64_t *row,
    uint32_t ncols,
    int32_t diff,
    void *user_data
);

/**
 * Snapshot callback: invoked once per current output tuple.
 */
typedef void (*wl_on_tuple_fn)(
    const char *relation,
    const int64_t *row,
    uint32_t ncols,
    void *user_data
);

/**
 * wl_compute_backend_t:
 *
 * Virtual function table for execution backends.
 * Each backend implements session lifecycle, delta propagation,
 * and result retrieval.
 */
typedef struct {
    const char *name;  /* "dd", "nanoarrow", etc. */

    /* Session lifecycle */
    int (*session_create)(const wl_ffi_plan_t *plan,
                          uint32_t num_workers,
                          wl_session_t **out);
    void (*session_destroy)(wl_session_t *session);

    /* Data mutation (staged, not propagated until step) */
    int (*session_insert)(wl_session_t *session,
                          const char *relation,
                          const int64_t *data,
                          uint32_t num_rows,
                          uint32_t num_cols);
    int (*session_remove)(wl_session_t *session,
                          const char *relation,
                          const int64_t *data,
                          uint32_t num_rows,
                          uint32_t num_cols);

    /* Propagation */
    int (*session_step)(wl_session_t *session);

    /* Output */
    void (*session_set_delta_cb)(wl_session_t *session,
                                 wl_on_delta_fn callback,
                                 void *user_data);
    int (*session_snapshot)(wl_session_t *session,
                            wl_on_tuple_fn callback,
                            void *user_data);
} wl_compute_backend_t;

/* Backend registry */
const wl_compute_backend_t *wl_backend_dd(void);
/* const wl_compute_backend_t *wl_backend_nanoarrow(void);  future */

4.2 Session Convenience Wrappers (wirelog/session.h)

wl_session_t *
wl_session_create(const wl_compute_backend_t *backend,
                  const wl_ffi_plan_t *plan,
                  uint32_t num_workers);

void wl_session_destroy(wl_session_t *session);

int wl_session_insert(wl_session_t *session,
                      const char *relation,
                      const int64_t *data,
                      uint32_t num_rows,
                      uint32_t num_cols);

int wl_session_remove(wl_session_t *session,
                      const char *relation,
                      const int64_t *data,
                      uint32_t num_rows,
                      uint32_t num_cols);

int wl_session_step(wl_session_t *session);

void wl_session_set_delta_cb(wl_session_t *session,
                              wl_on_delta_fn callback,
                              void *user_data);

int wl_session_snapshot(wl_session_t *session,
                        wl_on_tuple_fn callback,
                        void *user_data);

4.3 Internal Session Structure

/* session.c (not public) */
struct wl_session {
    const wl_compute_backend_t *backend;  /* vtable pointer */
    void *backend_session;                /* opaque backend state */
    uint64_t epoch;                       /* logical timestamp */
};

4.4 Usage Example

/* Setup */
wl_ffi_plan_t *plan = /* ... marshal from program ... */;
const wl_compute_backend_t *be = wl_backend_dd();
wl_session_t *s = wl_session_create(be, plan, 1);

/* Load initial data */
int64_t edges[] = {1,2, 2,3, 3,4};
wl_session_insert(s, "edge", edges, 3, 2);
wl_session_set_delta_cb(s, my_delta_handler, ctx);
wl_session_step(s);  /* initial results via delta callback */

/* Incremental update */
int64_t new_edge[] = {4, 5};
wl_session_insert(s, "edge", new_edge, 1, 2);
wl_session_step(s);  /* only new derivations reported */

/* Deletion */
int64_t del_edge[] = {2, 3};
wl_session_remove(s, "edge", del_edge, 1, 2);
wl_session_step(s);  /* retractions reported as diff=-1 */

/* Cleanup */
wl_session_destroy(s);

---

5. Rust-Side Design (DD Backend)

5.1 Architecture Change

Current:  timely::execute() -> ephemeral dataflow -> step_while(done) -> drop
Target:   background thread -> persistent dataflow -> command loop -> step per epoch

5.2 Key Data Structures

enum SessionCommand {
    Insert(String, Vec<Vec<i64>>),
    Remove(String, Vec<Vec<i64>>),
    Step(mpsc::Sender<Result<(), String>>),
    Snapshot(mpsc::Sender<Result<HashMap<String, Vec<Vec<i64>>>, String>>),
    Shutdown,
}

pub struct DdSession {
    cmd_tx: mpsc::Sender<SessionCommand>,
    worker_thread: Option<thread::JoinHandle<()>>,
    delta_callback: Arc<Mutex<Option<DeltaCallbackInfo>>>,
}

5.3 Critical Changes

Current Code	Delta Change	Location
worker.dataflow::<(), _, _>()	worker.dataflow::<u64, _, _>()	dataflow.rs:130
new_collection_from(rows)	scope.new_input::<Vec<i64>>()	dataflow.rs:136
Discarded _ (InputSession)	Store in HashMap<String, InputSession>	dataflow.rs:136
worker.step_while(|| !probe.done())	worker.step_while(|| probe.less_than(&epoch))	dataflow.rs:165
timely::execute() returns immediately	Runs in background thread, loops on commands	dataflow.rs:69
inspect fires only for diff > 0	Fire for both diff > 0 and diff < 0	dataflow.rs:149-158
Strata in separate worker.dataflow() scopes	All strata in single persistent scope	dataflow.rs:74-92

5.4 New FFI Entry Points

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_create(
    plan: *const WlFfiPlan, num_workers: u32,
    out: *mut *mut DdSession) -> c_int;

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_destroy(session: *mut DdSession);

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_insert(
    session: *mut DdSession, relation: *const c_char,
    data: *const i64, num_rows: u32, num_cols: u32) -> c_int;

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_remove(
    session: *mut DdSession, relation: *const c_char,
    data: *const i64, num_rows: u32, num_cols: u32) -> c_int;

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_step(
    session: *mut DdSession) -> c_int;

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_set_delta_cb(
    session: *mut DdSession, cb: WlOnDeltaFn,
    user_data: *mut c_void);

#[no_mangle]
pub unsafe extern "C" fn wl_dd_session_snapshot(
    session: *mut DdSession, cb: WlOnTupleFn,
    user_data: *mut c_void) -> c_int;

---

6. DD-to-nanoarrow Mapping

Session API	DD Backend (Rust)	nanoarrow Backend (C11, future)
session_create	timely::execute() on background thread, build dataflow with InputSession handles	Allocate hash tables per relation, build operator pipeline from plan
session_insert	input_session.insert(row)	Insert tuples into delta buffer
session_remove	input_session.remove(row)	Insert negative tuples into delta buffer
session_step	advance_to(epoch+1), flush(), step_while(probe.less_than)	Semi-naive evaluation loop to fixpoint
session_snapshot	Iterate output arrangements, emit tuples with positive diff	Iterate materialized relation hash tables
session_destroy	Drop InputSessions, signal shutdown, join thread	Free all relation storage

6.1 Code Reuse Summary

Layer	Reusable?	Notes
Session C API (session.h)	Yes	Both backends implement the same vtable
Backend vtable (backend.h)	Yes	Dispatch mechanism is backend-agnostic
Plan structure (dd_plan.h, dd_ffi.h)	No	DD operators differ from nanoarrow operators
Dataflow execution (dataflow.rs)	No	DD-specific; nanoarrow implements in C11
Pipeline driver (driver.c)	Mostly	Upper layers (parse, optimize) shared; execution dispatch changes
Fact loading	Partially	wl_dd_load_edb is DD-specific; session_insert is backend-agnostic

---

7. Implementation Phases

Phase 0: Backend Abstraction (C-only, ~2 days)

Goal: Introduce vtable without changing behavior.

Task	Size	Files
Define wl_compute_backend_t vtable	S	NEW: wirelog/backend.h
Implement wl_session_t convenience wrappers	S	NEW: wirelog/session.h, wirelog/session.c
Wrap existing DD execution behind vtable	M	NEW: wirelog/backend_dd.c
Update meson.build	S	wirelog/meson.build

Validation: All 410 existing tests pass. Zero behavior change.

Phase 1A: Insert-Only Non-Recursive Delta (~4 days)

Goal: Minimum viable delta. Single non-recursive stratum, insert-only.

Task	Size	Files
Rust DdSession with background thread + mpsc	L	NEW: rust/wirelog-dd/src/session.rs
Persistent dataflow with InputSession (non-recursive only)	L	MODIFY: rust/wirelog-dd/src/dataflow.rs (new function, not modifying existing)
Timestamp type () -> u64 for session path	M	NEW code path in session.rs
FFI session entry points	M	MODIFY: rust/wirelog-dd/src/ffi.rs
Wire into C vtable	S	MODIFY: wirelog/backend_dd.c
Integration tests	M	NEW: tests/test_session.c

Validation: Insert facts, step, observe outputs. Insert more, step, observe deltas only.

Phase 1B: Multi-Stratum + Deletions (~3 days)

Goal: Full non-recursive delta support.

Task	Size	Files
Unify non-recursive strata into single persistent dataflow	L	MODIFY: rust/wirelog-dd/src/session.rs
Implement session_remove with negative diffs	M	MODIFY: rust/wirelog-dd/src/session.rs, ffi.rs
Implement session_snapshot	M	MODIFY: rust/wirelog-dd/src/session.rs, ffi.rs
Multi-stratum + deletion tests	M	MODIFY: tests/test_session.c

Validation: Multi-stratum programs, insert/delete/step sequences, snapshot consistency.

Phase 1C: Recursive Strata (No Aggregation) (~4 days)

Goal: Incremental transitive closure, reachability queries.

Task	Size	Files
Persistent iterate() scope with Variable collection	XL	MODIFY: rust/wirelog-dd/src/session.rs
TC and cycle tests with incremental edge changes	L	MODIFY: tests/test_session.c

Validation: TC on chain graph, add edge -> 3 new tuples. Delete edge -> 4 retractions.

Phase 1D: Recursive Aggregation (Design Spike, ~5 days)

Goal: Incremental connected components (min-label), SSSP.

Task	Size	Files
Design spike: monotone aggregation inside iterate()	L	Design document
Replace apply_post_reduce for min/max in session path	XL	MODIFY: rust/wirelog-dd/src/session.rs
Document: count/sum in recursive context unsupported in delta mode	S	This document
CC and SSSP tests with incremental changes	L	MODIFY: tests/test_session.c

Validation: CC with min-label propagation works incrementally. SSSP distance updates on edge changes.

Phase 2: nanoarrow Backend (Future, Separate Project)

Goal: C11-native semi-naive delta evaluator behind same vtable.

Out of scope for this document.

---

8. Dependency Graph

Phase 0 (backend abstraction)
    |
    v
Phase 1A (insert-only, non-recursive) -----> Phase 1B (deletions, multi-stratum)
                                                  |
                                                  v
                                              Phase 1C (recursive, no agg)
                                                  |
                                                  v
                                              Phase 1D (recursive agg - design spike)

Phase 2 (nanoarrow) depends on Phase 0 only

Parallelizable:

• Phase 0 and Rust prototype (Phase 1A Rust work) can overlap
• Phase 2 can start after Phase 0, independent of Phase 1

---

9. Risk Assessment

#	Risk	Severity	Mitigation
1	Timely background thread lifecycle. timely::execute() takes ownership and runs to completion. Keeping it alive requires careful shutdown coordination.	High	Prototype the background-thread + mpsc pattern in an isolated Rust binary BEFORE integrating into wirelog-dd.
2	Cross-stratum dataflow unification. Current code runs each stratum as separate worker.dataflow(). Persistent sessions need all strata in one scope.	High	Phase 1A restricts to single stratum. Phase 1B tackles multi-stratum non-recursive only.
3	Recursive iterate under incremental input. DD's iterate() with changing input requires Variable collections that re-trigger fixpoint.	High	Phase 1C is scoped to recursive WITHOUT aggregation (well-understood case).
4	Batch performance regression. Shared code modifications could regress the 15 benchmarks.	Medium	Batch path (execute_plan()) stays completely untouched. Session gets its own code path.
5	apply_post_reduce is batch-only. The current recursive aggregation workaround cannot work incrementally.	Medium	Phase 1D is a separate design spike. Min/max can be moved inside iterate (monotone). Count/sum may be unsupported.

---

10. Acceptance Criteria

10.1 Batch Regression (All Phases)

After delta support is added, all 410 existing tests (325 C + 85 Rust) must pass with zero changes:

• meson test -C build reports 0 failures
• cargo test reports 0 failures
• Benchmark performance within 5% of baseline

10.2 Delta Insert (Phase 1A)

Given r(X,Y) :- a(X,Y), b(Y,Z) with initial a={(1,2)}, b={(2,3)}:

• Initial step produces r={(1,2,3)}
• Insert a=(1,4), b=(4,5), step
• Delta callback receives exactly: r(1,4,5) with diff=+1

10.3 Delta Delete (Phase 1B)

Same setup, delete a=(1,2), step:

• Delta callback receives: r(1,2,3) with diff=-1

10.4 Recursive Insert (Phase 1C)

Given edge={(1,2),(2,3)} producing tc={(1,2),(2,3),(1,3)}:

• Insert edge=(3,4), step
• Delta: tc += {(3,4),(2,4),(1,4)} (exactly 3 insertions)

10.5 Recursive Delete (Phase 1C)

Given edge={(1,2),(2,3),(3,4)}, delete edge=(2,3), step:

• Delta: tc -= {(2,3),(1,3),(2,4),(1,4)} (exactly 4 retractions)

10.6 Session Cleanup

After wl_session_destroy(), all Rust-side memory is freed. No leaks detected by ASan.

---

11. Open Questions

These must be resolved before Phase 1A implementation begins:

1. Initial execution output: Should the first session_step() produce all results as diff=+1 deltas, or should there be a separate "initial load" mode?

2. Worker-session relationship: Can one wl_dd_worker_t host multiple sessions? MVP says no (1:1), but the long-term model needs clarification.

3. Rule changes: Can the execution plan change between delta updates? Current proposal says no (session is bound to a plan at creation). Rule changes require session recreation.

4. Memory budget: What is the maximum acceptable memory overhead for a persistent delta session vs. one-shot batch? Critical for embedded targets (<256MB per ARCHITECTURE.md).

5. Epoch exposure: Should the C API expose the logical timestamp / epoch counter, or keep it internal?

6. Async stepping: Should session_step() be synchronous (block until propagation completes) or support an async model?

7. vtable extensibility: How to add new operations to wl_compute_backend_t without breaking ABI? Version field? Sentinel NULL check?

---

12. Edge Cases

#	Scenario	Expected Behavior
1	Empty initial data + delta insert	Session works correctly; dataflow initialized with empty inputs
2	Delete non-existent tuple	No-op; no output (DD will produce -1 diff that cancels to 0 after consolidate)
3	Insert duplicate tuple	No-op under set semantics (distinct collapses duplicates)
4	Rapid advance with no changes	No output, no crash
5	Very large batch update (100K tuples)	DD handles via batched InputSession; latency acceptable
6	Update EDB not referenced by any rule	No-op, no output, no crash
7	Session outlives worker	Undefined behavior; document that session must be destroyed before worker
8	New string values in delta	Caller must pre-intern via wl_intern_put() before passing as i64
9	Recursive aggregation deletion (SSSP)	Phase 1D; min-distance may increase; apply_post_reduce must be replaced
10	Multiple steps without consuming delta results	Delta callback fires during each step; uncollected results are not accumulated

---

13. Conflict Resolution Log

Topic	Architect	Analyst	Planner	Critic Resolution
Plan naming	Rename to wl_exec_plan_t	N/A	Keep wl_dd_*	Keep wl_dd_* -- plan encodes DD-specific operators
Batch-on-session	Reimplement batch via session	N/A	Keep separate	Keep separate -- protects 410 tests from regression
Phasing	5 sub-phases (1a-1e)	4 phases (A-D)	3 phases (0-2)	Planner outer (0/1/2) + Analyst inner (A-D)
apply_post_reduce	Brief mention	Critical flag	Not addressed	Phase 1D design spike -- monotone agg inside iterate
Cross-stratum	Not addressed	Correctly identified	Not addressed	Phase 1B -- unify non-recursive strata first

---

14. References

Internal Code References

• wirelog/ffi/dd_ffi.h -- Current FFI boundary
• wirelog/ffi/dd_plan.h -- Plan structure (DD-specific operators)
• rust/wirelog-dd/src/dataflow.rs -- Rust DD execution (one-shot model)
• rust/wirelog-dd/src/ffi.rs -- Rust FFI entry points
• wirelog/cli/driver.c -- Pipeline driver
• wirelog/intern.h -- Symbol interning

Architecture Documents

• ARCHITECTURE.md -- ComputeBackend abstraction (section 2.3), nanoarrow plan (Phase 3)

External References

• Differential Dataflow: https://github.com/TimelyDataflow/differential-dataflow
• DD InputSession API: differential_dataflow::input::InputSession
• nanoarrow: https://github.com/apache/arrow-nanoarrow
• DRed algorithm (deletion in recursive Datalog): Gupta et al., "Maintaining Views Incrementally"

[justinjoy]

DD was dropped