Concurrency & Executor
Relevant source files
The following files were used as context for generating this wiki page:
The Concurrency & Executor system in dbt-common provides the infrastructure for managed parallel execution of tasks, primarily through thread-based concurrency. It ensures that asynchronous operations remain consistent with the application's global state by managing invocation contexts and event propagation across execution boundaries.
This module is central to dbt's ability to run multiple node executions or parsing tasks simultaneously while maintaining a unified view of logging, environment variables, and metadata.
Sources: CHANGELOG.md:144, dbt_common/events/functions.py:126-150
Multi-Threaded Execution Architecture
The core of the concurrency system is the MultiThreadedExecutor. This component is responsible for orchestrating parallel tasks while ensuring that critical application state, specifically the InvocationContext, is preserved and accessible within worker threads.
Invocation Context Support
A significant enhancement to the executor was the addition of Invocation Context support. This ensures that identifiers like invocation_id and other global execution metadata are correctly tracked when tasks move from a main thread to a pool of worker threads.
Sources: CHANGELOG.md:144, dbt_common/events/functions.py:27-35
The diagram above illustrates how the MultiThreadedExecutor captures the current execution context and propagates it to distributed tasks in the worker pool.
Sources: CHANGELOG.md:144, dbt_common/events/functions.py:126-140
Event Management in Concurrent Environments
Concurrency introduces complexity in event handling. dbt-common manages this through a centralized EventManager that supports deferral and group-based firing to prevent race conditions and interleaved log output.
Deferred Event Firing
To maintain order and clarity in parallel execution, the system supports deferring events. Events can be assigned to specific groups and fired later, ensuring that related log messages appear together even if the underlying tasks were executed out of sync.
| Component | Functionality | Purpose |
|---|---|---|
fire_or_defer_event | Conditional event queuing | Queues an event to a specific EventGroupType instead of immediate execution. |
fire_deferred_events | Batch processing | Flushes all events stored for a specific group to the logger. |
EventGroupType | Logical grouping | Categorizes events (e.g., DEFAULT) for synchronized output. |
Sources: dbt_common/events/functions.py:153-176, CHANGELOG.md:16-25
Cross-Thread Scoped State
The system utilizes contextvars and global state management to handle metadata across concurrent processes. This is critical for functions like get_metadata_vars(), which extracts environment-specific custom variables (prefixed with DBT_ENV_CUSTOM_ENV_) that must be consistent across all concurrent tasks.
Sources: dbt_common/events/functions.py:179-192, CHANGELOG.md:139
Concurrency in Template Rendering
Jinja template rendering within dbt-common utilizes specialized environments and caching to handle high-frequency concurrent requests.
Template Caching
To optimize performance in multi-threaded scenarios, a TemplateCache is used to store compiled Jinja templates. This avoids the overhead of repeated compilation when multiple threads attempt to render the same macro or materialization.
Sources: dbt_common/clients/jinja.py:275-296
Macro Execution Flow
The CallableMacroGenerator manages the execution of macros, providing an exception_handler context manager that translates standard Jinja errors into dbt-specific exceptions like CaughtMacroErrorWithNodeError.
This flow shows how the generator ensures thread-safe access to cached templates and handles exceptions within a controlled scope.
Sources: dbt_common/clients/jinja.py:328-360
Configuration and Merge Behaviors
Concurrent updates to configuration objects are managed through strict merge strategies defined in BaseConfig. This prevents data corruption when multiple execution paths attempt to modify settings.
Merge Strategy Types
Configurations use MergeBehavior metadata to determine how values from different sources (or concurrent updates) are combined.
- Append: Combines lists (e.g.,
pre-hook,tags). - Update: Merges dictionaries (e.g.,
quoting,meta). - Clobber: Overwrites the existing value (Default).
- DictKeyAppend: Specifically handles merging lists within dictionary keys (e.g.,
grants). - Object: Deep update for specific configuration objects.
Sources: dbt_common/contracts/config/base.py:151-177, dbt_common/contracts/config/base.py:228-278
Implementation Details
Thread-Safe Formatting
The Formatting and Note event types (codes Z017 and Z050) are designed for unstructured output that may be interleaved during parallel execution. The PrintEvent (Z052) is particularly important for concurrency as it bypasses standard formatting to ensure a message reaches stdout even in quiet modes.
Sources: dbt_common/events/types.py:126-154
Safe Command Execution
When the executor runs system commands, it utilizes SystemExecutingCmd (Z006) and captures STDOUT/STDERR (Z007/Z008) in a way that attributes the output to the specific concurrent task via the invocation context.
Sources: dbt_common/events/types.py:101-118
Summary
The Concurrency & Executor framework provides a robust foundation for parallel operations. By integrating InvocationContext support directly into the MultiThreadedExecutor and providing a sophisticated deferred event system, it ensures that dbt can scale performance across multiple threads without losing the traceability or consistency of its internal state and logs.
Sources: CHANGELOG.md:144, dbt_common/events/functions.py:153-170