Engine Internals

This guide explains how weevr's execution engine turns configuration into running Spark pipelines. It covers the planning, execution, and result aggregation subsystems.

Introduction

The engine is responsible for a single question: given a set of configured threads, in what order should they run, and how should each thread's pipeline execute? The answer involves dependency analysis, parallel scheduling, caching, conditional execution, and structured result collection.

Key concepts

• ExecutionPlan — An immutable snapshot of thread ordering, dependency edges, parallel execution groups, cache targets, lookup dependency mappings (lookup_producers, lookup_consumers), and a lookup_schedule that assigns each lookup to its materialization point. Produced by the planner before any data is read. In notebooks, an ExecutionPlan renders automatically via _repr_html_(). Call plan.dag() to get a DAGDiagram SVG that can be saved with dag.save("plan.svg"). For plan mode results, result.dag() returns either a single-weave DAG or a loom-level swimlane diagram depending on how many weaves are present.
• DAG — A directed acyclic graph of thread dependencies. Dependencies are inferred from source/target path overlap and can also be declared explicitly.
• Parallel execution group — A set of threads with no mutual dependencies that can run concurrently on the same SparkSession.
• Cache target — A thread whose output is read by two or more downstream threads. The engine persists the post-write snapshot under the target's canonical identity (alias, path, and connection-declared forms all map to one key) and serves consumers from an explicit registry — a registry miss simply falls back to a direct Delta read. The snapshot materializes eagerly at registration, so parallel consumers in one group never race to compute it.

How it works

Execution follows a three-stage pipeline: plan → execute → aggregate.

Planning

The planner builds a DAG from thread configurations:

1. Check target ownership — Each target may have only one writing thread, unless one of two exemptions applies. Explicit dependencies that fully serialize the writers into a chain are one; downstream consumers then depend on the last writer, so they always see the target's final state. The other: every writer of the target carries a thread-entry condition. Declaring conditions on all writers asserts they are mutually exclusive — at most one runs per execution (say, a full-load thread and an incremental thread for the same table). The planner cannot prove the conditions exclude each other (they may read live table state), so the engine enforces the assertion at runtime: if a second writer's condition also evaluates true, the weave aborts before that writer executes. Consumers of such a target depend on every writer, and a lookup reading it materializes only after the last writer's group. Unordered writers sharing a target where any writer lacks a condition are rejected with a ConfigError naming the target and every writer — running them concurrently would corrupt the table.
2. Infer dependencies — If thread B reads from thread A's target path, B depends on A. This happens automatically by matching source paths/aliases to target paths/aliases across all threads.
3. Merge explicit dependencies — Weave entries can declare additional dependencies that are not data-based (e.g., ordering constraints).
4. Detect cycles — DFS with three-color marking (white/gray/black) finds back edges. If a cycle exists, execution fails immediately with a clear error showing the cycle path.
5. Topological sort — Kahn's algorithm produces a sequence of parallel groups. Threads within a group have no mutual dependencies and can run concurrently.
6. Analyze cache targets — Threads with two or more downstream dependents are flagged for caching, unless the thread explicitly sets cache: false. Setting cache: true is a no-op — it only prevents cache: false suppression but does not force caching for threads with fewer than two dependents.

The result is an immutable ExecutionPlan that the executor consumes without modification.

Execution

Thread execution is orchestrated at two levels:

Loom level (execute_loom) — Iterates weaves in declared order. Each weave gets its own plan and executor pass. Weave-level conditions are evaluated before planning; a false condition skips the entire weave. If a weave's status is "failure", the loom stops executing further weaves — remaining weaves are skipped.

Weave level (execute_weave) — Orchestrates the full weave lifecycle:

1. Initialize variables — A VariableContext is created from the weave's variables block. Variables are available to hook steps via ${var.name} placeholders.
2. Pre-steps — If the weave defines pre_steps, hook steps run before any materialization. Failures with on_failure: abort stop the weave. Pre-steps can set variables that downstream materialization and threads observe.
3. Lookup materialization (external) — External lookups (those not produced by a thread in the weave) are materialized after pre-steps and before any thread runs. Internal lookups whose source is produced by a thread in group N are deferred and materialized at the group N+1 boundary — after their producer completes. Materialization reads the source exactly once: the narrowed frame persists first and both the warming count and the unique_key check run against the cache. A strategy: broadcast lookup persists and carries the broadcast hint — consuming joins build their exchange from the resident cache instead of re-reading the source per join, so its memory profile is "resident cache + per-join exchange" rather than a fresh exchange per join. Non-materialized (materialize: false) lookups keep their per-thread re-read, but a declared unique_key validates once per weave — the first consumer validates (concurrent group-mates block on that outcome), and a failure outcome is observed consistently by every later consumer.
4. Column set materialization — Column sets resolve to from→to mapping dicts once per declaring scope: loom-level sets resolve at loom start and pass down as resolved mappings; a weave materializes only its own declarations (including any overrides of loom sets — precedence is unchanged); a thread materializes only thread-level declarations and merges the already-resolved parent mappings. Delta/YAML sources are read via read_source(); param sources resolve from runtime parameters; connection-backed sets resolve through the merged loom + weave connection dict. Results are captured in WeaveTelemetry.column_set_results.
5. Thread execution — Parallel groups are processed sequentially. At each group's top, thread conditions evaluate (fresh per group, memoized within it); threads whose condition is false are skipped without cascading to their dependents. Where condition-gated writers share a target, survivors are tracked as conditions resolve: a second surviving writer means the conditions overlap, and the weave aborts before that writer executes. In the same group this happens before any write; across groups the earlier writer's complete, sequential write stands, and the error says so — the abort exists to prevent the second write, not to undo the first. Remaining threads are then submitted to a ThreadPoolExecutor for concurrent execution. Threads reference lookups via source.lookup and receive the cached DataFrame.
6. Post-steps — After all threads complete, post_steps hook steps run. Failures with on_failure: abort mark the weave as failed.

After each thread completes:

• If it is a cache target, the output DataFrame is persisted.
• Consumer reference counts are decremented; caches are unpersisted when no consumers remain.
• Thread-level telemetry collectors are merged into the weave collector.

Thread level (execute_thread) — A single thread runs through an 18-step pipeline:

1. Initialize thread-level resources (variables, lookups, column_sets)
2. Run thread-level pre_steps
3. Resolve lookup-based sources from cached or on-demand DataFrames
4. Read all remaining declared sources (with watermark/CDC filtering if applicable). For mode: incremental_watermark, the primary source is filtered against the prior HWM and the new HWM is captured from the filtered DataFrame before transforms. For generic mode: cdc (explicit cdc.operation_column) composed with a watermark_column, the same pattern applies and the HWM aggregate runs before I/U/D operation routing — delete rows participate in advancing the window so append-only CDC history tables (e.g. SAP Open Database Mirror) do not reprocess deletes on every run. Empty filtered windows yield new_hwm = None so step 14 leaves the persisted HWM untouched. The Delta CDF preset path (cdc.preset: delta_cdf) is unchanged: it captures _commit_version instead of a column-based watermark and rejects watermark_column at config-parse time. When load.watermark_format is set, both reader paths wrap the column in to_timestamp/to_date(col, format) so string-typed watermark columns (SAP, mainframe, JSON) can be parsed declaratively.
5. Set the primary (first) source as the working DataFrame
6. Run pipeline steps against the working DataFrame
7. Evaluate validation rules; quarantine or abort on failures
8. Apply column and table naming normalization, including reserved word protection (if configured). Seven strategies are available: quote, prefix, suffix, error, rename, revert, and drop. See configuration keys for details.
9. Compute business keys and change detection hashes
10. Resolve the target write path
11. Apply target column mapping
12. Inject audit columns (bypasses mapping mode; applied for all write modes)
13. Write to the Delta target (standard write or CDC merge routing). For fact targets, the sentinel advisory then runs against the written table in a single pass (warnings only — never blocking)
14. Persist watermark or CDC state
15. Run post-write assertions
16. Write exports (secondary outputs, if configured)
17. Run thread-level post_steps
18. Build telemetry and return ThreadResult

Failure handling

Each thread has a configurable failure policy:

Policy	Behavior
abort_weave	All remaining threads in the weave are skipped (default)
skip_downstream	Only transitive dependents of the failed thread are skipped
continue	Same as skip_downstream — other independent threads proceed

Transitive dependents are computed via BFS over the reverse dependency graph.

Cache lifecycle

The CacheManager uses reference counting to manage in-memory DataFrames:

1. When a cache-target thread completes, its output DataFrame is persisted directly at MEMORY_AND_DISK level (no re-read from Delta).
2. Each downstream consumer calls notify_complete() after finishing, which decrements the consumer count.
3. When the count reaches zero, the cached DataFrame is automatically unpersisted.
4. A cleanup() call in the executor's finally block force-unpersists anything remaining, preventing memory leaks.

Cache failures are non-fatal — if persistence fails, execution continues without the cache optimization.

Module map

The weevr.engine namespace is federated across the two wheels: the pure-Python planner and the column/duration-formatting helpers ship from weevr-core, and the rest — everything that touches Spark — ships from the engine wheel. The "Wheel" column below records which package each file lives in.

Module	Wheel	Responsibility
engine/planner.py	weevr-core	DAG construction, cycle detection, topological sort, cache analysis
engine/formatting.py	weevr-core	Pure-Python column-name and duration formatters used by result rendering
engine/executor.py	weevr	Thread-level pipeline: read → transform → validate → write
engine/runner.py	weevr	Weave and loom orchestration, parallel dispatch, failure handling
engine/result.py	weevr	Immutable result models: ThreadResult, WeaveResult, LoomResult
engine/cache_manager.py	weevr	Reference-counted DataFrame caching
engine/conditions.py	weevr	Condition evaluation: parameter resolution, built-in functions, boolean parsing
engine/hooks.py	weevr	Pre/post hook step execution: quality gates, SQL statements, log messages
engine/lookups.py	weevr	Lookup materialization: pre-read, narrow projection, caching/broadcast
engine/variables.py	weevr	Weave-scoped variable binding and resolution
engine/display.py	weevr	SVG visualization (DAG, flow, timeline, waterfall) and HTML result rendering for notebooks

Design decisions

• Immutable plans and results — ExecutionPlan, ThreadResult, WeaveResult, and LoomResult are all frozen. This prevents accidental mutation during concurrent execution.
• Per-thread telemetry collectors — Each thread gets its own SpanCollector during concurrent execution, avoiding lock contention. Collectors merge into the weave-level collector after thread completion.
• Fail-fast cycle detection — Cycles are detected before any data is read, giving clear errors without wasted compute.
• Safe condition evaluation — Conditions are parsed with a recursive descent evaluator (no eval()). Only whitelisted operators and built-in functions are supported.

Further reading

• Thread, Weave, Loom — The configuration hierarchy
• Add a Thread — Step-by-step thread creation
• Observability — Telemetry spans and logging