`yggdrasil.data.options`¶

options ¶

CastOptions — the one options object every :class:DataIO takes.

What it carries¶

Two :class:Field s (source, target) — the inferred / desired schema at each end of a cast. Because :class:Schema is a subclass of :class:StructField which is a :class:Field, the same slot covers "I have a single column" and "I have a full schema" — the type just promotes through :meth:Field.from_ on construction (pa.Schema lands as :class:StructField, pa.DataType lands as a leaf :class:Field). Callers that want a Schema-shaped view specifically can call .target.to_schema() / .source.to_schema().

A safe flag for strict-vs-permissive semantics (overflow, truncation, nulls-in-non-nullable). Sizing knobs row_size / byte_size that batch-oriented readers/writers honour. A :class:Mode pair for write semantics: mode controls the data write (overwrite / append / error-if-exists), schema_mode controls how schema drift is handled. An optional arrow_memory_pool for callers routing allocations through a bounded pool.

The canonical entry point¶

:meth:CastOptions.check is what every :class:DataIO public method funnels through. It accepts any of: an existing CastOptions to reuse, a dict of overrides, a :class:pa.DataType / :class:pa.Field / :class:pa.Schema to promote to a target hint, or None for defaults. It merges everything into a single :class:CastOptions instance — immutable from there, so no per-call mutation hazards.

`...` sentinel¶

:data:... distinguishes "caller didn't pass this" from "caller passed None". The latter is a real value (e.g., row_size=None means "no row cap"); the former should inherit whatever the base options had. :meth:strip_... drops ...-valued keys from a mapping so .check() doesn't overwrite existing values with "I didn't say anything."

Field normalization¶

__post_init__ runs every field-shaped input through :meth:Field.from_, so callers can pass a :class:pa.Schema, a :class:pa.Field, a :class:pa.DataType, a yggdrasil :class:Schema or :class:Field, or a dict spec — all land as a uniform :class:Field. Stops CastOptions(target=pa_schema) from propagating an un-wrapped pyarrow object into the casting engines.

Engine dispatch¶

:meth:cast and the per-engine / per-shape variants all delegate to the matching :class:Field methods — :class:Field owns the dispatch table (engine detection via :meth:ObjectSerde.module_and_name, shape detection via isinstance under the engine's lazy-imported module). :class:CastOptions adds exactly two things on top:

The target is None short-circuit — valid on options but not on a Field, which always carries a dtype. Lets :meth:CastOptions.cast pass obj through unchanged when no target is bound.
options=self plumbing — callers threading options through a long pipeline don't have to re-specify at each cast site.

Single source of truth for the dispatch table: adding a new engine is a one-site edit in :class:Field.

CastOptions `dataclass` ¶

CastOptions(
    source: "Field | None" = None,
    target: "Field | None" = None,
    safe: bool = False,
    checked_cast: bool = False,
    mode: Mode = Mode.AUTO,
    schema_mode: Mode = Mode.AUTO,
    row_size: int | None = None,
    byte_size: int | None = None,
    row_limit: int | None = None,
    use_threads: bool = True,
    match_by: list["Field"] | None = None,
    unique_by: list["Field"] | None = None,
    time_sample_by: list["Field"] | None = None,
    fill_strategy: str = "ffill",
    predicate: Predicate | None = None,
    wait: WaitingConfig = WaitingConfig.default(),
    spark_session: "SparkSession | None" = None,
    arrow_memory_pool: MemoryPool | None = None,
    update_column_names: list[str] | None = None,
    zorder_by: list[str] | None = None,
    optimize_after_merge: bool = False,
    vacuum_hours: int | None = None,
    retry: WaitingConfigArg | None = None,
    return_data: bool = False,
    safe_merge: bool = False,
    sync_metadata: bool = True,
)

Options carried through every :class:DataIO read and write.

Frozen so a single instance can safely be shared across threads / tasks; mutation requires :meth:copy. Slotted for cheap construction on the hot path (hundreds of options objects get built per batched write in a folder-of-folders persist).

All fields default to safe no-ops:

source / target = None → no cast coercion. :meth:cast returns inputs unchanged.
safe = False → permissive cast: bad rows / overflow become null. Strict semantics are opt-in via safe=True.
mode / schema_mode = :attr:Mode.AUTO → writer picks the appropriate behaviour from context.
row_size / byte_size = None → no batch caps; readers stream whatever size is natural for the format.
arrow_memory_pool = None → use pyarrow's default pool.

merged `property` ¶

merged: Field | None

Target reconciled with source under :attr:schema_mode.

With the default :attr:Mode.AUTO, a target field that matches a source field by name is merged — so variant (ObjectType / NullType) target slots adopt the source dtype — while the target's field set is preserved (source-only columns are not pulled in). This is the schema casts coerce to: see the cast_* dispatchers, which run against merged rather than the raw target so a bare columns= projection autotypes against the bound source.

column_names `property` ¶

column_names: list[str] | None

The target field's column names, if a target field is bound.

match_by_keys `property` ¶

match_by_keys: list[str] | None

Resolved key column names to dedup on.

Pulls the :attr:Field.name of each entry in :attr:match_by. Returns None when no keys are set so callers can branch on "keys vs no-keys" with a single truthiness check.

select_source_column_names ¶

select_source_column_names() -> list[str] | None

The source field's column names, if a source field is bound.

read_columns ¶

read_columns() -> list[str] | None

Columns a source reader must keep — the projection plus the predicate's columns.

:attr:column_names is what the read should end up with, but the predicate row-filter runs before the cast projects down to it, so any column the predicate touches has to survive the read even when the caller didn't ask for it (columns=["a"] + predicate on b). None means no projection — read everything.

check `classmethod` ¶

check(options: CastOptionsArg = None, /, **overrides: Any) -> T

Canonical entry point — coerce anything into a :class:CastOptions.

Dispatch by what options is:

None — construct fresh :class:CastOptions(**overrides).
:class:CastOptions — if no overrides given, return it unchanged; if overrides given, .copy(**overrides).
:class:Mapping (including dict) — merge into overrides and construct fresh (override args win on key collision).
:class:pa.DataType / :class:pa.Field / :class:pa.Schema / :class:Field / :class:Schema — treat as a target hint. Equivalent to check(target=options, **overrides).

source= / target= go straight into the dataclass slots (after :meth:Field.from_ normalization in __post_init__). Callers that want the peek-and-bind "only set if not already bound" semantic should chain :meth:check_source / :meth:check_target after the call.

columns= shortcut: a sequence of column names describing the desired output projection. When a target is already bound the names narrow it (target.select(columns)); otherwise they are promoted to a struct-shaped target field whose children default to :class:ObjectType — a "I want these columns, leave their types alone" placeholder that drives projection without casting. It lands on target (not source) so it never shadows the real source schema inferred at read time.

:raises TypeError: if options is a type the dispatch table doesn't cover.

field_names `cached` `classmethod` ¶

field_names() -> frozenset[str]

Frozenset of this class's constructor-accepting field names.

Used by :meth:_build to filter **overrides down to keys the constructor will accept — callers funnel mixed kwargs through .check() (DataIO public methods often pass user kwargs straight through), and we don't want a stray filter= or columns= to crash construction.

Excludes init=False fields (the private memoization slots for merged / merged_schema); those are not valid __init__ keywords and a copy via dataclasses.replace would crash if it tried to forward them.

Cached per-class via :func:functools.cache so subclasses with extra fields get their own expanded set on first access.

copy ¶

copy(**overrides: Any) -> T

Return a copy with overrides applied.

... values in overrides are ignored (keep existing). Pass source=/target= to swap either slot — :class:Field normalization runs in __post_init__ so any :class:Field-shaped input (pa.Schema, pa.DataType, dict, …) is accepted.

Implementation note: bypasses :func:dataclasses.replace, which rebuilds via cls(**all_fields) and pays a full __init__ + __post_init__ traversal even when the caller only tweaked a single bool. Cast pipelines call copy repeatedly per batched write (with_source / check_source / with_target all funnel through here), so the fast clone below — :func:object.__new__ + slot copy + targeted __post_init__ normalization for the overridden keys — is meaningfully cheaper.

check_source ¶

check_source(obj: Any = None, *, copy: bool = True) -> T

Bind a :attr:source if one isn't already set.

Two ways to supply one:

source= on :meth:check / :meth:copy — explicit Field / Schema / pa type. Wins even if self.source is already set (explicit override).
obj= here — a peekable object. Only runs the peek when self.source is currently None — an already- bound field is never clobbered by a peek.

Returns self unchanged when neither is given. Used from :class:DataIO methods (collect_schema, read_arrow_dataset) that want to pin a source schema before running a batch walk.

checked_cast=True short-circuits — the caller guarantees the batch shape matches the target, so the peek (which would rebuild a yggdrasil :class:Field from the batch's :class:pa.Schema) is wasted work. Combined with the :meth:cast_arrow_tabular short-circuit, this collapses every per-batch cast pass to a single attribute read on the leaf write path — ~150 us / batch saved on a RESPONSE_SCHEMA-shaped write.

check_target ¶

check_target(obj: Any = ..., *, copy: bool = True) -> T

Bind a :attr:target if one isn't already set.

Symmetry partner for :meth:check_source. See that method for the argument semantics — source/target behave identically.

with_source ¶

with_source(source: 'Field', copy: bool = False) -> T

Return a copy with source as the new source field.

Accepts the same shapes :meth:Field.from_ does (pa schema, yggdrasil Field, dict, etc.) — normalized in __post_init__ via :func:dataclasses.replace. The frozen slot is updated through :func:object.__setattr__ in the post-init hook; we don't bypass it here because going through replace gets the normalization for free.

with_target ¶

with_target(target: 'Field', copy: bool = True) -> T

Return a copy with target as the new target field.

with_checked_cast ¶

with_checked_cast(value: bool = True, copy: bool = False) -> T

Return a copy (or in-place) with :attr:checked_cast set.

Mirror of :meth:with_source / :meth:with_target — keeps the per-call mutation behind a named method instead of having every writer-side caller reach for :func:dataclasses.replace / :func:object.__setattr__. Set when the caller knows every batch already matches the target (came from a :class:pa.Table, a :class:pa.RecordBatchReader, a polars / pandas frame, or another writer that just emitted the same schema); the leaf's :meth:check_source / :meth:cast_arrow_tabular then short-circuit straight to the write path.

need_cast ¶

need_cast(
    source: Any | None = None,
    target: Any | None = None,
    check_names=False,
    check_dtypes=True,
    check_metadata=False,
    check_nullable: bool = False,
) -> bool

Return True if source and target fields differ enough to need casting.

When either field is unbound, returns False — there's nothing to compare against, so assume caller already sorted it.

Field equality semantics are the :meth:Field.equals rules: names, dtypes, metadata — each independently gateable. Metadata is off by default because it's commonly decorative (pandas preserves indices through metadata, arrow carries codec hints in field metadata) and comparing on it would demand a cast for cosmetic differences.

check_nullable is off by default because nullability rarely warrants a real value-level cast — primitives and lists pass through unchanged when only the flag differs. Tabular / struct casts pass check_nullable=True so the rebuild fires when child fields differ on nullability: Spark / Delta refuse to implicitly cast nullable→NOT NULL inside a struct (even when the data is in fact non-null), so the cast has to emit the target's field types verbatim to keep MERGE happy.

finalize ¶

finalize(obj: Any, *, default_scalar: Any = None) -> Any

Finalize any object — delegates to :meth:Field.finalize.

finalize_spark_cast ¶

finalize_spark_cast(obj: Any, *, default_scalar: Any = None) -> Any

Fill nulls and alias a Spark Column to the target name.

Direct parallel of :meth:finalize_polars_cast — Spark Columns, like polars Series/Expr, carry a name that can diverge from the target after a cast chain, so the alias step belongs in finalize rather than in each cast site.

finalize_arrow_cast ¶

finalize_arrow_cast(obj: Any, *, default_scalar: Any = None) -> Any

Fill nulls on a pyarrow object to finish a cast chain.

No alias step: :class:pa.Array / :class:pa.ChunkedArray don't carry a name, and tabular rename (Table/RecordBatch) is a schema-level rebuild that :meth:cast_arrow_tabular already handles inline via the target schema. Finalize here just means "apply the default-scalar null fill."

finalize_pandas_cast ¶

finalize_pandas_cast(obj: Any, *, default_scalar: Any = None) -> Any

Fill nulls on a pandas object to finish a cast chain.

No alias step exposed on :class:CastOptions for pandas — Series .name and DataFrame column labels get set by the cast methods directly. Finalize is fill-only, matching :meth:finalize_arrow_cast.

cast ¶

cast(obj: Any) -> Any

Cast obj to :attr:target using its native engine.

Dispatches arrow types through :meth:cast_arrow, Tabular through :meth:cast_tabular. Everything else delegates to :meth:Field.cast.

cast_pyarrow ¶

cast_pyarrow(obj: Any) -> Any

Cast any pyarrow object — delegates to :meth:Field.cast_arrow.

cast_arrow_array ¶

cast_arrow_array(array: Any) -> Any

Cast a :class:pa.Array or :class:pa.ChunkedArray.

cast_arrow_batch ¶

cast_arrow_batch(batch: 'pa.RecordBatch') -> 'pa.RecordBatch'

Filter + cast a :class:pa.RecordBatch.

cast_arrow_table ¶

cast_arrow_table(table: 'pa.Table') -> 'pa.Table'

Filter + cast a :class:pa.Table.

cast_arrow_tabular ¶

cast_arrow_tabular(data: 'ArrowTabular') -> 'ArrowTabular'

Filter + cast an :class:ArrowTabular (batch by batch).

cast_arrow ¶

cast_arrow(data: Any) -> Any

Dispatch arrow types to the specific cast method.

cast_tabular ¶

cast_tabular(data: Any) -> Any

Cast any Tabular-like object.

dedup_columns_on_read ¶

dedup_columns_on_read() -> 'list[str]'

Return the column names that need client-side dedup at read time.

Sourced from :attr:unique_by — each Field's :attr:Field.name is the column the read pass must deduplicate on. Returns an empty list when :attr:unique_by is unset / empty.

dedup_arrow_batches ¶

dedup_arrow_batches(
    batches: "Iterator[pa.RecordBatch]",
) -> "Iterator[pa.RecordBatch]"

Collapse duplicate rows on the columns flagged unique.

Resolves the dedup column set via :meth:dedup_columns_on_read, then delegates to :func:yggdrasil.arrow.ops.dedup_arrow_batches for the pure-Arrow group-by + take pass. Identity short-circuit when no column needs collapsing keeps the read path zero-cost on the common case (no target / no unique column / source already unique).

resample_on_read ¶

resample_on_read() -> 'tuple[str, int, list[str], str] | None'

Return (time_column, sampling_seconds, partition_by, fill_strategy) to resample.

Picks the first entry of :attr:time_sample_by whose time_sampling metadata carries a positive ISO-8601 duration. The result drives :func:yggdrasil.arrow.ops.resample_arrow_table — a single (column, interval) is all that op consumes (you can only have one time axis per table to resample on at a time).

Each Field's sampling lives under its :attr:Field.metadata's non-prefixed b"time_sampling" key as an ISO-8601 duration string ("PT1H" / "P1D"). The non-prefixed key keeps the value off the schema-level tag registry (it's a per-call option, not a contract that rides with the data on disk).

partition_by is derived from the target schema's :attr:Field.primary_key set, minus the resample column itself if it's also primary. The rationale: on a per-entity time series (one symbol per row, partitioned by symbol), each entity's timeline should bucket independently — without partition_by the resample would collapse rows across instruments. Schemas with no primary keys (or where the only primary is the timestamp) fall back to a flat resample.

Returns None when :attr:time_sample_by is unset / empty or every listed Field's metadata fails to parse.

resample_arrow_batches ¶

resample_arrow_batches(
    batches: "Iterator[pa.RecordBatch]",
) -> "Iterator[pa.RecordBatch]"

Snap rows to the target's time_sampling grid.

Resolves the resample column / interval / partition keys via :meth:resample_on_read, then delegates to :func:yggdrasil.arrow.ops.resample_arrow_batches. Identity short-circuit when no field is flagged keeps the read path zero-cost on the common case.

apply_post_read_table ¶

apply_post_read_table(table: 'pa.Table') -> 'pa.Table'

Run column projection + resample + dedup on a materialised :class:pa.Table.

Same operations and same order as the streaming wraps — column projection first (trim I/O cost before any compute), resample second (its bucket collapse trims rows before the unique-tag walk), then dedup. Identity short-circuit when no pass is configured so the common case stays zero-cost.

Pyarrow / polars / pandas read paths that already produce a Table funnel through this method instead of the iterator wraps; the result is one Table.from_batches + one Table.take (per pass) instead of two Table.from_batches + a Table.to_batches rebatch sandwich.

apply_post_read_spark_frame ¶

apply_post_read_spark_frame(df: Any) -> Any

Run resample + dedup directly on a Spark DataFrame.

Spark-side mirror of :meth:apply_post_read_table — same op order (resample first, dedup second), same identity short-circuit when neither is configured. Routes through :mod:yggdrasil.spark.ops so the heavy lifting stays on the executors (groupBy + applyInArrow for the partitioned resample, SQL window functions otherwise) instead of collecting the frame to the driver as Arrow.

Used by :meth:yggdrasil.spark.tabular.Dataset._read_spark_frame to apply the read-time passes before handing the frame back — saving a full df.toArrow → arrow.ops → createDataFrame round trip per configured op.

cast_arrow_batch_iterator ¶

cast_arrow_batch_iterator(batches: Any) -> Any

Cast a stream of :class:pa.RecordBatch and rechunk by byte_size / row_size.

With a bound target: per-batch tabular cast + streamed rechunking via :meth:Field.cast_arrow_batch_iterator (which routes through the struct-side helper).

Without a target: rechunk-only when byte_size / row_size is set, otherwise passthrough. Lets callers that did an in-engine cast upstream still pick up the optimized rechunker.

fill_arrow_nulls ¶

fill_arrow_nulls(obj: Any, *, default_scalar: Any = None) -> Any

Engine-level null fill — delegates to :meth:Field.fill_arrow.

fill_arrow_array_nulls ¶

fill_arrow_array_nulls(array: Any, *, default_scalar: Any = None) -> Any

Narrow null fill for a :class:pa.Array / :class:pa.ChunkedArray.

cast_polars ¶

cast_polars(obj: Any) -> Any

Cast any polars object — delegates to :meth:Field.cast_polars.

cast_polars_series ¶

cast_polars_series(series: Any, *, default_scalar: Any = None) -> Any

Cast a :class:pl.Series.

cast_polars_expr ¶

cast_polars_expr(expr: Any, *, default_scalar: Any = None) -> Any

Cast a :class:pl.Expr.

Wraps the expression tree with a cast operator — actual work fires when the containing LazyFrame is collected.

cast_polars_tabular ¶

cast_polars_tabular(data: Any) -> Any

Cast a :class:pl.DataFrame or :class:pl.LazyFrame.

fill_polars_nulls ¶

fill_polars_nulls(obj: Any, *, default_scalar: Any = None) -> Any

Engine-level polars null fill — delegates to :meth:Field.fill_polars.

polars_alias ¶

polars_alias(obj: Any) -> Any

Rename a polars Series/Expr to the target name — no-op if matching.

Delegates to :meth:Field.polars_alias. When target is unbound there's no name to rename to, so we pass through.

cast_pandas ¶

cast_pandas(obj: Any) -> Any

Cast any pandas object — delegates to :meth:Field.cast_pandas.

fill_pandas_nulls ¶

fill_pandas_nulls(obj: Any, *, default_scalar: Any = None) -> Any

Engine-level pandas null fill — delegates to :meth:Field.fill_pandas.

cast_spark_frame ¶

cast_spark_frame(df: Any) -> Any

Filter + cast a Spark DataFrame.

Applies :attr:predicate (when set) then delegates to :meth:Field.cast_spark_tabular for schema coercion.

cast_spark_tabular ¶

cast_spark_tabular(data: Any) -> Any

Filter + cast a :class:Dataset (Spark Tabular wrapper).

cast_spark_column ¶

cast_spark_column(obj: Any) -> Any

Cast a Spark Column.

cast_spark ¶

cast_spark(obj: Any) -> Any

Dispatch spark types to the specific cast method.

fill_spark_nulls ¶

fill_spark_nulls(obj: Any, *, default_scalar: Any = None) -> Any

Engine-level spark null fill — delegates to :meth:Field.fill_spark.

spark_alias ¶

spark_alias(obj: Any) -> Any

Rename a Spark Column to the target name — delegates to :meth:Field.spark_alias.

timedelta_to_iso_duration ¶

timedelta_to_iso_duration(td: timedelta) -> str

Render a :class:datetime.timedelta as an ISO-8601 duration string.

Inverse of :func:yggdrasil.data.types.primitive.temporal._parse_iso_duration, minus the calendar units (Y/M/W) that the parser collapses on the way in. Useful for stamping b"time_sampling" metadata on a :class:Field before handing it to :attr:CastOptions.time_sample_by::

from yggdrasil.data import field
from yggdrasil.data.options import timedelta_to_iso_duration
import datetime as dt, pyarrow as pa

f = field("ts", pa.timestamp("us", "UTC"),
          metadata={b"time_sampling":
                    timedelta_to_iso_duration(dt.timedelta(hours=1)).encode()})

yggdrasil.data.options¶