tables

etl_craft.handlers.sql.tables ¶

Target tables: creating their shape, their ROW_ID key, and keeping them in step with the SELECT.

Every table the engine creates carries, after the SELECT's own columns, PIPELINE_RUN_ID, the audit columns of its action (AUDIT_COLUMNS) and ROW_ID, a generated key that business rules reference. Only CREATE_TABLE and SETUP_TABLE create tables; every other action needs its target to exist already, created by a SETUP_TABLE task or by hand.

Before OVERWRITE_TABLE and the merges write a row, the target is checked against the SELECT:

a target missing one of the action's engine-managed columns fails, naming them;
a SELECT missing a column the target has fails: columns are never dropped;
a SELECT with a new column fails unless SCHEMA_EVOLUTION is true, when the target is extended with nullable columns at the end, preserving existing rows and properties.

Where the warehouse has transactional DDL (PostgreSQL, DuckDB) an action's statements commit or roll back together. Trino and the cloud warehouses commit each statement, so an action that fails part-way can leave its work half done; every action re-derives its effect from the target's current state, so running the task again converges.

AUDIT_COLUMNS `module-attribute` ¶

AUDIT_COLUMNS: dict[str, tuple[str, ...]] = {SqlAction.CREATE_TABLE: ('PIPELINE_ID', 'TASK_RUN_ID'), SqlAction.OVERWRITE_TABLE: ('PIPELINE_ID', 'TASK_RUN_ID', 'UPDATE_DATE'), SqlAction.APPEND_TABLE: ('PIPELINE_ID', 'TASK_RUN_ID', 'CREATE_DATE'), SqlAction.SCD1_MERGE: ('PIPELINE_ID', 'TASK_RUN_ID', 'HASH_KEY', 'CREATE_DATE', 'CREATED_BY', 'UPDATE_DATE', 'UPDATED_BY', 'DELETE_FLAG'), SqlAction.SCD2_MERGE: ('PIPELINE_ID', 'TASK_RUN_ID', 'HASH_KEY', 'CREATE_DATE', 'CREATED_BY', 'UPDATE_DATE', 'UPDATED_BY', 'DELETE_FLAG', 'ACTIVE_FLAG')}

The audit columns each action adds after PIPELINE_RUN_ID, in order.

ENGINE_COLUMNS `module-attribute` ¶

ENGINE_COLUMNS = frozenset(name.lower() for name in ('PIPELINE_RUN_ID', ROW_ID_COLUMN, *(column for columns in AUDIT_COLUMNS.values() for column in columns)))

The columns etl-craft writes itself, lower case; a SELECT may not return them.

build_stage ¶

build_stage(session: Session, select_sql: str, *, empty: bool = False) -> str

Materialize the SELECT once into this task run's stage table; return its name.

Every later statement reads the stage, so the SELECT runs exactly once. empty keeps only its shape. The stage's columns are checked before anything reads it (see check_stage_columns).

check_stage_columns ¶

check_stage_columns(session: Session, stage: str) -> None

Refuse a SELECT whose columns the engine cannot write; SqlGuardError naming them.

A column etl-craft writes itself (ENGINE_COLUMNS), as a SELECT * over a table the engine wrote returns, would clash with the engine's own value. A column whose name needs quoting could not be named in the statements the engine writes: one that is not a plain identifier, or, where the warehouse folds unquoted names (identifier_case), one that is not in that case, such as "CustomerId" on PostgreSQL.

create_target_shape ¶

create_target_shape(session: Session, stage: str, audit_columns: tuple[str, ...]) -> None

Create the target, empty: the stage's columns, PIPELINE_RUN_ID, audit_columns.

sequence_name ¶

sequence_name(session: Session) -> str

Return the target's own ROW_ID sequence (DuckDB), in the target's schema.

add_row_id ¶

add_row_id(session: Session) -> None

Give a new target its ROW_ID key: identity, sequence default, or computed (Iceberg).

require_target ¶

require_target(session: Session) -> list[tuple[str, str]]

Return the target's columns; SqlGuardError naming the remedy when it does not exist.

check_target_audit ¶

check_target_audit(session: Session, action: str) -> list[tuple[str, str]]

Refuse an absent target or missing action-managed columns before any write.

check_or_evolve ¶

check_or_evolve(session: Session, stage: str, action: str, *, schema_evolution: bool) -> None

Check the existing target against the stage, adding new columns when allowed.

evolve ¶

evolve(session: Session, columns: list[tuple[str, str]]) -> None

Append nullable columns without rebuilding rows, keys, properties or dependencies.

refuse_null_keys ¶

refuse_null_keys(session: Session, stage: str, merge_key: tuple[str, ...]) -> None

Fail when a staged row has a NULL in a merge key column, before the target is touched.

A NULL key matches no target row, so a merge would insert it again on every run and a delete would never find it.

dedupe ¶

dedupe(session: Session, stage: str, merge_key: tuple[str, ...], order: str | None) -> str

Return a stage with one row per merge key: this one, or a copy keeping order's first.

Without MERGE_DEDUPE_ORDER, duplicate keys fail before the target is touched: which row should win is the author's call, never the engine's. For the same reason an order that ties between rows that differ fails too; rows that are identical may tie.

add_hash_key ¶

add_hash_key(session: Session, stage: str, compare_columns: tuple[str, ...]) -> None

Add HASH_KEY to the stage, over MERGE_COMPARE_COLUMNS.

Added after the schema check, which compares the SELECT's own columns only.

check_identity_types ¶

check_identity_types(session: Session, columns: tuple[str, ...] = IDENTITY_COLUMNS) -> None

Refuse identity columns that cannot hold the engine's integer ids.