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)

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)

read(spec, params_instance, /)

Read source spec and return lazy frame.

Parameters:

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

• 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) – 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.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