yggdrasil.databricks.cluster

cluster

Databricks cluster resource + service.

:class:Cluster is the cluster handle for all-purpose / classic compute.

:class:ClusterStatementExecutor exposes a cluster as a backing for yggdrasil.data.executor.StatementExecutor — i.e. lets the cluster stand in for a SQL warehouse when an :class:SQLEngine needs a cluster-driven SQL path.

Cluster

Cluster(
    service: Clusters | None = None,
    cluster_id: str | None = None,
    cluster_name: str | None = None,
    *,
    details: Optional[ClusterDetails] = None,
    details_refresh_time: float = 0.0,
    contexts: dict[str, ExecutionContext] = None,
    singleton_ttl: Any = ...
)

Bases: Singleton, DatabricksResource, URLBased

High-level Databricks cluster helper.

Parameters

service Cluster service wrapper used to resolve and validate cluster specs. cluster_id Existing Databricks cluster id. cluster_name Existing cluster name. If provided without cluster_id, the cluster is resolved on initialization.

Notes

Cluster caches ClusterDetails and refreshes them lazily. Inherits :class:Singleton (_SINGLETON_TTL = None) so two callers asking for the same cluster under the same service share the live ClusterDetails cache and the per-cluster execution contexts.

URL-addressable through :class:URLBased under :attr:Scheme.DATABRICKS_CLUSTER (dbks+cluster://): a cluster round-trips through dbks+cluster://<host>/<cluster_id> so a caller with just the URL can rebuild the live handle via :meth:URLBased.dispatch / :meth:Cluster.from_url.

sql property

sql: 'SQLEngine'

Shorthand for self.service.client.sql — the active :class:SQLEngine.

explore_url property

explore_url: URL

Workspace UI URL pointing at this cluster's compute page.

details property

details: Optional[ClusterDetails]

Return cached cluster details.

If no details are cached and the cluster id is known, details are fetched lazily from the Databricks API.

state property

state: State

Return the latest cluster state.

is_running property

is_running: bool

Whether the cluster is currently running.

is_pending property

is_pending: bool

Whether the cluster is in a transitional state.

is_error property

is_error: bool

Whether the cluster is in an error state.

spark_version property

spark_version: Optional[str]

Return the configured Databricks runtime string.

runtime_version property

runtime_version: Optional[str]

Return the major.minor Databricks runtime version.

Example

14.3.x-scala2.12 -> 14.3

python_version_info property

python_version_info: Optional[VersionInfo]

Return the Python version mapped from the Databricks runtime.

for_scheme classmethod

for_scheme(scheme: Any) -> 'type[URLBased]'

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

dispatch(url: Any, **kwargs: Any) -> 'URLBased'

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

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.

url

url() -> URL

Deprecated alias for :attr:explore_url (method form).

to_url

to_url() -> URL

Render this cluster as dbks+cluster://<host>/<cluster_id>.

The host comes from the bound :class:DatabricksClient; the single path segment is the cluster id. cluster_name is not emitted — the canonical identity for a cluster is its id, and a consumer that needs the name can refresh from the live details.

from_url classmethod

from_url(url: 'URL | str', **kwargs: Any) -> 'Cluster'

Build a :class:Cluster from a dbks+cluster://host/<id> URL.

Resolves the underlying :class:DatabricksClient from the URL host (or :meth:DatabricksClient.current when the URL carries no host). The first path segment is the cluster id; trailing segments are tolerated for forward compatibility but ignored.

clusters_client

clusters_client() -> ClustersAPI

Return the Databricks SDK clusters client.

libraries_client

libraries_client()

Return the Databricks SDK libraries client.

fresh_details

fresh_details(max_delay: float | None = None) -> Optional[ClusterDetails]

Return cluster details, refreshing cache when stale.

Parameters

max_delay Maximum allowed age of cached details in seconds. None means refresh immediately.

refresh

refresh(max_delay: float | None = None) -> 'Cluster'

Refresh cached cluster details and return self.

set_details

set_details(details: Optional[ClusterDetails | Wait]) -> 'Cluster'

Update local cached details.

Accepts either a ClusterDetails object, a Databricks Wait handle, or None.

raise_for_status

raise_for_status() -> 'Cluster'

Raise DatabricksError if cluster is in error state.

wait_for_status

wait_for_status(
    wait: WaitingConfigArg = True, raise_error: bool = True
) -> "Cluster"

Wait until the cluster leaves its pending state.

Also waits for library installation completion after the cluster becomes stable.

update

update(
    *,
    single_user_name: str | None = None,
    libraries: Optional[list[Union[str, Library]]] = None,
    permissions: Optional[list[str | ClusterAccessControlRequest]] = None,
    wait: WaitingConfigArg = True,
    **cluster_spec: Any
) -> "Cluster"

Update cluster configuration.

Parameters

single_user_name Optional Databricks single-user assignment. libraries Libraries to install before / alongside update flow. permissions Cluster ACL entries. Strings are interpreted as: - email -> user permission - other string -> group permission wait Waiting behavior for cluster and library readiness. **cluster_spec Cluster edit spec supported by the service layer and Databricks SDK.

update_permissions

update_permissions(
    permissions: Optional[list[str | ClusterAccessControlRequest]] = None,
) -> "Cluster"

Update cluster permissions.

If a referenced group does not exist and Databricks reports it through a GroupName(...) error, the group is created and the operation is retried once.

ensure_running

ensure_running(wait: WaitingConfigArg = True) -> 'Cluster'

Ensure the cluster is running.

start

start(wait: WaitingConfigArg = True) -> 'Cluster'

Start the cluster if needed.

If the initial start call races with a transient state transition, the method waits once and retries.

restart

restart(wait: WaitingConfigArg = True) -> 'Cluster'

Restart the cluster if already running, otherwise start it.

delete

delete() -> None

Delete the cluster if it exists.

context

context(
    *,
    language: Optional[Language] = None,
    context_id: str | None = None,
    context_key: str | None = None
) -> ExecutionContext

Return an execution context for this cluster.

When context_key is provided, the context is cached and reused.

command

command(
    context: Optional[ExecutionContext | str] = None,
    command: Optional[str | Callable] = None,
    *,
    command_str: str | None = None,
    func: Optional[Callable] = None,
    language: Optional[Language] = None,
    command_id: str | None = None,
    environ: Optional[Union[Iterable[str], Dict[str, str]]] = None
) -> "CommandExecution"

Create a command execution bound to a reusable execution context.

install_libraries

install_libraries(
    libraries: Optional[List[Union[str, Library]]] = None,
    wait: WaitingConfigArg = True,
    pip_settings: Optional[PipIndexSettings] = None,
    remove_failed: bool = True,
    raise_error: bool = True,
) -> "Cluster"

Install libraries on the cluster.

String inputs are normalized as: - *.jar -> jar library - *.whl -> wheel library - *requirements.txt -> requirements file - otherwise -> PyPI package

PyPI packages whose bare name appears in :data:PIP_INSTALL_BLACKLIST are silently skipped to protect the Databricks runtime from destabilising overwrites (e.g. pyspark, tensorflow).

installed_library_statuses

installed_library_statuses()

Return Databricks library installation statuses for this cluster.

uninstall_libraries

uninstall_libraries(
    pypi_packages: Optional[list[str]] = None,
    libraries: Optional[list[Library]] = None,
    restart: bool = True,
) -> "Cluster"

Uninstall libraries from the cluster.

Parameters

pypi_packages Optional list of PyPI package names to remove. libraries Explicit library objects to uninstall. If provided, this takes precedence over pypi_packages filtering. restart Whether to restart the cluster after uninstalling.

wait_installed_libraries

wait_installed_libraries(
    wait: WaitingConfigArg = True,
    raise_error: bool = True,
    remove_failed: bool = True,
) -> "Cluster"

Wait until all cluster libraries finish resolving / installing.

Failed libraries can optionally be uninstalled automatically.

Clusters

Clusters(client: Optional[DatabricksClient] = None)

Bases: DatabricksService

wheels property

wheels: 'Wheels'

Wheel registry service (shorthand for client.wheels).

environments property

environments: 'Environments'

Base-environment service (shorthand for client.environments).

tables property

tables: 'Tables'

Collection-level Unity Catalog table service (shorthand for client.tables).

views property

views: 'Tables'

Alias for :attr:tables — :class:Table covers both managed/external tables and view-shaped securables.

catalogs property

catalogs: 'Catalogs'

Collection-level Unity Catalog hierarchy service (shorthand for client.catalogs).

schemas property

schemas: 'Schemas'

Collection-level Unity Catalog schema service (shorthand for client.schemas).

volumes property

volumes: 'Volumes'

Collection-level Unity Catalog volume service (shorthand for client.volumes).

genie property

genie: 'Genie'

Genie service (shorthand for client.genie).

ai property

ai: 'DatabricksAI'

Databricks AI umbrella service (shorthand for client.ai).

default_tags

default_tags(update: bool = True) -> dict[str, str]

Return default resource tags for Databricks assets.

Returns:

Type	Description
dict[str, str]	A dict of default tags.

default

default(*, raise_error: bool = False) -> Optional[Cluster]

The project's default all-purpose cluster — the single-user cluster named for the running client project (its capitalized display name, :attr:DatabricksClient.product_name), as provisioned by ygg databricks deploy. Resolves the existing cluster (no creation); None when it isn't deployed unless raise_error. The cluster sibling of dbc.environments.default() / warehouses.default().

ClusterPreparedStatement

ClusterPreparedStatement(
    text: str = "",
    *,
    key: Optional[str] = None,
    retry: Optional[WaitingConfigArg] = None,
    language: Optional[Language] = None,
    context_key: Optional[str] = None,
    output_path: Optional["VolumePath"] = None,
    **kwargs: Any
)

Bases: PreparedStatement

SQL statement routed through a Databricks cluster.

Adds two cluster-specific knobs:

• language — defaults to Language.SQL; the executor uses it to pick the REPL the command runs in.
• context_key — keyed reuse of an :class:ExecutionContext on the cluster. None opts into the cluster-side default (the executor picks one keyed off the bound volume).
• output_path — when non-None, the executor wrote INSERT OVERWRITE DIRECTORY <output_path> so the result can read its rows back from this Parquet folder. Set by the executor at prepare time; callers shouldn't pass it directly.

opened property

opened: bool

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

closed property

closed: bool

Inverse of :attr:opened.

retryable property

retryable: bool

Whether a non-None retry policy has been configured.

Convenience for the lifecycle code; self.retry is not None works equivalently.

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.

with_text

with_text(value: str, inplace: bool = False) -> 'PreparedStatement'

Return a copy with text replaced (or mutate in place).

with_retry

with_retry(
    retry: Optional[WaitingConfigArg], *, inplace: bool = False
) -> "PreparedStatement"

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

looks_like_query(text: Any) -> bool

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

from_(
    statement: "PreparedStatement | StatementResult | str",
) -> "PreparedStatement"

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

prepare(
    statement: "PreparedStatement | str", **kwargs: Any
) -> "PreparedStatement"

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

clear() -> None

Clear all state associated with this statement.

clear_temporary_resources

clear_temporary_resources() -> None

Unlink the staged output folder, if any.

ClusterStatementBatch

ClusterStatementBatch(
    executor: "StatementExecutor",
    statements: Optional[Iterable["PS | str"]] = None,
    parallel: int = 1,
    **kwargs: Any
)

Bases: StatementBatch

Batch of cluster-backed statements.

Inherits the base batch contract unchanged — submission goes through :meth:ClusterStatementExecutor.send; the batch only adds the typed result-class pin.

opened property

opened: bool

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

closed property

closed: bool

Inverse of :attr:opened.

text property

text: str

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.

raise_for_status

raise_for_status() -> 'StatementBatch'

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.

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.

cancel

cancel(
    wait: WaitingConfigArg = False, raise_error: bool = False, **kwargs
) -> "StatementBatch"

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.

progress

progress() -> 'float | None'

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

watch(
    on_tick: "Any" = None, *, interval: float = 0.1, raise_error: bool = True
) -> "Awaitable"

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

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

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.

mark_dirty

mark_dirty() -> None

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

for_scheme classmethod

for_scheme(scheme: Any) -> 'type[URLBased]'

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

dispatch(url: Any, **kwargs: Any) -> 'URLBased'

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

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.

matches_static

matches_static(
    predicate: "Predicate", *, free_cols: "tuple[str, ...] | None" = None
) -> bool

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 (None when default=None).
• :class:Tabular — returned as-is. When as_folder is True and obj is a local :class:Path, wraps it in a :class:Folder.
• str / :class:os.PathLike — coerced via :class:Path.from_. When as_folder is True, 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

options_class() -> 'type[O]'

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

check_options(
    options: "O | None" = None, overrides: "dict | None" = None, **kwargs: Any
) -> O

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

cleanup(wait: 'Any' = False) -> int

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

optimize(byte_size: 'int | None' = None, **kwargs: Any) -> int

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

collect_schema(options: 'O | None' = None, **kwargs: Any) -> 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

count(options: 'O | None' = None, **kwargs: Any) -> int

Return the number of rows in this tabular.

scan_arrow_batches

scan_arrow_batches(
    options: "O | None" = None, **kwargs: Any
) -> Iterator[pa.RecordBatch]

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

scan_arrow_table(options: 'O | None' = None, **kwargs: Any) -> pa.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

scan_arrow_batch_reader(
    options: "O | None" = None, **kwargs: Any
) -> "pa.RecordBatchReader"

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_table(options: 'O | None' = None, **kwargs: Any) -> 'Tabular | None'

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

write_table(obj: Any, options: 'O | None' = None, **kwargs: Any) -> None

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

union(other: 'Any', *, mode: 'ModeLike | None' = None) -> 'Tabular'

Return a Tabular representing self UNION ALL other.

mode controls how mismatched schemas are reconciled:

• Mode.IGNORE (default) — keep self'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_spark_dataset(options: 'O | None' = None, **kwargs: Any) -> 'SparkDataset'

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

read_record_iterator(
    options: "O | None" = None, **kwargs: Any
) -> "Iterator[Mapping[str, Any]]"

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

read_records(options: 'O | None' = None, **kwargs: Any) -> 'Iterator[Any]'

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

unique(by: 'str | Any | Iterable[Any]') -> 'Tabular'

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

select(*columns: 'str | Any') -> 'Tabular'

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

drop(*columns: 'str | Any') -> 'Tabular'

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

filter(predicate: 'PredicateLike') -> 'Tabular'

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:Predicate node (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(options: 'O | None' = None, **kwargs) -> 'Tabular'

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

display(n: int = 10, *, max_width: int = 32) -> str

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

lazy() -> 'LazyTabular'

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.

clear_temporary_resources

clear_temporary_resources() -> 'StatementBatch'

Release per-result scratch. Does not cancel or drop anything.

add

add(statement: 'PS | str', key: Optional[str] = None) -> str

Enqueue a statement; return its key.

key collisions (against pending statements or completed results) raise :class:ValueError.

extend

extend(statements: Iterable['PS | str']) -> list[str]

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(key: str) -> Optional[SR]

Remove an entry by key.

Pending statement → dropped, None returned. In-flight result → cancelled, scratch released, instance returned. Unknown key → :class:KeyError.

clear

clear() -> 'StatementBatch'

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.

materialized

materialized() -> Iterator[tuple[str, SR]]

Yield (key, result) pairs for every submitted result.

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.

ClusterStatementResult

ClusterStatementResult(
    executor: "ClusterStatementExecutor",
    statement: Optional[ClusterPreparedStatement] = None,
    *,
    command: Optional["CommandExecution"] = None,
    **kwargs: Any
)

Bases: StatementResult

Single cluster command tracked as a :class:StatementResult.

Wraps a :class:CommandExecution plus the SQL bound to it. The :meth:start / :meth:cancel / :meth:refresh_status hooks forward to the command; :meth:_read_arrow_batches reads the staged Parquet directory back when the statement was a SELECT rewritten through INSERT OVERWRITE DIRECTORY.

opened property

opened: bool

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

closed property

closed: bool

Inverse of :attr:opened.

progress

progress() -> 'float | None'

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

watch(
    on_tick: "Any" = None, *, interval: float = 0.1, raise_error: bool = True
) -> "Awaitable"

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

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

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.

mark_dirty

mark_dirty() -> None

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

for_scheme classmethod

for_scheme(scheme: Any) -> 'type[URLBased]'

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

dispatch(url: Any, **kwargs: Any) -> 'URLBased'

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

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.

matches_static

matches_static(
    predicate: "Predicate", *, free_cols: "tuple[str, ...] | None" = None
) -> bool

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 (None when default=None).
• :class:Tabular — returned as-is. When as_folder is True and obj is a local :class:Path, wraps it in a :class:Folder.
• str / :class:os.PathLike — coerced via :class:Path.from_. When as_folder is True, 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

options_class() -> 'type[O]'

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

check_options(
    options: "O | None" = None, overrides: "dict | None" = None, **kwargs: Any
) -> O

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

cleanup(wait: 'Any' = False) -> int

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

optimize(byte_size: 'int | None' = None, **kwargs: Any) -> int

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

collect_schema(options: 'O | None' = None, **kwargs: Any) -> 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

count(options: 'O | None' = None, **kwargs: Any) -> int

Return the number of rows in this tabular.

scan_arrow_batches

scan_arrow_batches(
    options: "O | None" = None, **kwargs: Any
) -> Iterator[pa.RecordBatch]

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

scan_arrow_table(options: 'O | None' = None, **kwargs: Any) -> pa.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

scan_arrow_batch_reader(
    options: "O | None" = None, **kwargs: Any
) -> "pa.RecordBatchReader"

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_table(options: 'O | None' = None, **kwargs: Any) -> 'Tabular | None'

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

write_table(obj: Any, options: 'O | None' = None, **kwargs: Any) -> None

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

union(other: 'Any', *, mode: 'ModeLike | None' = None) -> 'Tabular'

Return a Tabular representing self UNION ALL other.

mode controls how mismatched schemas are reconciled:

• Mode.IGNORE (default) — keep self'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_spark_dataset(options: 'O | None' = None, **kwargs: Any) -> 'SparkDataset'

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

read_record_iterator(
    options: "O | None" = None, **kwargs: Any
) -> "Iterator[Mapping[str, Any]]"

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

read_records(options: 'O | None' = None, **kwargs: Any) -> 'Iterator[Any]'

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

unique(by: 'str | Any | Iterable[Any]') -> 'Tabular'

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

select(*columns: 'str | Any') -> 'Tabular'

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

drop(*columns: 'str | Any') -> 'Tabular'

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

filter(predicate: 'PredicateLike') -> 'Tabular'

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:Predicate node (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(options: 'O | None' = None, **kwargs) -> 'Tabular'

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

display(n: int = 10, *, max_width: int = 32) -> str

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

lazy() -> 'LazyTabular'

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.

ClusterStatementExecutor

ClusterStatementExecutor(
    cluster: "Cluster",
    volume: "Volume",
    *,
    default_language: Language = Language.SQL,
    default_context_key: Optional[str] = None
)

Bases: DatabricksResource, StatementExecutor[ClusterPreparedStatement, ClusterStatementResult, ClusterStatementBatch]

Cluster-backed :class:StatementExecutor.

Wraps a :class:Cluster (the actual execution backend) and a :class:Volume (used to stage SELECT result Parquet folders). Inherits :meth:execute / :meth:execute_many / :meth:batch from the base — only :meth:_submit_statement is implemented locally, plus the SELECT-rewrite helper.

Singleton-cached on (cluster, volume) so every statement against the same staging surface lands on one executor — same REPL context, same cluster session, same per-instance cache. The cluster cap of 145 user execution contexts is one of the main reasons for this: an unintended duplicate executor is an unintended second context.

opened property

opened: bool

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

closed property

closed: bool

Inverse of :attr:opened.

sql property

sql: 'SQLEngine'

Shorthand for self.service.client.sql — the active :class:SQLEngine.

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.

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.

submit_command

submit_command(statement: ClusterPreparedStatement) -> 'CommandExecution'

Build a :class:CommandExecution from statement (no start).

Exposed so :class:ClusterStatementResult.start can mint the command lazily — the result owns the lifecycle, the executor only owns the construction.