loom.etl.maintenance¶

Delta table maintenance — backend-agnostic vacuum, compact, and Z-Order.

Public API¶

Authoring:

from loom.etl.maintenance import MaintenanceStep, MaintainTable, MaintainSchema

Running:

from loom.etl.maintenance import MaintenanceRunner, MaintenanceReport

Result types:

from loom.etl.maintenance import (
    VacuumResult,
    OptimizeResult,
    TableMaintenanceResult,
    MaintenanceError,
)

Backend protocol (for custom implementations):

from loom.etl.maintenance import DeltaTableMaintainer

Operation specs (for custom operations and backends):

from loom.etl.maintenance import (
    MaintenanceSpec,
    OpSpec,
    OperationDeclaration,
    VacuumSpec,
    CompactSpec,
    ZOrderSpec,
)

Example:

class NightlyMaintenance(MaintenanceStep[None]):
    operations = [
        MaintainSchema("raw").vacuum(retention_hours=168, dry_run=False).compact(),
        MaintainTable("staging.orders")
            .vacuum(dry_run=False)
            .z_order_by(["run_date", "_id"]),
    ]

runner = MaintenanceRunner.from_config(config)
report = runner.run(NightlyMaintenance)
report.raise_if_errors()

Custom operations¶

An operation is any class implementing OpSpec: a name class variable, which keys its result in TableMaintenanceResult.op_results, and an execute(maintainer, uri, location) method returning a VacuumResult or an OptimizeResult (or a subclass). To run it, wrap a declaration in an OperationDeclaration whose resolve appends the op to each MaintenanceSpec:

from dataclasses import dataclass, replace
from typing import ClassVar

from deltalake import DeltaTable

@dataclass(frozen=True)
class CheckpointSpec:
    name: ClassVar[str] = "checkpoint"

    def execute(self, maintainer, uri, location) -> OptimizeResult:
        DeltaTable(uri, storage_options=location.storage_options).create_checkpoint()
        return OptimizeResult()

class Checkpointed:
    def __init__(self, inner: OperationDeclaration) -> None:
        self._inner = inner

    def resolve(self, config: StorageConfig) -> list[MaintenanceSpec]:
        return [
            replace(spec, ops=(*spec.ops, CheckpointSpec()))
            for spec in self._inner.resolve(config)
        ]

class NightlyMaintenance(MaintenanceStep[None]):
    operations = [Checkpointed(MaintainSchema("raw").vacuum(dry_run=False))]

class loom.etl.maintenance.MaintenanceStep[source]¶

Bases: Generic[ParamsT]

Base class for Delta table maintenance steps.

Subclass and declare operations as a list of MaintainTable and/or MaintainSchema instances.

Example:

class NightlyMaintenance(MaintenanceStep[None]):
    operations = [
        MaintainSchema("raw").vacuum(retention_hours=168, dry_run=False).compact(),
        MaintainTable("staging.orders")
            .vacuum(dry_run=False)
            .z_order_by(["run_date", "_id"]),
    ]

The step is executed via MaintenanceRunner:

runner = MaintenanceRunner.from_config(config)
report = runner.run(NightlyMaintenance)
report.raise_if_errors()

operations_for(params)[source]¶

Return the list of operations to execute for this run.

Override to build the list dynamically from params — useful when operations are driven by Prefect deployment parameters instead of being hardcoded at class-definition time.

The default implementation returns the class-level operations list (static behaviour, backwards-compatible).

Parameters:: params (ParamsT)
Return type:: list[OperationDeclaration]

class loom.etl.maintenance.MaintainTable(table_ref)[source]¶

Bases: object

Declare maintenance operations for one explicitly named Delta table.

Example:

MaintainTable("raw.events")
    .vacuum(retention_hours=168, dry_run=False)
    .compact()

Parameters:: table_ref (str) – Logical table reference (e.g. "raw.events").

vacuum(retention_hours=None, *, dry_run=True, full=False)[source]¶

Schedule a VACUUM step.

Parameters:

retention_hours (int | None) – Files older than this are eligible. None uses the table’s delta.deletedFileRetentionDuration (default 168 h / 7 days).
dry_run (bool) – When True (default), lists files without deleting. Set dry_run=False explicitly for production runs.
full (bool) – When True, also deletes every file under the table directory that the Delta log does not reference and that is older than the retention, whoever wrote it. See VacuumSpec.

Return type:

MaintainTable

compact(target_size=None)[source]¶

Schedule a bin-packing compaction step.

Mutually exclusive with z_order_by().

Parameters:: target_size (int | None)
Return type:: MaintainTable

z_order_by(columns, target_size=None)[source]¶

Schedule a Z-Order clustering step.

Mutually exclusive with compact().

Parameters:

columns (list[str]) – Columns to Z-Order by. Must be non-empty.
target_size (int | None)

Return type:

MaintainTable

resolve(config)[source]¶

Implement OperationDeclaration.

Parameters:: config (StorageConfig)
Return type:: list[MaintenanceSpec]

class loom.etl.maintenance.MaintainSchema(schema_prefix)[source]¶

Bases: object

Declare maintenance operations for all tables under a schema prefix.

Discovery is config-driven: _expand() filters StorageConfig.tables for routes whose name starts with {schema_prefix}.. No filesystem scanning is performed.

Example:

MaintainSchema("raw").vacuum(retention_hours=168, dry_run=False).compact()

Parameters:: schema_prefix (str) – Schema prefix to match (e.g. "raw" matches "raw.events", "raw.orders").

vacuum(retention_hours=None, *, dry_run=True, full=False)[source]¶

Schedule a VACUUM step for all discovered tables.

Takes the same arguments as MaintainTable.vacuum().

Parameters:

retention_hours (int | None)
dry_run (bool)
full (bool)

Return type:

MaintainSchema

compact(target_size=None)[source]¶

Schedule a compaction step for all discovered tables.

Mutually exclusive with z_order_by().

Parameters:: target_size (int | None)
Return type:: MaintainSchema

z_order_by(columns, target_size=None)[source]¶

Schedule a Z-Order step for all discovered tables.

Mutually exclusive with compact().

Parameters:

columns (list[str])
target_size (int | None)

Return type:

MaintainSchema

resolve(config)[source]¶

Implement OperationDeclaration.

Parameters:: config (StorageConfig)
Return type:: list[MaintenanceSpec]

class loom.etl.maintenance.MaintenanceRunner(maintainer, locator, config, *, missing_table_policy='skip')[source]¶

Bases: object

Execute a MaintenanceStep against Delta tables.

Construct via from_config() for the standard delta-rs path:

runner = MaintenanceRunner.from_config(config)
report = runner.run(NightlyMaintenance)
report.raise_if_errors()

Parameters:

maintainer (DeltaTableMaintainer) – Backend implementation (default: DeltaRsMaintainer).
locator (TableLocator) – Resolves logical table refs to physical TableLocation.
config (StorageConfig) – Full storage config (needed by MaintainSchema expansion and run_from_config()).
missing_table_policy (Literal['skip', 'error']) – "skip" silently omits tables not present in config (or not yet initialized as Delta tables); "error" raises immediately.

classmethod from_config(config, *, missing_table_policy='skip')[source]¶

Build a MaintenanceRunner from a StorageConfig.

Uses delta-rs (deltalake package) as the only backend. Calls to_path_locator() which reuses the same PrefixLocator / MappingLocator logic as the Polars ETL backend — no private helpers crossed.

Raises:

ValueError – If config has no path routes configured (only catalog/UC refs, which delta-rs cannot resolve directly).

Parameters:

config (StorageConfig)
missing_table_policy (Literal['skip', 'error'])

Return type:

MaintenanceRunner

run(step_cls, params=None)[source]¶

Execute all operations declared on step_cls.

MaintainSchema entries are expanded against the config at run time; MaintainTable entries are compiled directly.

Parameters:

step_cls (type[MaintenanceStep[Any]]) – A MaintenanceStep subclass.
params (Any) – Reserved for future parameterisation (currently unused).

Returns:

MaintenanceReport with per-table results.

Return type:

MaintenanceReport

run_from_config()[source]¶

Execute maintenance using the maintenance block in StorageConfig.

Discovers all tables listed under the configured schema prefixes and applies the vacuum / compact / z-order settings from StorageConfig.maintenance.

Returns an empty MaintenanceReport when no maintenance config is defined or no matching tables are found.

Return type:: MaintenanceReport

class loom.etl.maintenance.MaintenanceReport(results=<factory>)[source]¶

Bases: object

Aggregate result of a maintenance run.

Parameters:: results (list[TableMaintenanceResult]) – Per-table outcomes (one entry per table processed).

property has_errors: bool¶: True when at least one table reported an error.

raise_if_errors()[source]¶

Raise MaintenanceError if any table failed.

Useful as a Prefect flow gate:

report = runner.run(NightlyMaintenance)
report.raise_if_errors()

Return type:: None

exception loom.etl.maintenance.MaintenanceError(failed)[source]¶

Bases: Exception

Raised by MaintenanceReport.raise_if_errors() when any table failed.

Parameters:: failed (list[TableMaintenanceResult])
Return type:: None

class loom.etl.maintenance.VacuumResult(files_deleted=<factory>)[source]¶

Bases: object

Result of a VACUUM operation.

Parameters:: files_deleted (list[str]) – Paths of the files that were deleted (or that would be deleted when dry_run=True).

class loom.etl.maintenance.OptimizeResult(num_files_added=0, num_files_removed=0)[source]¶

Bases: object

Result of a compaction or Z-Order operation.

Uses counts (not lists) — consistent with the numFilesAdded / numFilesRemoved keys returned by the deltalake metrics dict.

Parameters:

num_files_added (int)
num_files_removed (int)

class loom.etl.maintenance.TableMaintenanceResult(table_ref, op_results=<factory>, error=None, duration_seconds=0.0)[source]¶

Bases: object

Per-table maintenance outcome.

Parameters:

table_ref (str) – Logical table reference (e.g. "raw.events").
op_results (dict[str, VacuumResult | OptimizeResult]) – Keyed results for each op that ran, by op.name.
error (Exception | None) – Exception caught during execution, or None on success.
duration_seconds (float) – Wall-clock time for all ops on this table.

property ok: bool¶: True when no error occurred.

class loom.etl.maintenance.DeltaTableMaintainer(*args, **kwargs)[source]¶

Bases: Protocol

Protocol for executing maintenance operations on a single Delta table.

Implementations receive the full TableLocation so they can propagate storage_options, writer_properties, and commit_properties to every delta-rs call — matching the behaviour of loom.etl.backends.polars._writer.PolarsTargetWriter._write_kwargs().

vacuum(uri, spec, location)[source]¶

Run VACUUM on the table at uri.

Parameters:

uri (str)
spec (VacuumSpec)
location (TableLocation)

Return type:

VacuumResult

compact(uri, spec, location)[source]¶

Run bin-packing compaction on the table at uri.

Parameters:

uri (str)
spec (CompactSpec)
location (TableLocation)

Return type:

OptimizeResult

z_order(uri, spec, location)[source]¶

Run Z-Order clustering on the table at uri.

Parameters:

uri (str)
spec (ZOrderSpec)
location (TableLocation)

Return type:

OptimizeResult

class loom.etl.maintenance.OperationDeclaration(*args, **kwargs)[source]¶

Bases: Protocol

Unified interface for declarative operation builders.

Both MaintainTable and MaintainSchema implement this Protocol so the runner can iterate them uniformly without isinstance branching.

class loom.etl.maintenance.OpSpec(*args, **kwargs)[source]¶

Bases: Protocol

Protocol implemented by VacuumSpec, CompactSpec, ZOrderSpec — and any future op.

Each spec knows how to dispatch itself against a DeltaTableMaintainer. Adding a new operation type requires only a new spec class; the runner does not change.

class loom.etl.maintenance.MaintenanceSpec(table_ref, ops=())[source]¶

Bases: object

The ops to run, in order, against one table.

resolve() returns one per table; the runner resolves table_ref to a location and runs each op.

Parameters:

table_ref (str) – Logical table reference (e.g. "raw.events").
ops (tuple[OpSpec, ...]) – Operations to execute against the table, in order.

class loom.etl.maintenance.VacuumSpec(retention_hours=None, dry_run=True, full=False)[source]¶

Bases: object

Parameters for a VACUUM operation.

Parameters:

retention_hours (int | None) – Files older than this are eligible for deletion. None uses the table’s delta.deletedFileRetentionDuration property (default 168 hours / 7 days).
dry_run (bool) – When True (the default), lists eligible files without deleting them. Set dry_run=False explicitly in production.
full (bool) – When True, also deletes every file under the table directory that the Delta log does not reference and that is older than the retention, whoever wrote it; only names starting with an underscore are kept. False (the default) deletes only the files the log records as removed.

class loom.etl.maintenance.CompactSpec(target_size=None)[source]¶

Bases: object

Parameters for a file compaction (bin-packing) operation.

Parameters:: target_size (int | None) – Target output file size in bytes. None lets delta-rs choose based on the table’s configuration.

class loom.etl.maintenance.ZOrderSpec(columns=<factory>, target_size=None)[source]¶

Bases: object

Parameters for a Z-Order clustering operation.

Parameters:

columns (list[str]) – Columns to Z-Order by. Must be non-empty.
target_size (int | None) – Target output file size in bytes. None uses delta-rs default.