Skip to content

yggdrasil.data.executor

executor

Backend-agnostic statement executor.

Public surface

  • :class:ExecutionOptions — cross-backend execution policy (waiting, raise-on-failure, parallelism). Not statement configuration — parameters, external tables, byte/row limits, routing hints all live as typed fields on the :class:PreparedStatement subclass.
  • :class:StatementExecutor — abstract base with a single subclass hook (:meth:_submit_statement). Coercion, batching, lifecycle, dispose semantics, and the execute / execute_many driver methods are provided here.

Subclassing

Subclasses pin their concrete types via the three :class:ClassVar attributes _PREPARED_CLASS, _RESPONSE_CLASS, _BATCH_CLASS, and implement :meth:_submit_statement. Cross-cutting behavior (logging, retries, metrics) is best added by overriding :meth:_execute — it sees an already-coerced statement and a resolved :class:ExecutionOptions, so it doesn't have to re-implement the kwargs dance.

ExecutionOptions dataclass 

ExecutionOptions(
    wait: WaitingConfigArg = True,
    raise_error: bool = True,
    parallel: Optional[int] = None,
)

Cross-backend execution policy.

Decoupled from statement-level config (parameters, external tables, byte/row limits, etc.) which lives as typed fields on the concrete :class:PreparedStatement subclass. An ExecutionOptions instance only describes how to run, never what.

Fields

wait Waiting policy. True (default polling), False (return once submitted), or a :class:WaitingConfig for custom timing. raise_error Whether to raise on a backend-reported failure. When False, the caller inspects result.failed / result.raise_for_status themselves. parallel Used only by :meth:StatementExecutor.execute_many. None = executor default. <= 1 = sequential wait.

Construction

Build with ExecutionOptions(wait=False, parallel=4) or coerce from a kwargs dict via :meth:from_kwargs. :meth:replace returns a derived options object with overrides applied — useful for layered overrides (e.g. a batch-level default + per-statement tweaks).

waits property 

waits: bool

True if this policy will block on completion at all.

from_ classmethod 

from_(
    value: "ExecutionOptions | Mapping[str, Any] | None" = None,
    **overrides: Any
) -> "ExecutionOptions"

Coerce to an :class:ExecutionOptions, applying any overrides.

  • None -> defaults
  • an existing instance -> returned with replace() applied
  • a Mapping -> constructed from it, then overrides applied

StatementExecutor 

StatementExecutor(*args: Any, **kwargs: Any)

Bases: Singleton, Disposable, Generic[PS, SR, SB]

Abstract base for backend-specific statement executors.

A :class:StatementExecutor IS a :class:Session over a transport that speaks SQL instead of HTTP: the same prepare → send pipeline drives both, and the same singleton-by-config + pickle pattern keeps connection pools and in-flight result maps shared across callers in-process.

Subclasses implement exactly one hook — :meth:_submit_statement — which turns a coerced :class:PreparedStatement into a backend- specific :class:StatementResult. The base provides the Session-shaped surface:

  • :meth:prepare — coerce raw input into the typed :class:PreparedStatement subclass (analogue of :meth:Session.prepare_request_before_send),
  • :meth:send — dispatch a prepared statement and return its :class:StatementResult. start=False returns an idled result whose backend submission is deferred until :meth:StatementResult.start fires — same shape as :meth:Session.send with start=False,
  • :meth:execute / :meth:execute_many — kwargs-friendly sugar that resolves :class:ExecutionOptions and waits.
Singleton + pickle

Concrete subclasses opt into the inherited :class:Session / :class:Singleton cache by:

  1. setting _SINGLETON_TTL = None (process-lifetime caching),
  2. overriding :meth:_singleton_key to project the identity- bearing constructor arguments into a hashable tuple,
  3. guarding __init__ with if getattr(self, "_initialized", False): return so Python's re-entry after a cache hit doesn't clobber live state,
  4. extending _TRANSIENT_STATE_ATTRS with any non-picklable handles they hold (locks, urllib3 pools, live SDK sessions).

The base default — _SINGLETON_TTL = ... (from :class:Singleton) — keeps caching opt-in so executor subclasses that genuinely don't have a stable identity (a hand-rolled test double, an anonymous executor) still work without surprise sharing.

Class-level configuration

The prepared / response / batch types are pinned on the inherited :class:Session ClassVars — :attr:Session._PREPARED_CLASS / :attr:Session._RESPONSE_CLASS / :attr:Session._BATCH_CLASS — overridden here to the SQL-shaped defaults. Backend subclasses (SQLWarehouse, SparkStatementExecutor, …) re-pin them to their concrete types so the prepare → send pipeline produces the right shape without per-call coercion.

opened property 

opened: bool

True iff :meth:_acquire has run and :meth:_release hasn't.

closed property 

closed: bool

Inverse of :attr:opened.

prepare 

prepare(statement: 'PS | PreparedStatement | str') -> PS

Coerce statement into this executor's prepared-statement type.

Mirrors :meth:Session.prepare_request_before_send: takes whatever the caller passed (raw string, cross-backend :class:PreparedStatement, already-typed instance) and returns the concrete :attr:_PREPARED_CLASS every downstream hook expects. Subclasses that need to inject per-statement defaults (warehouse routing, catalog binding, SELECT-rewrite for cluster execution) override this — same shape as Session's hook.

send 

send(statement: 'PS | PreparedStatement | str', *, start: bool = True) -> SR

Dispatch statement and return its tracking :class:StatementResult.

Mirrors :meth:Session.send. start=True (default) fires the backend submission eagerly — the result comes back in flight (or already terminal for synchronous backends). start=False returns the idled :class:StatementResult whose backend submission is deferred until :meth:StatementResult.start fires.

The returned result is always bound to this executor — every subclass _submit_statement is supposed to thread executor=self through the constructor, but that's easy to forget and downstream code (StatementResult.wait, retry, raise_for_status) needs the back-reference. Setting it here when it's missing makes the contract enforceable from one place instead of audited per backend.

execute 

execute(
    statement: "PS | PreparedStatement | str",
    *,
    options: Optional[ExecutionOptions] = None,
    wait: WaitingConfigArg = True,
    raise_error: bool = True,
    start: bool = True
) -> SR

Submit a single statement and optionally wait for completion.

Two ways to pass execution policy:

  • Per-call kwargs wait / raise_error (ergonomic, matches the previous public API).
  • An :class:ExecutionOptions via options= (when you want to reuse the same policy across many calls or compose from layered defaults).

The two are merged: options provides the base, kwargs override any field they explicitly set. Unknown kwargs go nowhere — they are not forwarded to the backend. Use a typed :class:PreparedStatement subclass for backend-specific configuration (parameters, byte limits, routing, etc.).

execute_many 

execute_many(
    statements: Iterable["PS | PreparedStatement | str"],
    *,
    options: Optional[ExecutionOptions] = None,
    wait: WaitingConfigArg = True,
    raise_error: bool = True,
    parallel: Optional[int] = None,
    **batch_kwargs: Any
) -> SB

Run several statements as a batch and return the populated batch.

Convenience wrapper around :meth:batch: enqueues every statement, submits, and (by default) waits. parallel controls the wait phase only — submission itself is sequential, since most backends either accept fast or reject fast.

**batch_kwargs are forwarded to the batch constructor (e.g. external_paths= for :class:WarehouseStatementBatch).

batch 

batch(
    statements: Optional[Iterable["PS | PreparedStatement | str"]] = None,
    *,
    executor: "StatementExecutor | None" = None,
    parallel: Optional[int] = None,
    **kwargs: Any
) -> SB

Construct a batch bound to this executor.

cancel_all 

cancel_all() -> None

Best-effort cancel every live result this executor has produced.

open 

open() -> 'Disposable'

Acquire the resource and cascade into owned children.

Order:

  1. Run our own :meth:_acquire (subclass body).
  2. Flip :attr:opened to True and mark _self_opened.

  3. For each owned child, in registration order:

  4. If the child is already opened, just :meth:_claim it. It stays self-opened — the existing self-open is what keeps it alive after we let go.

  5. Otherwise, call :meth:open on the child (which recursively cascades into ITS owned children), then clear the child's _self_opened flag so the child knows its open is parent-driven, then :meth:_claim it. Without that flag clear, the eventual :meth:_unclaim would refuse to close — it would see "I'm self-opened, someone explicitly opened me, leave me alone."

Both branches record the child in our per-frame scratch list so :meth:_release knows what to unclaim.

Transactional rollback: if any child's open or claim raises, we walk back through the children we already touched (in reverse), unclaim each, then call our own :meth:_release with committed=False and re-raise the original exception. From the caller's view, the open atomically either succeeded with the whole graph live, or failed with nothing changed.

Not reentrant: raises :class:RuntimeError if already opened. Nesting is expressed via with self: blocks, not via paired :meth:open calls.

commit 

commit()

Commit current state

rollback 

rollback()

Rollback current state

close 

close(force: bool = False) -> None

Release the resource and cascade into owned children.

Order:

  1. If currently held open by an outside parent claim (_claim_count > 0) AND we are not in self-opened state, this is a no-op — the parents that opened us still need us live. (Handled inside :meth:_do_close.)
  2. Walk our scratch list of acquired children in REVERSE registration order; :meth:_unclaim each. A child whose claim count hits zero and isn't otherwise self-opened closes itself.
  3. Run :meth:_before_release, then :meth:_release — with committed reflecting the dirty bit (cleared on exception by __exit__).

Idempotent: no-op when already closed, unless force.

force=True runs teardown even when :attr:closed. Intended for error-recovery paths where subclass state might be inconsistent.

Does NOT touch :attr:depth — the with-stack counter belongs to :meth:__enter__/:meth:__exit__ exclusively. If a caller calls :meth:close inside an active with block, the outer :meth:__exit__ will harmlessly skip the now-no-op close on unwind.

mark_dirty 

mark_dirty() -> None

Signal pending mutations — commit on next clean :meth:close.

to_singleton 

to_singleton(ttl: Any = ...) -> 'Singleton'

Promote this instance into the per-class _INSTANCES cache.

Hot listing paths (iterdir / _ls / glob) build children with singleton_ttl=False so the bounded cache doesn't fill up with thousands of short-lived entries. When a caller decides one of those children is worth keeping around (handing it to a long-running worker, returning it from an API), :meth:to_singleton registers self into the cache so the next constructor call with the same key collapses to the same instance.

ttl defaults to the subclass's _SINGLETON_TTL (... = no caching, None = process lifetime, or a seconds count). When a different instance is already cached under this key, that pre-existing one wins and is returned unchanged — the cache is the source of truth.

invalidate_singleton 

invalidate_singleton(remove_global: bool = True) -> None

Pop self from the per-class _INSTANCES cache.

Mutating ops on a Singleton-cached object (writes, deletes, schema invalidations on a Databricks table, put_object on an :class:S3Path) want to make sure the next caller asking for the same key gets a fresh build rather than collapsing onto this stale handle — that's what remove_global=True (the default) does. The pop is :meth:identity-guarded: only an entry that still points at self is removed, so a concurrent re-construction that already raced past this thread is left alone.

remove_global=False is a no-op. The keyword exists so subclass invalidators (invalidate_singleton, _invalidate_entity_tag_cache, …) can offer the same switch without branching at the call site.