yggdrasil.spark.tabular

tabular

In-memory :class:Tabular holding a (mutable) Spark DataFrame.

:class:Dataset is the canonical Spark surface. Two interleaved roles, both satisfied by one class:

• :class:Tabular contract — :meth:_read_arrow_batches / :meth:_write_arrow_batches plus the Spark-native :meth:_read_spark_frame / :meth:_write_spark_frame so the holder fits anywhere the IO surface expects a Tabular.
• Rich :class:pyspark.sql.DataFrame wrapper — schema-aware map / apply / filter / explode / cast over :meth:pyspark.sql.DataFrame.mapInArrow, schema inference off dynamic-mode (cloudpickled-object) frames. The user function and row payloads travel as cloudpickle (the serializer pyspark already ships), so executors need no ygg install; cluster-side libraries are provisioned by the environment (e.g. DatabricksEnv).

The held DataFrame is mutable: writes replace it (OVERWRITE) or union to it (APPEND). :meth:read_spark_frame returns the held frame untouched (no driver collect); :meth:read_arrow_batches falls back to df.toArrow().to_batches() (which does collect to the driver — fine when the frame is small enough, but check before reaching for it in a hot path).

Executor cache

:meth:persist leverages Spark's own :meth:pyspark.sql.DataFrame.persist with :data:pyspark.StorageLevel.MEMORY_AND_DISK so the partitions land on the executors' memory and spill to executor-local disk under pressure. The call is idempotent — when :attr:pyspark.sql.DataFrame.is_cached is already True the persist is skipped silently. :meth:unpersist mirrors the path: it calls :meth:pyspark.sql.DataFrame.unpersist first (when the frame was cached) before dropping the local reference, so an intentional persist → ... → unpersist round trip cleans up the executor cache too.

Dynamic vs typed

Two modes, distinguished by :attr:schema:

• Dynamic (schema is None) — the underlying Spark frame carries the single-column _pickle schema and rows are arbitrary pickled Python objects. Transforms (map, filter, apply, explode) see unpickled inner objects.
• Typed (schema set) — the underlying Spark frame matches schema.to_spark_schema(). Transforms receive dict rows and outputs are cast back through the Schema.cast_arrow chain driven by mapInArrow.

Backwards-compat aliases

:class:Dataset (this module) and yggdrasil.spark.frame.Dataset are kept as aliases for :class:Dataset so external imports of the legacy spellings keep resolving to the same class. self.df is a property over the underlying frame slot so call sites using either frame= / df= spelling keep working.

SparkDataset

SparkDataset(
    frame: Optional["SparkDataFrame"] = None,
    schema: "Schema | None" = None,
    *,
    df: Optional["SparkDataFrame"] = None,
    spark: Optional["SparkSession"] = None
)

Bases: Tabular[CastOptions]

:class:Tabular + Spark-DataFrame surface in one class.

The frame is the holder's only state; reads of :meth:_read_spark_frame return it as-is, writes mutate it in place. The Spark session is cached off the frame on construction (or set explicitly) so an empty buffer can still synthesize an empty DataFrame on read.

Wrap a Spark DataFrame, optionally with a yggdrasil schema.

Two accepted spellings: Dataset(frame=df) (the Tabular-style argument name) and Dataset(df=df) (the legacy Dataset spelling). Passing both raises.

frame property writable

frame: Optional['SparkDataFrame']

Currently-held Spark DataFrame, or None when empty.

df property writable

df: Optional['SparkDataFrame']

Alias for :attr:frame — keeps the legacy Dataset spelling working without forking the API surface.

schema property writable

schema: 'Schema | None'

Yggdrasil :class:Schema describing the frame, when set.

Reads through :attr:Tabular._schema_cache (the base-class slot) and surfaces the ... sentinel as None so the legacy Dataset "schema is None means dynamic mode" contract still holds.

is_dynamic property

is_dynamic: bool

True iff this holder is in dynamic (pickled-object) mode.

spark_schema property

spark_schema

The underlying Spark schema, or None when no frame held.

sparkSession property

sparkSession: 'SparkSession'

Bound :class:SparkSession. Raises when no session is reachable.

is_cached property

is_cached: bool

True iff the underlying Spark frame is currently persisted.

Mirrors :attr:pyspark.sql.DataFrame.is_cached and answers False whenever no frame is held — there's nothing to cache. :meth:persist skips when this is already True.

opened property

opened: bool

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

closed property

closed: bool

Inverse of :attr:opened.

cache

cache()

Cache the underlying frame on Spark executors.

persist

persist(
    engine: str = "auto",
    *,
    data: Any = None,
    storage_level: "StorageLevel | str | None" = None
) -> "SparkDataset"

Cache the underlying frame on Spark executors.

data= replaces the held frame first (legacy stash path used by the Statement materialization shims). storage_level= overrides the default :data:pyspark.StorageLevel.MEMORY_AND_DISK — pass "MEMORY_ONLY" / "DISK_ONLY" / a real StorageLevel when the workload calls for it.

Skip-if-cached: when :attr:pyspark.sql.DataFrame.is_cached already reads True (or no frame is held) the persist is a no-op. So a chain like df.persist().persist().persist() triggers one cache and the rest are free. Same skip path on a holder whose frame was persisted upstream — we don't double-stamp the executor-side cache.

unpersist

unpersist() -> None

Drop the held frame and release any executor-side cache.

When :attr:is_cached is true, :meth:pyspark.sql.DataFrame.unpersist is called first so the partitions are evicted from executor memory / disk. Failures during unpersist are swallowed (best-effort), then the local reference is dropped unconditionally — calling unpersist on an already-empty holder is fine.

from_spark_frame classmethod

from_spark_frame(
    df: "SparkDataFrame", schema: "Schema | None" = None
) -> "SparkDataset"

Wrap a Spark frame, optionally re-casting it against schema.

schema=None infers a yggdrasil :class:Schema from the Spark frame's schema. A non-None schema first runs the frame through :meth:Schema.cast_spark_tabular so the held frame matches the declared shape.

from_iterable classmethod

from_iterable(
    items: "Iterable[Any]",
    schema: "Schema | None" = None,
    *,
    spark_session: Optional["SparkSession"] = None,
    byte_size: int = 128 * 1024 * 1024
) -> "SparkDataset"

Build a frame from an in-memory iterable.

schema=None pickles each element into a dynamic frame. schema=<Schema> casts the iterable on the driver and returns a typed frame whose underlying DataFrame matches schema.

from_sql classmethod

from_sql(
    text: str,
    *,
    spark_session: Optional["SparkSession"] = None,
    schema: "Schema | None" = None
) -> "SparkDataset"

Execute SQL and wrap the resulting Spark frame as a :class:Dataset.

Resolves a :class:SparkSession through the environment when none is passed — in a Databricks context this picks up the active notebook / Connect / Job session automatically.

from_table classmethod

from_table(
    name: str,
    *,
    spark_session: Optional["SparkSession"] = None,
    schema: "Schema | None" = None
) -> "SparkDataset"

Read a table by its fully-qualified name (catalog.schema.table).

Thin wrapper around spark.table(name) that auto-resolves the session from the environment.

to_table

to_table(
    name: str,
    *,
    mode: str = "overwrite",
    format: str = "delta",
    partition_by: "list[str] | None" = None
) -> "SparkDataset"

Write the held frame to a Unity Catalog / Spark table.

Returns self so the call can be chained::

(Dataset.from_sql("SELECT ...")
 .map(transform)
 .to_table("main.curated.output"))

parallelize classmethod

parallelize(
    inputs: "Iterable[Any]",
    function: "Callable[..., Any] | None" = None,
    schema: "Schema | None" = None,
    *,
    spark_session: Optional["SparkSession"] = None,
    byte_size: int = 128 * 1024 * 1024
) -> "SparkDataset"

Distribute function over inputs via mapInArrow, or create a frame directly from inputs when no function is given.

Two call shapes:

• parallelize(inputs, function, schema=...) — apply function to each element of inputs on Spark executors.
• parallelize(inputs, schema=...) — wrap inputs as a :class:Dataset (delegates to :meth:from_iterable).

Per-input dispatch goes through :func:yggdrasil.dataclasses.build_row_invoker so any pyfunc shape is accepted (single-arg, multi-arg, **kwargs, *args); dict inputs spread as kwargs, list/tuple inputs spread as positional args when the function declares a *args catch-all. schema=None returns a dynamic frame of pickled outputs; schema=<Schema> casts outputs and returns a typed frame.

infer_schema

infer_schema(
    *, limit: "int | None" = None, force: bool = False, inplace: bool = True
) -> "Schema"

Infer a yggdrasil :class:Schema from the row contents.

Dynamic mode: each row is unpickled and shape-inferred via :meth:Schema.from_; per-partition schemas are merged in APPEND mode (union of fields, widening of nullability), then folded on the driver into the final schema.

Typed mode: returns :attr:schema unchanged unless force=True, in which case the underlying batches are re-inferred from row dicts — useful after a heterogeneous transform whose output schema is looser than the declared one.

map

map(
    function: "Callable[..., Any]",
    schema: "Schema | None" = None,
    *,
    byte_size: int = 128 * 1024 * 1024
) -> "SparkDataset"

1:1 map over rows.

Input rows are unpickled objects (dynamic mode) or row-dicts (typed mode). The function's signature drives the row-shape adaptation via :func:yggdrasil.dataclasses.build_row_invoker: single-arg functions get the row directly (or row[arg_name] when the arg name matches a key in a dict row), multi-arg / **kwargs functions get the dict spread as kwargs, *args catch-alls get sequence rows spread positionally, and annotated parameters are coerced through the :func:yggdrasil.data.cast.convert registry. Typed-mode batches go through :func:yggdrasil.dataclasses.build_batch_invoker so a single-positional-by-name function gets a vectorised column cast. Output schema follows schema if given, else the result is a dynamic frame.

apply

apply(
    function: "Callable[..., Any]",
    schema: "Schema | None" = None,
    *,
    byte_size: int = 128 * 1024 * 1024
) -> "SparkDataset"

Map function over each row, optionally casting against schema.

function may carry any signature — single-arg (def f(row): ...), multi-arg (def f(id: int, name: str): ...), **kwargs catch-all (def f(**row): ...), or a *args catch-all over tuple/list rows. Per-row dispatch is built once per partition via :func:yggdrasil.dataclasses.build_row_invoker, which:

• passes the row directly when function has one positional parameter,
• extracts row[arg_name] when the row is a mapping and the single positional arg name matches a column key — def f(id: int) on a {"id": ..., "name": ...} row gets the id value, not the whole dict,
• spreads mapping rows as **kwargs (filtered to declared names, unless **kwargs catches the rest) when function has multiple named parameters,
• spreads sequence rows as *args when function declares a *args catch-all and no other positional,
• coerces annotated parameters through :func:yggdrasil.data.cast.convert so string-shaped inputs reach the function as the annotated Python type.

Typed-mode batches additionally route through :func:yggdrasil.dataclasses.build_batch_invoker, which picks the cheapest dispatch shape per batch:

• whole-batch tabular — function annotated def f(batch: pa.RecordBatch) / pa.Table / pl.DataFrame / pl.LazyFrame / pd.DataFrame receives the whole batch converted to that type in one call (zero per-row Python overhead),
• column-by-name vectorised cast — single-positional annotated + arg name matches a column → the column is cast via :func:pyarrow.compute.cast (one C++ kernel call) and function runs per cell,
• per-row fallback — every other shape walks batch.to_pylist() rows through the row invoker.

Without a schema this is :meth:map. With a schema, function may return any tabular shape (dict, dataclass, list-of-rows, polars/pandas/arrow frame, pa.RecordBatch); outputs are streamed through :func:any_to_arrow_batch_iterator in one pass.

filter

filter(
    predicate: "Callable[[Any], bool] | Any",
    schema: "Schema | None" = None,
    *,
    byte_size: int = 128 * 1024 * 1024
) -> "SparkDataset"

Drop rows where predicate is false.

Two shapes, dispatched by the predicate's runtime type:

• Predicate-like — a SQL string, a yggdrasil :class:Expression / :class:Predicate node, or a native engine expression (:class:pyarrow.compute.Expression, :class:polars.Expr, :class:pyspark.sql.Column). Routes through the typed :meth:_filter hook; the cross-engine :meth:Tabular.filter contract — the predicate compiles to a :class:pyspark.sql.Column and Spark's Catalyst plans the filter natively.
• Pure callable (the legacy :class:Dataset shape) — predicate(row) evaluated per row inside a :meth:mapInArrow worker; sees unpickled objects on a dynamic frame or row-dicts on a typed frame. Kept for backwards compatibility — row-by-row callables are the last-resort path when no vectorised predicate fits.

When called on a typed frame without a schema argument the existing schema is preserved (no re-cast needed); the byte_size cap only applies to the callable path.

explode

explode(
    schema: "Schema | None" = None, *, byte_size: int = 128 * 1024 * 1024
) -> "SparkDataset"

Explode rows of iterables into one row per element.

Only meaningful in dynamic mode — typed rows are dicts, not iterables. Pass a schema to type the flattened output.

to_dynamic

to_dynamic(*, byte_size: int = 128 * 1024 * 1024) -> 'SparkDataset'

Drop typing: re-pickle row-dicts back into a dynamic frame.

No-op when already dynamic.

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.

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.