yggdrasil.data.statement¶
statement ¶
Backend-agnostic statement abstractions.
Three concrete types, each with a clear single responsibility:
- :class:
PreparedStatement— a value object: the SQL text plus any binding metadata. No execution state. - :class:
StatementResult— a handle to a statement being or having been executed by some backend. Carries the per-execution state (statement_id, response, materialized data) and exposes Arrow I/O via :class:Tabular. Lifecycle hooks:start,cancel,refresh_status,done,failed. - :class:
StatementBatch— a collection of pending statements + their in-flight / completed results. Not a Tabular: the batch as a whole has no rows, only its individual results do. Convenience for add-then-wait flows; parallelism is opt-in.
A few design rules the cleanup pass enforces:
StatementBatchis a plain collection — it doesn't pretend to be a tabular result. Iterating it yields keys;__getitem__looks up a result by key;materialized()walks finished results.add()returns the key (per its docstring);submit()is not a generator and returnsself.clear()cancels everything and drops it;clear_temporary_resources()only releases per-statement scratch, never drops results.- Subclass overrides have one place to land:
_coercefor input normalization,_submit_onefor backend dispatch,_after_submitfor any post-submit bookkeeping.
ExternalStatementData ¶
ExternalStatementData(
text_key: str,
tabular: Optional[Tabular] = None,
*,
text_value: Optional[str] = None
)
A tabular binding that a statement carries alongside its SQL text.
Lets a :class:PreparedStatement reference data by a placeholder
name ({text_key}) which an engine resolves to a concrete
expression at submit time. Three fields:
text_key— the placeholder identifier;{text_key}in the statement text is replaced withtext_valueon submit.text_value— the SQL fragment substituted into the text. May beNoneinitially: engines that materialize the binding (Spark registers a temporary view, the warehouse stages a Parquet volume) fill it in before substitution. Pre-set this yourself when you've already arranged the storage (e.g. an existing :class:VolumePathyou wrote to up front).tabular— the data the placeholder represents. Engines pull a frame off it via :meth:Tabular.read_spark_frame/ :meth:Tabular.read_arrow_batchesto register / stage it.
Plain value object — engines treat it as mutable scratch and may
rewrite text_value during materialization.
from_
classmethod
¶
from_(
value: "ExternalStatementData | Tabular | str | tuple",
*,
text_key: Optional[str] = None
) -> "ExternalStatementData"
Coerce value into an :class:ExternalStatementData.
- Already an instance → pass-through (
text_keyignored). - :class:
Tabular→ bind under the suppliedtext_key. str→text_valueonly (no tabular, caller staged it).(tabular, text_value)tuple → both fields set.
Raises ValueError when text_key is required but missing.
PreparedStatement ¶
PreparedStatement(
text: str = "",
key: Optional[str] = None,
retry: Optional[WaitingConfigArg] = None,
*,
external_data: Optional[
Mapping[str, "ExternalStatementData | Tabular | str | tuple"]
] = None
)
Bases: Disposable
Configuration for a single statement execution.
Plain value object — SQL text, parameter bindings, external-table
aliases. Runtime/execution state lives on :class:StatementResult.
Mutators (with_text, clear) return a new instance unless
inplace=True is passed.
Retry config (read by :meth:StatementResult.retry):
retry— :class:WaitingConfigcontrolling the result-level retry loop.None(the default) means not retryable. When set,retry.retries + 1total attempts are made, with sleep between attempts driven by :meth:WaitingConfig.sleep(exponential backoff capped atmax_interval, terminated bytimeout).
Subclasses set their own retry default by passing a
:class:WaitingConfig to super().__init__(retry=...) from their
own __init__ defaults. The base default is None — caller
must opt in.
retryable
property
¶
Whether a non-None retry policy has been configured.
Convenience for the lifecycle code; self.retry is not None
works equivalently.
with_text ¶
Return a copy with text replaced (or mutate in place).
with_retry ¶
Return (or update in place) a copy with retry set.
retry=None clears the policy (statement becomes non-retryable);
anything else is normalized through :meth:WaitingConfig.from_.
looks_like_query
staticmethod
¶
Return True when text parses as a SQL SELECT-like query.
Skips leading whitespace and SQL comments; a string is treated as a
query when its first keyword is SELECT, WITH, VALUES,
TABLE, or FROM. Non-string inputs return False.
from_
classmethod
¶
Coerce statement into an instance of cls.
Already-an-instance pass-through, str → cls(str), StatementResult
→ recurse on its underlying statement. Subclasses can extend this
but the common cases all fall through here.
prepare
classmethod
¶
Coerce + bind metadata. Base impl handles only the text; subclasses override to thread parameters / external tables onto their typed fields.
apply_external_substitution
staticmethod
¶
apply_external_substitution(
text: str, external_data: Optional[Mapping[str, ExternalStatementData]]
) -> str
Substitute every {text_key} in text with its text_value.
Engine-agnostic: the caller (Spark / warehouse / ...) is
responsible for filling in each entry's text_value (registering
a temp view, staging a Parquet volume, ...) before invoking this.
Entries whose text_value is still None raise — better to
fail loudly than silently leave an unsubstituted placeholder in
the SQL.
clear_temporary_resources ¶
Release per-statement scratch (staged volumes, temp views, ...).
Default is a no-op. Subclasses that allocate scratch on
:meth:prepare override this. Idempotent — callers may invoke
it more than once.
open ¶
Acquire the resource and cascade into owned children.
Order:
- Run our own :meth:
_acquire(subclass body). - Flip :attr:
openedto True and mark_self_opened. -
For each owned child, in registration order:
-
If the child is already opened, just :meth:
_claimit. It stays self-opened — the existing self-open is what keeps it alive after we let go. - Otherwise, call :meth:
openon the child (which recursively cascades into ITS owned children), then clear the child's_self_openedflag so the child knows its open is parent-driven, then :meth:_claimit. Without that flag clear, the eventual :meth:_unclaimwould 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.
close ¶
Release the resource and cascade into owned children.
Order:
- 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.) - Walk our scratch list of acquired children in REVERSE
registration order; :meth:
_unclaimeach. A child whose claim count hits zero and isn't otherwise self-opened closes itself. - Run :meth:
_before_release, then :meth:_release— withcommittedreflecting 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.
StatementResult ¶
StatementResult(
statement: PS,
*,
key: Optional[str] = None,
executor: Optional["StatementExecutor"] = None,
iteration: int = 0,
start_timestamp: Optional[int] = None,
**kwargs: Any
)
Bases: Tabular, Awaitable, Generic[PS]
Backend-agnostic handle to a running or completed statement.
Inherits :class:Tabular (Arrow I/O) and :class:Awaitable
(start / wait / cancel lifecycle with retry). Subclasses implement
the three Awaitable hooks:
- :meth:
_poll— refresh state from the backend - :meth:
_start— submit the statement - :meth:
_error_for_status— return the backend-specific error
progress ¶
Completion fraction for a progress bar (style.track).
A warehouse doesn't report a running query's % mid-flight, so execution
is indeterminate (None → an animated sweep): 0.0 before it
starts, None while it runs, 1.0 once it's succeeded.
watch ¶
Drive to completion, calling on_tick(self) each poll.
The hook a UI (spinner / progress bar) connects to without this trait
importing any UI — keeping the layering clean. Starts the awaitable if
it hasn't been, polls until done, then surfaces a failure (unless
raise_error is False). Pairs with :func:yggdrasil.cli.style.track.
open ¶
Acquire the resource and cascade into owned children.
Order:
- Run our own :meth:
_acquire(subclass body). - Flip :attr:
openedto True and mark_self_opened. -
For each owned child, in registration order:
-
If the child is already opened, just :meth:
_claimit. It stays self-opened — the existing self-open is what keeps it alive after we let go. - Otherwise, call :meth:
openon the child (which recursively cascades into ITS owned children), then clear the child's_self_openedflag so the child knows its open is parent-driven, then :meth:_claimit. Without that flag clear, the eventual :meth:_unclaimwould 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.
close ¶
Drop the schema cache and forward to any cooperative close.
Tabular itself has no resources to release — the schema cache
is the only state it owns. Subclasses that mix Tabular with a
lifecycle (Disposable-derived IO, holders, …) inherit
this hook through cooperative super().close(); pure
Tabular subclasses without a lifecycle peer get a harmless
no-op forward.
for_scheme
classmethod
¶
Return the :class:URLBased subclass registered for scheme.
Lazy: if no subclass is registered yet, this routes through
:meth:Scheme.path_class which imports the backend module on
demand (firing :meth:__init_subclass__ as a side effect).
Raises :class:ValueError for an unknown scheme and
:class:ImportError when the backend's optional dependencies
aren't installed.
dispatch
classmethod
¶
Build the right :class:URLBased subclass from url.
Looks up the subclass via :meth:for_scheme, then delegates
to that subclass's :meth:from_url. Used as the cross-cutting
entry point when the caller has a URL but doesn't know (or
care) which concrete class owns its scheme.
URL.from_(url).scheme drives the lookup; an empty scheme
falls back to the file:// handler so bare paths work.
to_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 ¶
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.
matches_static ¶
True iff predicate could match any row given
:attr:static_values. Conservative on undecidables (column
not in static values, predicate evaluator failure) so the
caller still reads.
Builds a one-row pyarrow Table from the predicate's free columns that we have static values for, then evaluates the predicate against it — generalises the partition-only prune so any aggregator (folder read, future warehouse file skip) reuses the one helper.
free_cols lets a caller that's about to prune the same
predicate against N children precompute the free-column
tuple once and reuse it — :func:free_columns walks the
AST every call, so on a 64-OR predicate (the cache batch
lookup shape) the saving is N-1 full walks per
iter_children loop. Default None keeps the call
site short for one-off prune checks.
from_
classmethod
¶
from_(
obj: Any,
*,
media_type: "MediaType | MimeType | str | None" = None,
default: Any = ...,
as_folder: bool = False,
**kwargs: Any
) -> "Tabular | None"
Coerce obj into a :class:Tabular.
Routes:
None— returns default (Nonewhendefault=None).- :class:
Tabular— returned as-is. When as_folder isTrueand obj is a local :class:Path, wraps it in a :class:Folder. str/ :class:os.PathLike— coerced via :class:Path.from_. When as_folder isTrue, wraps in :class:Folder.- File-like objects — drained into :class:
Memory; media_type required.
Falls back to default on unrecognised shapes when supplied;
otherwise raises :class:TypeError.
options_class
classmethod
¶
The :class:CastOptions subclass this implementer consumes.
Default :class:CastOptions. Format-specific leaves with
their own knobs (Parquet compression, CSV delimiter, …)
override.
check_options
classmethod
¶
Validate and merge caller kwargs into a resolved options.
Canonical pattern: a public method passes overrides=locals()
and the ...-defaulted entries are stripped, the rest merged.
cleanup ¶
Garbage-collect stale state on this backend.
Default no-op (returns 0) — single-file leaves and
warehouse-backed tables don't have a sweep concept the
client owns. Folder-shaped subclasses override to unlink
stale part-* files, throttled by TTL.
wait controls sync vs async dispatch on backends that
support it: a truthy :class:yggdrasil.dataclasses.waiting.WaitingConfig
(or True / a positive timeout) blocks until the sweep
finishes; a falsy value (the default) hands the work off to a
background thread. Backends without an async path treat both
the same.
Returns the number of files / rows removed when known; 0
for fire-and-forget async dispatch or a no-op backend.
optimize ¶
Repartition / compact this Tabular's storage.
Default implementation is a no-op and returns 0 — single-file
leaves (parquet, csv, arrow IPC, …) don't have a compaction
concept. Aggregator subclasses (:class:Folder) override
this to walk their child leaves and bin-pack small part files
into bundles near byte_size.
Files already close to the target size are left alone so a
repeated call is cheap.
byte_size=None keeps the legacy "collapse every leaf with
more than one part into a single file" behavior, which is what
the local-cache compaction loop in :class:Session expects.
Any extra keyword arguments are accepted and ignored so
upstream callers can pass forward-compatible knobs without the
base raising.
delete ¶
delete(
predicate: "PredicateLike" = None,
*,
wait: "WaitingConfigArg" = True,
missing_ok: bool = False,
delete_staging: bool = True,
**kwargs: Any
) -> "Table"
Delete rows matching predicate; return this tabular.
predicate is a :class:Predicate from
:mod:yggdrasil.execution.expr or a SQL string that parses into
one ("id IN (1,2,3)", "price > 100 AND region = 'EU'").
None means "no filter" — every row is removed (DELETE FROM t
with no WHERE).
wait / missing_ok / delete_staging are honoured by
resource-backed subclasses (e.g.
:class:yggdrasil.databricks.table.table.Table, which drops the
table asset); the generic row-rewrite path ignores them. Any extra
**kwargs (e.g. options=DeltaOptions(...)) flow through to
:meth:_delete.
The default implementation reads every batch, drops rows the
predicate accepts, and rewrites the leaf with the survivors.
Aggregator subclasses (:class:yggdrasil.path.folder.Folder)
override to walk children, prune subtrees whose partition bounds
make the predicate trivially false, and only rewrite the leaves
that actually hold matched rows.
collect_schema ¶
Return this Tabular's :class:Schema, caching the first hit.
The cache slot is :attr:_schema_cache; on first call this
method stamps the resolved schema into it so subsequent
collect_schema calls short-circuit. Writers overwrite
the slot via :meth:_persist_schema; lifecycle hooks clear
it via :meth:_unpersist_schema.
count ¶
Return the number of rows in this tabular.
scan_arrow_batches ¶
Zero-copy scan — yield the source's :class:pa.RecordBatch views verbatim.
The lazy / zero-copy counterpart to :meth:read_arrow_batches,
mirroring :meth:read_polars_frame vs :meth:scan_polars_frame.
Where read_arrow_batches layers the full options pipeline on
every batch — target cast, projection, resample, dedup, row-limit
slicing, each of which can copy or re-encode — scan_arrow_batches
hands back exactly what the leaf produced, untouched. For an
in-memory source (:class:~yggdrasil.arrow.tabular.ArrowTabular)
those batches are views over the held buffers (no copy); for a
byte-backed leaf they're the freshly-decoded batches with none of
the extra processing copies layered on. Use it when you want the
raw Arrow stream and will project / filter downstream yourself.
scan_arrow_table ¶
Zero-copy scan into one chunked :class:pa.Table (no rechunk, no cast).
The zero-copy counterpart to :meth:read_arrow_table. Assembles
the source batches with :func:pa.Table.from_batches, which
references the batch buffers as table chunks rather than copying
them — so no cast, no projection, no rechunk memcpy that
read_arrow_table performs to coalesce + conform the result. An
empty source yields an empty table carrying the bound schema.
The batches must share one schema (the zero-copy contract):
read_arrow_table reconciles parts that drifted across writes,
scan_arrow_table does not — reach for read_arrow_table when
a source's parts are known to be heterogeneous.
scan_arrow_batch_reader ¶
Zero-copy scan as a streaming :class:pa.RecordBatchReader view.
The raw-reader counterpart to :meth:read_arrow_batch_reader: wraps
the source batch stream in a reader without the per-batch
conform / target-cast pass, so batches flow through as views over
the source buffers. The reader's schema is the source's own — taken
from the first batch, so it matches the raw views exactly (no
collect_schema probe, which on a byte cursor would consume the
stream out from under the read). Only the first batch is pulled up
front to seed the schema; the rest stay lazy behind the reader.
read_table ¶
Read into an in-memory :class:Tabular.
When options.spark_session is set, reads via
:meth:_read_spark_frame and wraps in a :class:Dataset.
Otherwise materializes Arrow batches into :class:ArrowTabular.
Returns None when empty.
write_table ¶
Dispatch obj to the best _write_* hook based on its runtime type.
Recognizes another :class:Tabular (drained as a pyarrow
record-batch stream), pa.Table / pa.RecordBatch /
pa.RecordBatchReader, polars DataFrame / LazyFrame,
pandas DataFrame, pyspark DataFrame, list[dict],
dict[str, list], and iterables of any of the above.
Module-name sniffing keeps optional engine deps out of the
import graph — we only touch a frame's API once we've
confirmed it's an instance of one we know how to drain.
union ¶
Return a Tabular representing self UNION ALL other.
mode controls how mismatched schemas are reconciled:
Mode.IGNORE(default) — keepself's schema; extra columns in other are dropped, missing ones are filled null.Mode.APPEND— widen to the superset schema (every field from both sides survives).
Concrete subclasses override :meth:_union for in-place
mutation (Arrow batch append, Spark unionByName).
Accepts :class:Tabular, pa.RecordBatch, pa.Table,
list[Response], or a Spark DataFrame.
None returns self unchanged.
read_spark_dataset ¶
Read into a :class:Dataset holder.
Mirrors :meth:read_arrow_dataset for the Spark engine: the
return type is a yggdrasil holder rather than the bare engine
frame, so callers keep the Tabular surface (chained transforms,
persist / insert / schema, …) without an extra wrap
at the call site. :class:Dataset overrides
:meth:_read_spark_dataset to return itself in place — no
materialise round trip when the source already speaks Spark.
read_record_iterator ¶
Stream rows as plain dict. True streaming — the full
table never materializes; batch.to_pylist() does the
column→row rotation in pyarrow C++ once per batch.
read_records ¶
Stream rows as :class:yggdrasil.data.record.Record. Lower
per-row allocation than :meth:read_pylist for stable-schema
sources — the underlying :class:Schema is materialized once
and shared by reference across every record.
unique ¶
Drop duplicate rows on by; keep first occurrence per key tuple.
Parameters¶
by
One or more column references — :class:str column names,
:class:yggdrasil.data.Field instances (resolved via
:attr:Field.name), or any iterable mixing the two. Empty
/ None is a no-op — returns self.
Returns¶
Tabular
A new holder carrying the deduped rows. Spark-shaped
inputs (anything whose :meth:_native_spark_frame
exposes a :class:pyspark.sql.DataFrame) return a fresh
:class:yggdrasil.spark.tabular.Dataset over the
spark-side dedup; everything else collects through Arrow
and returns an :class:yggdrasil.arrow.tabular.ArrowTabular.
resample ¶
resample(
on: "str | Any",
sampling: "int | float | Any",
*,
partition_by: "str | Any | Iterable[Any] | None" = None,
fill_strategy: "str | None" = "ffill"
) -> "Tabular"
Align rows to a fixed time grid on on; one row per bucket.
Parameters¶
on
The time column to resample on — column name
(:class:str) or :class:yggdrasil.data.Field.
sampling
Bucket size. Accepted shapes:
* :class:`int` / :class:`float` — seconds (floats are
rounded to the nearest integer second).
* :class:`datetime.timedelta` — total seconds.
* :class:`str` — ISO-8601 duration (``"PT1H"``,
``"P1D"``, ``"PT15M"``) parsed via
:func:`yggdrasil.data.types.primitive.temporal._parse_iso_duration`.
``sampling <= 0`` is a short-circuit — returns ``self``.
partition_by
Entity columns the resample is independent on. None /
empty → flat global timeline. Same coercion as
:meth:unique's by.
fill_strategy
How to fill nulls left by the bucket's "first" aggregation.
"ffill" (default), "bfill", or "none" /
None to disable. See
:func:yggdrasil.arrow.ops.fill_arrow_table for the
full semantics.
Returns¶
Tabular
Spark-shaped holders return a :class:Dataset over the
spark-side resample; everything else returns an
:class:ArrowTabular over the arrow-side resample.
select ¶
Project to columns and return a new Tabular.
Each entry is a column reference — :class:str, a
:class:yggdrasil.data.Field (resolved via
:attr:Field.name), or an iterable mixing both. The result
preserves the caller's order, which matches both
:meth:pyarrow.Table.select and
:meth:pyspark.sql.DataFrame.select semantics.
Raises :class:ValueError on an empty selection — a zero-
column projection is almost always a caller mistake; pass
:class:Schema.empty projections through the cast surface
instead.
drop ¶
Return a new Tabular with the named columns removed.
Columns missing from the source are silently ignored —
matches Spark's :meth:DataFrame.drop and pyarrow's
:meth:Table.drop_columns (when filtered to existing
names). An empty argument list is a no-op that returns
self.
filter ¶
Drop rows where predicate is false.
predicate accepts every shape
:meth:yggdrasil.execution.expr.Expression.from_
recognises:
- a SQL predicate string (
"x > 0 AND y IS NOT NULL"), parsed by the in-tree SQL parser; - a yggdrasil :class:
Predicatenode (col("x") > 0, :func:is_in, :func:between, …); - a native engine expression —
:class:
pyarrow.compute.Expression, :class:polars.Expr, or :class:pyspark.sql.Column— lifted via the matching backend.
The predicate is parsed once and dispatched to the typed
:meth:_filter hook; the engine-side filter then runs in
its native kernel (Arrow C++, Spark Catalyst) so the row
scan stays vectorised.
cast ¶
Cast rows, returning a new :class:Tabular.
Accepts a :class:Schema or :class:CastOptions. When
options is given, reads to arrow and casts each batch
through :meth:CastOptions.cast_arrow_batch.
display ¶
Render the first n rows as an aligned, typed text table.
Columns and their types come from this Tabular's own
:meth:collect_schema — the header is two rows: the column names,
then their type tags (the project :class:~yggdrasil.data.Field's
:meth:Field.short → :meth:DataType.short, recursive for nested types
— i64 / str / list<str> / struct<name:str, age:i64>).
Columns are separated by │ with a ─┼─ rule; numbers/booleans
right-align; nested cell values are compacted to one line. Long values
and headers are clipped (cells to max_width, type/name tags to a
slightly larger cap) so one long string or column name can't balloon the
table. The n rows are pushed down as a row_limit so no more than
that is ever read.
print(dbc.sql.execute("SELECT * FROM t").display())
print(IO.from_("data.parquet").display(5))
lazy ¶
Return a :class:LazyTabular wrapping this source.
Transformations on the returned object (select, filter,
join, …) accumulate in an :class:ExecutionPlan without
touching data. Any read_* call materialises the plan.
StatementBatch ¶
StatementBatch(
executor: "StatementExecutor",
statements: Optional[Iterable["PS | str"]] = None,
parallel: int = 1,
**kwargs: Any
)
Bases: StatementResult[PS], Generic[PS, SR]
A pending queue of statements plus a map of in-flight / completed results.
A batch IS a :class:StatementResult — its aggregate state is the
composite of every child's state (any failed → failed once every
child has settled, all done with none failed → succeeded, anything
still active → running). This lets a batch flow through APIs that
accept a single result, and lets the same lifecycle primitives
(wait, cancel, raise_for_status) work on either a single
statement or a whole batch.
Lifecycle::
batch.add(stmt_or_str) # enqueue (key auto-generated)
batch.add(stmt, key="custom") # enqueue with explicit key
batch.submit() # drain queue, hand each to the executor
batch.wait() # block until all results terminal
for key, result in batch.results.items():
...
parallel > 1 runs the wait phase concurrently — appropriate
because each :meth:StatementResult.wait is I/O-bound polling.
text
property
¶
Aggregate text — every child's text joined with "; ".
Overrides :attr:StatementResult.text (which dereferences
self.statement) because a batch has no single statement.
Useful for diagnostics / repr — the executor never reads this
for submission.
add ¶
Enqueue a statement; return its key.
key collisions (against pending statements or completed
results) raise :class:ValueError.
extend ¶
Enqueue multiple; return the list of assigned keys.
Inlined over :meth:add to hoist the self.executor /
_PREPARED_CLASS / send lookups out of the per-item
loop. The auto-key path skips add's collision check —
:meth:PreparedStatement.__init__ already mints a fresh key
per statement, so no duplicates are possible from the
auto-keyed path. Callers needing explicit keys still route
through :meth:add.
remove ¶
Remove an entry by key.
Pending statement → dropped, None returned. In-flight result
→ cancelled, scratch released, instance returned. Unknown key →
:class:KeyError.
clear ¶
Cancel every in-flight result, drop every pending statement.
Removes results from self.results after cancelling. For a
cancel-but-keep version (so callers can still inspect failures),
call :meth:cancel instead.
clear_temporary_resources ¶
Release per-result scratch. Does not cancel or drop anything.
materialized ¶
Yield (key, result) pairs for every submitted result.
wait ¶
wait(
*, wait: WaitingConfigArg = True, raise_error: bool = True, **kwargs: Any
) -> "StatementBatch"
Wait for every submitted statement to reach a terminal state.
Auto-submits any pending statements first so callers can add()
then wait() without an intermediate submit(). When
parallel > 1 the per-result waits run on a thread pool — each
:meth:StatementResult.wait is I/O-bound polling.
Per-result scratch (:meth:StatementResult.clear_temporary_resources)
fires from inside :meth:StatementResult.wait on success — we
don't re-sweep here because the cleanup is idempotent and the
re-walk is pure overhead. Batch-wide scratch (e.g. warehouse-
level :attr:external_volume_paths) stays under the typed
:meth:clear_temporary_resources override on the subclass and
runs when the caller closes / drops the batch.
retry ¶
retry(
*, wait: WaitingConfigArg = True, raise_error: bool = True, **kwargs: Any
) -> "StatementBatch"
Retry every failed result whose statement is retryable.
Walks self.results once, picks the entries that are both
failed and retryable, and calls :meth:StatementResult.retry
on each. Honors self.parallel exactly like :meth:wait.
Non-retryable failures are left alone — they'll surface through
raise_for_status at the end if raise_error=True. Pending
statements (never submitted) are submitted first, same as
:meth:wait.
cancel ¶
Cancel every in-flight statement; drop everything still pending.
Idempotent. Does not drop completed results from self.results
— callers may still want to inspect failure status.
raise_for_status ¶
Surface the latest backend failure directly — no generic wrapper.
Walks self.results in submission order; with one or more
failed items, propagates the last failed result's typed
backend exception (e.g. :class:SQLError with the full
DELTA_CONCURRENT_APPEND payload) so the caller sees the
actual error instead of a wrapped RuntimeError("Batch item
... failed."). Earlier failures are logged via
:meth:StatementResult.raise_for_status so their diagnostics
aren't swallowed by the one we re-raise.
progress ¶
Completion fraction for a progress bar (style.track).
A warehouse doesn't report a running query's % mid-flight, so execution
is indeterminate (None → an animated sweep): 0.0 before it
starts, None while it runs, 1.0 once it's succeeded.
watch ¶
Drive to completion, calling on_tick(self) each poll.
The hook a UI (spinner / progress bar) connects to without this trait
importing any UI — keeping the layering clean. Starts the awaitable if
it hasn't been, polls until done, then surfaces a failure (unless
raise_error is False). Pairs with :func:yggdrasil.cli.style.track.
open ¶
Acquire the resource and cascade into owned children.
Order:
- Run our own :meth:
_acquire(subclass body). - Flip :attr:
openedto True and mark_self_opened. -
For each owned child, in registration order:
-
If the child is already opened, just :meth:
_claimit. It stays self-opened — the existing self-open is what keeps it alive after we let go. - Otherwise, call :meth:
openon the child (which recursively cascades into ITS owned children), then clear the child's_self_openedflag so the child knows its open is parent-driven, then :meth:_claimit. Without that flag clear, the eventual :meth:_unclaimwould 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.
close ¶
Drop the schema cache and forward to any cooperative close.
Tabular itself has no resources to release — the schema cache
is the only state it owns. Subclasses that mix Tabular with a
lifecycle (Disposable-derived IO, holders, …) inherit
this hook through cooperative super().close(); pure
Tabular subclasses without a lifecycle peer get a harmless
no-op forward.
for_scheme
classmethod
¶
Return the :class:URLBased subclass registered for scheme.
Lazy: if no subclass is registered yet, this routes through
:meth:Scheme.path_class which imports the backend module on
demand (firing :meth:__init_subclass__ as a side effect).
Raises :class:ValueError for an unknown scheme and
:class:ImportError when the backend's optional dependencies
aren't installed.
dispatch
classmethod
¶
Build the right :class:URLBased subclass from url.
Looks up the subclass via :meth:for_scheme, then delegates
to that subclass's :meth:from_url. Used as the cross-cutting
entry point when the caller has a URL but doesn't know (or
care) which concrete class owns its scheme.
URL.from_(url).scheme drives the lookup; an empty scheme
falls back to the file:// handler so bare paths work.
to_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 ¶
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.
matches_static ¶
True iff predicate could match any row given
:attr:static_values. Conservative on undecidables (column
not in static values, predicate evaluator failure) so the
caller still reads.
Builds a one-row pyarrow Table from the predicate's free columns that we have static values for, then evaluates the predicate against it — generalises the partition-only prune so any aggregator (folder read, future warehouse file skip) reuses the one helper.
free_cols lets a caller that's about to prune the same
predicate against N children precompute the free-column
tuple once and reuse it — :func:free_columns walks the
AST every call, so on a 64-OR predicate (the cache batch
lookup shape) the saving is N-1 full walks per
iter_children loop. Default None keeps the call
site short for one-off prune checks.
from_
classmethod
¶
from_(
obj: Any,
*,
media_type: "MediaType | MimeType | str | None" = None,
default: Any = ...,
as_folder: bool = False,
**kwargs: Any
) -> "Tabular | None"
Coerce obj into a :class:Tabular.
Routes:
None— returns default (Nonewhendefault=None).- :class:
Tabular— returned as-is. When as_folder isTrueand obj is a local :class:Path, wraps it in a :class:Folder. str/ :class:os.PathLike— coerced via :class:Path.from_. When as_folder isTrue, wraps in :class:Folder.- File-like objects — drained into :class:
Memory; media_type required.
Falls back to default on unrecognised shapes when supplied;
otherwise raises :class:TypeError.
options_class
classmethod
¶
The :class:CastOptions subclass this implementer consumes.
Default :class:CastOptions. Format-specific leaves with
their own knobs (Parquet compression, CSV delimiter, …)
override.
check_options
classmethod
¶
Validate and merge caller kwargs into a resolved options.
Canonical pattern: a public method passes overrides=locals()
and the ...-defaulted entries are stripped, the rest merged.
cleanup ¶
Garbage-collect stale state on this backend.
Default no-op (returns 0) — single-file leaves and
warehouse-backed tables don't have a sweep concept the
client owns. Folder-shaped subclasses override to unlink
stale part-* files, throttled by TTL.
wait controls sync vs async dispatch on backends that
support it: a truthy :class:yggdrasil.dataclasses.waiting.WaitingConfig
(or True / a positive timeout) blocks until the sweep
finishes; a falsy value (the default) hands the work off to a
background thread. Backends without an async path treat both
the same.
Returns the number of files / rows removed when known; 0
for fire-and-forget async dispatch or a no-op backend.
optimize ¶
Repartition / compact this Tabular's storage.
Default implementation is a no-op and returns 0 — single-file
leaves (parquet, csv, arrow IPC, …) don't have a compaction
concept. Aggregator subclasses (:class:Folder) override
this to walk their child leaves and bin-pack small part files
into bundles near byte_size.
Files already close to the target size are left alone so a
repeated call is cheap.
byte_size=None keeps the legacy "collapse every leaf with
more than one part into a single file" behavior, which is what
the local-cache compaction loop in :class:Session expects.
Any extra keyword arguments are accepted and ignored so
upstream callers can pass forward-compatible knobs without the
base raising.
delete ¶
delete(
predicate: "PredicateLike" = None,
*,
wait: "WaitingConfigArg" = True,
missing_ok: bool = False,
delete_staging: bool = True,
**kwargs: Any
) -> "Table"
Delete rows matching predicate; return this tabular.
predicate is a :class:Predicate from
:mod:yggdrasil.execution.expr or a SQL string that parses into
one ("id IN (1,2,3)", "price > 100 AND region = 'EU'").
None means "no filter" — every row is removed (DELETE FROM t
with no WHERE).
wait / missing_ok / delete_staging are honoured by
resource-backed subclasses (e.g.
:class:yggdrasil.databricks.table.table.Table, which drops the
table asset); the generic row-rewrite path ignores them. Any extra
**kwargs (e.g. options=DeltaOptions(...)) flow through to
:meth:_delete.
The default implementation reads every batch, drops rows the
predicate accepts, and rewrites the leaf with the survivors.
Aggregator subclasses (:class:yggdrasil.path.folder.Folder)
override to walk children, prune subtrees whose partition bounds
make the predicate trivially false, and only rewrite the leaves
that actually hold matched rows.
collect_schema ¶
Return this Tabular's :class:Schema, caching the first hit.
The cache slot is :attr:_schema_cache; on first call this
method stamps the resolved schema into it so subsequent
collect_schema calls short-circuit. Writers overwrite
the slot via :meth:_persist_schema; lifecycle hooks clear
it via :meth:_unpersist_schema.
count ¶
Return the number of rows in this tabular.
scan_arrow_batches ¶
Zero-copy scan — yield the source's :class:pa.RecordBatch views verbatim.
The lazy / zero-copy counterpart to :meth:read_arrow_batches,
mirroring :meth:read_polars_frame vs :meth:scan_polars_frame.
Where read_arrow_batches layers the full options pipeline on
every batch — target cast, projection, resample, dedup, row-limit
slicing, each of which can copy or re-encode — scan_arrow_batches
hands back exactly what the leaf produced, untouched. For an
in-memory source (:class:~yggdrasil.arrow.tabular.ArrowTabular)
those batches are views over the held buffers (no copy); for a
byte-backed leaf they're the freshly-decoded batches with none of
the extra processing copies layered on. Use it when you want the
raw Arrow stream and will project / filter downstream yourself.
scan_arrow_table ¶
Zero-copy scan into one chunked :class:pa.Table (no rechunk, no cast).
The zero-copy counterpart to :meth:read_arrow_table. Assembles
the source batches with :func:pa.Table.from_batches, which
references the batch buffers as table chunks rather than copying
them — so no cast, no projection, no rechunk memcpy that
read_arrow_table performs to coalesce + conform the result. An
empty source yields an empty table carrying the bound schema.
The batches must share one schema (the zero-copy contract):
read_arrow_table reconciles parts that drifted across writes,
scan_arrow_table does not — reach for read_arrow_table when
a source's parts are known to be heterogeneous.
scan_arrow_batch_reader ¶
Zero-copy scan as a streaming :class:pa.RecordBatchReader view.
The raw-reader counterpart to :meth:read_arrow_batch_reader: wraps
the source batch stream in a reader without the per-batch
conform / target-cast pass, so batches flow through as views over
the source buffers. The reader's schema is the source's own — taken
from the first batch, so it matches the raw views exactly (no
collect_schema probe, which on a byte cursor would consume the
stream out from under the read). Only the first batch is pulled up
front to seed the schema; the rest stay lazy behind the reader.
read_table ¶
Read into an in-memory :class:Tabular.
When options.spark_session is set, reads via
:meth:_read_spark_frame and wraps in a :class:Dataset.
Otherwise materializes Arrow batches into :class:ArrowTabular.
Returns None when empty.
write_table ¶
Dispatch obj to the best _write_* hook based on its runtime type.
Recognizes another :class:Tabular (drained as a pyarrow
record-batch stream), pa.Table / pa.RecordBatch /
pa.RecordBatchReader, polars DataFrame / LazyFrame,
pandas DataFrame, pyspark DataFrame, list[dict],
dict[str, list], and iterables of any of the above.
Module-name sniffing keeps optional engine deps out of the
import graph — we only touch a frame's API once we've
confirmed it's an instance of one we know how to drain.
union ¶
Return a Tabular representing self UNION ALL other.
mode controls how mismatched schemas are reconciled:
Mode.IGNORE(default) — keepself's schema; extra columns in other are dropped, missing ones are filled null.Mode.APPEND— widen to the superset schema (every field from both sides survives).
Concrete subclasses override :meth:_union for in-place
mutation (Arrow batch append, Spark unionByName).
Accepts :class:Tabular, pa.RecordBatch, pa.Table,
list[Response], or a Spark DataFrame.
None returns self unchanged.
read_spark_dataset ¶
Read into a :class:Dataset holder.
Mirrors :meth:read_arrow_dataset for the Spark engine: the
return type is a yggdrasil holder rather than the bare engine
frame, so callers keep the Tabular surface (chained transforms,
persist / insert / schema, …) without an extra wrap
at the call site. :class:Dataset overrides
:meth:_read_spark_dataset to return itself in place — no
materialise round trip when the source already speaks Spark.
read_record_iterator ¶
Stream rows as plain dict. True streaming — the full
table never materializes; batch.to_pylist() does the
column→row rotation in pyarrow C++ once per batch.
read_records ¶
Stream rows as :class:yggdrasil.data.record.Record. Lower
per-row allocation than :meth:read_pylist for stable-schema
sources — the underlying :class:Schema is materialized once
and shared by reference across every record.
unique ¶
Drop duplicate rows on by; keep first occurrence per key tuple.
Parameters¶
by
One or more column references — :class:str column names,
:class:yggdrasil.data.Field instances (resolved via
:attr:Field.name), or any iterable mixing the two. Empty
/ None is a no-op — returns self.
Returns¶
Tabular
A new holder carrying the deduped rows. Spark-shaped
inputs (anything whose :meth:_native_spark_frame
exposes a :class:pyspark.sql.DataFrame) return a fresh
:class:yggdrasil.spark.tabular.Dataset over the
spark-side dedup; everything else collects through Arrow
and returns an :class:yggdrasil.arrow.tabular.ArrowTabular.
resample ¶
resample(
on: "str | Any",
sampling: "int | float | Any",
*,
partition_by: "str | Any | Iterable[Any] | None" = None,
fill_strategy: "str | None" = "ffill"
) -> "Tabular"
Align rows to a fixed time grid on on; one row per bucket.
Parameters¶
on
The time column to resample on — column name
(:class:str) or :class:yggdrasil.data.Field.
sampling
Bucket size. Accepted shapes:
* :class:`int` / :class:`float` — seconds (floats are
rounded to the nearest integer second).
* :class:`datetime.timedelta` — total seconds.
* :class:`str` — ISO-8601 duration (``"PT1H"``,
``"P1D"``, ``"PT15M"``) parsed via
:func:`yggdrasil.data.types.primitive.temporal._parse_iso_duration`.
``sampling <= 0`` is a short-circuit — returns ``self``.
partition_by
Entity columns the resample is independent on. None /
empty → flat global timeline. Same coercion as
:meth:unique's by.
fill_strategy
How to fill nulls left by the bucket's "first" aggregation.
"ffill" (default), "bfill", or "none" /
None to disable. See
:func:yggdrasil.arrow.ops.fill_arrow_table for the
full semantics.
Returns¶
Tabular
Spark-shaped holders return a :class:Dataset over the
spark-side resample; everything else returns an
:class:ArrowTabular over the arrow-side resample.
select ¶
Project to columns and return a new Tabular.
Each entry is a column reference — :class:str, a
:class:yggdrasil.data.Field (resolved via
:attr:Field.name), or an iterable mixing both. The result
preserves the caller's order, which matches both
:meth:pyarrow.Table.select and
:meth:pyspark.sql.DataFrame.select semantics.
Raises :class:ValueError on an empty selection — a zero-
column projection is almost always a caller mistake; pass
:class:Schema.empty projections through the cast surface
instead.
drop ¶
Return a new Tabular with the named columns removed.
Columns missing from the source are silently ignored —
matches Spark's :meth:DataFrame.drop and pyarrow's
:meth:Table.drop_columns (when filtered to existing
names). An empty argument list is a no-op that returns
self.
filter ¶
Drop rows where predicate is false.
predicate accepts every shape
:meth:yggdrasil.execution.expr.Expression.from_
recognises:
- a SQL predicate string (
"x > 0 AND y IS NOT NULL"), parsed by the in-tree SQL parser; - a yggdrasil :class:
Predicatenode (col("x") > 0, :func:is_in, :func:between, …); - a native engine expression —
:class:
pyarrow.compute.Expression, :class:polars.Expr, or :class:pyspark.sql.Column— lifted via the matching backend.
The predicate is parsed once and dispatched to the typed
:meth:_filter hook; the engine-side filter then runs in
its native kernel (Arrow C++, Spark Catalyst) so the row
scan stays vectorised.
cast ¶
Cast rows, returning a new :class:Tabular.
Accepts a :class:Schema or :class:CastOptions. When
options is given, reads to arrow and casts each batch
through :meth:CastOptions.cast_arrow_batch.
display ¶
Render the first n rows as an aligned, typed text table.
Columns and their types come from this Tabular's own
:meth:collect_schema — the header is two rows: the column names,
then their type tags (the project :class:~yggdrasil.data.Field's
:meth:Field.short → :meth:DataType.short, recursive for nested types
— i64 / str / list<str> / struct<name:str, age:i64>).
Columns are separated by │ with a ─┼─ rule; numbers/booleans
right-align; nested cell values are compacted to one line. Long values
and headers are clipped (cells to max_width, type/name tags to a
slightly larger cap) so one long string or column name can't balloon the
table. The n rows are pushed down as a row_limit so no more than
that is ever read.
print(dbc.sql.execute("SELECT * FROM t").display())
print(IO.from_("data.parquet").display(5))
lazy ¶
Return a :class:LazyTabular wrapping this source.
Transformations on the returned object (select, filter,
join, …) accumulate in an :class:ExecutionPlan without
touching data. Any read_* call materialises the plan.