loom.etl.backends.polars

Polars backend for the Loom ETL framework.

class loom.etl.backends.polars.PolarsFileWriter

Bases: object

Write FILE targets with Polars DataFrame writers.

write(frame, spec, *, streaming=False)

Write frame to the path declared in spec.

Parameters:

• frame (polars.LazyFrame)

• spec (FileSpec)

• streaming (bool)

Return type:

None

class loom.etl.backends.polars.PolarsPhysicalSchema(schema, partition_columns=())

Bases: object

Physical schema snapshot for a resolved Polars target.

Parameters:

• schema (polars.Schema) – Native Polars schema for write-time alignment.

• partition_columns (tuple[str, ...]) – Ordered partition columns.

class loom.etl.backends.polars.PolarsSourceReader(locator, *, route_resolver=None, file_locator=None, mongo_reader=None, clickhouse_reader=None, dynamodb_reader=None)

Bases: SourceReader

Read table, file, and MongoDB sources using Polars.

Parameters:

• locator (str | TableLocator)

• route_resolver (TableRouteResolver | None)

• file_locator (FileLocator | None)

• mongo_reader (MongoSourceReader | None)

• clickhouse_reader (ClickHouseSourceReader | None)

• dynamodb_reader (DynamoDbSourceReader | None)

read(spec, params_instance, /)

Read source spec and return lazy frame.

Parameters:

• spec (TableSourceSpec | FileSourceSpec | TempSourceSpec | MongoSourceSpec | ClickHouseSourceSpec | DynamoDbSourceSpec)

• params_instance (Any)

Return type:

polars.LazyFrame

read_streaming(spec, params_instance, /)

Route streaming reads to the matching capability-aware sub-reader.

Only ClickHouseSourceSpec is supported today; other kinds raise TypeError. We refuse to silently fall back to read() because a step that opts into streaming for memory safety must fail loudly rather than risk OOM on a non-streaming path.

Parameters:

• spec (TableSourceSpec | FileSourceSpec | TempSourceSpec | MongoSourceSpec | ClickHouseSourceSpec | DynamoDbSourceSpec) – Compiled source specification.

• params_instance (Any) – Concrete params for current run.

Returns:

Lazy frame whose collect(engine="streaming") is memory-bounded.

Raises:

TypeError – When spec is not a ClickHouse spec, or the matching sub-reader was not configured at construction time.

Return type:

polars.LazyFrame

execute_sql(frames, query)

Execute SQL query against backend frames.

Parameters:

• frames (dict[str, Any])

• query (str)

Return type:

polars.LazyFrame

class loom.etl.backends.polars.PolarsTargetWriter(locator, *, route_resolver=None, missing_table_policy=MissingTablePolicy.SCHEMA_MODE, file_locator=None, audit_config=None)

Bases: _WritePolicy[LazyFrame, DataFrame, PolarsPhysicalSchema]

Polars target writer using delta-rs for Delta tables.

Parameters:

• locator (str | TableLocator)

• route_resolver (TableRouteResolver | None)

• missing_table_policy (MissingTablePolicy)

• file_locator (FileLocator | None)

• audit_config (AuditConfig | None)

append(frame, table_ref, params_instance, *, streaming=False)

Append frame to table (legacy API, creates table on first write).

Parameters:

• frame (polars.LazyFrame)

• table_ref (Any)

• params_instance (Any)

• streaming (bool)

Return type:

None

to_frame(records, /)

Convert lineage records into a Polars LazyFrame.

Parameters:

records (Sequence[PipelineRunRecord | ProcessRunRecord | StepRunRecord])

Return type:

polars.LazyFrame

loom.etl.backends.polars.decode_json_columns(frame, json_columns)

Decode JSON string columns the way PolarsSourceReader does at read time.

Each column named in json_columns holds JSON text and is replaced by the value decoded into its declared LoomType. Rows are first normalised to that type, so a field that is a string in some rows and an object in others decodes instead of raising polars.exceptions.ComputeError. A null or unparseable value decodes to null. Other columns pass through unchanged.

Use it to reproduce in a test what a step receives from a source declared with parse_json.

Parameters:

• frame (FrameT) – Eager or lazy frame holding the JSON text columns.

• json_columns (Iterable[JsonColumnSpec]) – Column declarations, e.g. FromTable(...).json_columns or FromFile(...).json_columns.

Returns:

A frame of the same kind as frame, or frame itself when json_columns is empty.

Return type:

FrameT

Example:

from loom.etl.backends.polars import decode_json_columns

events = FromTable("raw.events").parse_json("payload", Payload)
decoded = decode_json_columns(raw_frame, events.json_columns)

loom.etl.backends.polars.loom_type_to_polars(lt)

Recursively convert a LoomType to a concrete Polars DataType.

Returns an instance (not a class) so it can be used directly in pl.col(...).cast(...) and pl.lit(None).cast(...).

Parameters:

lt (LoomDtype | ListType | ArrayType | StructType | DecimalType | DatetimeType | DurationType | CategoricalType | EnumType) – Any LoomType value.

Returns:

Concrete Polars DataType instance.

Raises:

TypeError – If lt is a coarse structural LoomDtype (e.g. LoomDtype.LIST) whose inner type is unknown. Use ListType, StructType, etc. instead.

Return type:

polars.DataType

loom.etl.backends.polars.polars_to_loom_type(dtype)

Recursively convert a Polars DataType to its LoomType.

Handles arbitrarily nested types — List[Struct[...]], Struct[List[...]], Decimal(18,4), Datetime("ns","UTC"), etc.

For primitive types the result is the corresponding LoomDtype member (which is part of the LoomType union).

Parameters:

dtype (pl.DataType | DataTypeClass) – Polars DataType instance.

Returns:

Matching LoomType value.

Return type:

LoomType

loom.etl.backends.polars.read_delta_physical_schema(uri, storage_options=None)

Read physical schema directly from a Delta table URI.

Parameters:

• uri (str) – Delta table URI/path (including uc://catalog.schema.table).

• storage_options (dict[str, str] | None) – Optional delta-rs storage options.

Returns:

Physical schema when the table exists, else None.

Return type:

PolarsPhysicalSchema | None