warehouse

etl_craft.dialects.warehouse ¶

Warehouse dialects: one per database and table format.

postgres            native tables
duckdb              native tables in a local file
duckdb_iceberg      DuckDB compute over an Iceberg REST catalog
trino_iceberg       Trino over an Iceberg catalog
databricks          Delta tables
databricks_iceberg  Delta tables with UniForm, readable as Iceberg
snowflake           ordinary Snowflake tables
snowflake_iceberg   Snowflake Iceberg tables

A task's dialect comes from the warehouse connection and the task's table format: its own TABLE_FORMAT parameter, else Warehouse.Table_format. The vendors' SQLAlchemy dialects are optional extras; nothing here imports them.

AUDIT_COLUMN_TYPES `module-attribute` ¶

AUDIT_COLUMN_TYPES: dict[str, str] = {'TASK_RUN_ID': 'BIGINT', 'PIPELINE_ID': 'BIGINT', 'HASH_KEY': 'VARCHAR(32)', 'CREATE_DATE': 'TIMESTAMP WITH TIME ZONE', 'UPDATE_DATE': 'TIMESTAMP WITH TIME ZONE', 'CREATED_BY': 'VARCHAR(255)', 'UPDATED_BY': 'VARCHAR(255)', 'DELETE_FLAG': 'VARCHAR(1)', 'ACTIVE_FLAG': 'VARCHAR(1)'}

The DDL type of each engine-managed audit column, used to create a column with no rows.

Presented `dataclass` ¶

How one new connection presents its credential to the driver.

username and password go into the SQLAlchemy URL built inside the connection creator, never the engine's own logged URL. query joins that URL's query string, for drivers that read auth settings there (Trino). connect_args are passed to the driver after the dialect's own connect arguments, for settings that must never travel in a URL.

WarehouseDialect ¶

What differs between warehouses; the defaults are the ANSI behaviour PostgreSQL follows.

replace_strategy `class-attribute` `instance-attribute` ¶

replace_strategy: ReplaceStrategy = 'copy_and_restore'

How a replacement protects an existing target from a failed write.

identity_in_create `class-attribute` `instance-attribute` ¶

identity_in_create: bool = False

Whether ROW_ID must be declared when creating the table, rather than added later.

session_sql `class-attribute` `instance-attribute` ¶

session_sql: str | None = None

Setup SQL for each new warehouse connection.

overwrite_uses_ctas `class-attribute` `instance-attribute` ¶

overwrite_uses_ctas: bool = False

Whether an overwrite publishes a replacement snapshot with CTAS.

update_uses_merge `class-attribute` `instance-attribute` ¶

update_uses_merge: bool = False

Whether a joined update needs MERGE rather than UPDATE ... FROM.

identity_table_ddl ¶

identity_table_ddl(target: str, columns: str, params: Mapping[str, str]) -> str

Create an empty table with the supplied columns and a generated ROW_ID.

identity_replacement_ddl ¶

identity_replacement_ddl(conn: Connection, target: str, candidate: str, columns: str, params: Mapping[str, str], *, existing: bool) -> tuple[str, str]

Return candidate creation and atomic publication SQL for an identity table.

existing_table_format ¶

existing_table_format(conn: Connection, target: str) -> TableFormat

Read a table's format, or use the sole format supported by this dialect.

row_id_generated ¶

row_id_generated(conn: Connection, target: str) -> bool

Whether inserts into this existing table omit ROW_ID and let its default fill it.

present ¶

present(profile: ConnectionProfile, secret: str, url: WarehouseUrl) -> Presented

Return how one new connection authenticates by profile.auth_mode.

Called once per connection, so a credential minted here is current. The defaults: a user and password, a bearer token in the password position, and a private key through key_file_connect_args.

present_bearer ¶

present_bearer(token: str, user: str | None) -> Presented

Present a bearer token in the password position.

oauth_token ¶

oauth_token(profile: ConnectionProfile, secret: str, url: WarehouseUrl) -> str

Mint an access token by the profile's client-credentials grant.

on_connect ¶

on_connect(dbapi_connection: Any, profile: ConnectionProfile, secret: str) -> None

Apply this dialect's session settings, closing its setup cursor on failure too.

load_table_metadata ¶

load_table_metadata(conn: Connection, schema: str, table: str) -> None

Make information_schema.columns describe schema.table; it already does by default.

create_table_clause ¶

create_table_clause() -> str

Return the clause between CREATE TABLE <name> and AS <select>.

create_table_as ¶

create_table_as(conn: Connection, qualified_name: str, select_sql: str, params: Mapping[str, str]) -> None

Run CREATE TABLE ... AS SELECT in this dialect's table format.

params are the task's parameters, for dialects that read storage settings there.

replacement_comment ¶

replacement_comment(conn: Connection, target: str) -> str | None

Capture the comment before a transactional replacement removes the old object.

preserve_replacement_properties ¶

preserve_replacement_properties(conn: Connection, target: str, candidate: str) -> None

Refuse promotion unless the dialect can preserve the target's protected properties.

backup_table ¶

backup_table(conn: Connection, backup: str, target: str) -> None

Keep a durable row copy without reusing the target's configured storage path.

replacement_ddl ¶

replacement_ddl(conn: Connection, target: str, select_sql: str, params: Mapping[str, str], *, existing: bool) -> str

Render a single atomic replacement, or refuse when the dialect cannot do it.

overwrite_statement ¶

overwrite_statement(target: str, columns: str, select_sql: str) -> str

Replace rows without a separate truncate, where the warehouse supports it.

mirror_table_ddl ¶

mirror_table_ddl(name: str, column_ddl: str, cloning: CloningConfig) -> str | None

Return the DDL for a cloning mirror, or None for a plain CREATE TABLE.

task_storage_problem ¶

task_storage_problem(params: Mapping[str, str]) -> str | None

Return why a task's storage parameters cannot work here, or None.

unsupported_storage_problem ¶

unsupported_storage_problem(params: Mapping[str, str]) -> str | None

Return which storage parameters the task sets that do not apply here, or None.

Such a parameter would be ignored, leaving the table somewhere the author did not intend.

cloning_storage_problem ¶

cloning_storage_problem(cloning: CloningConfig) -> str | None

Return why cloning cannot create mirrors here as configured, or None.

alter_table_keyword ¶

alter_table_keyword() -> str

Return the keyword that alters a table this dialect created.

column_metadata_query ¶

column_metadata_query(conn: Connection, table: str, columns: str) -> tuple[str, dict[str, str | None]]

Read information-schema columns for a bare, schema- or catalog-qualified table.

full_column_types ¶

full_column_types(conn: Connection, table: str) -> dict[str, str]

Read reusable DDL types, including nested types and declared sizes.

full_column_type ¶

full_column_type(conn: Connection, table: str, column: str) -> str

Return a column's complete warehouse type, refusing missing metadata.

column_addition_problem ¶

column_addition_problem(data_type: str) -> str | None

Return why this type cannot be added without rebuilding, or None.

same_column_type ¶

same_column_type(source: str, target: str) -> bool

Compare types as this warehouse stores and reports them.

evolution_column_type ¶

evolution_column_type(data_type: str) -> str

Return the full stage type in the target table format's DDL representation.

is_bigint_type ¶

is_bigint_type(data_type: str) -> bool

Whether a stored column has the signed BIGINT type used for task-run ids.

audit_column_type ¶

audit_column_type(column: str) -> str

Return the DDL type of an engine-managed audit column.

scratch_table_keyword ¶

scratch_table_keyword() -> str

Return TEMPORARY TABLE, or TABLE where temporary tables are unusable.

hash_expression ¶

hash_expression(values: list[str], types: list[str] | None = None) -> str

Return the version-2 MD5 of typed, NULL-tagged, length-prefixed values.

canonical_text ¶

canonical_text(value: str, kind: str) -> str

Render supported scalar values independently of session output formats.

decimal_text ¶

decimal_text(value: str, kind: str) -> str

Keep the decimal's declared scale, including trailing zeroes.

date_text ¶

date_text(value: str) -> str

Render dates in ISO format.

timestamp_text ¶

timestamp_text(value: str, kind: str) -> str

Render a UTC timestamp with exactly six fractional digits.

update_from_stage ¶

update_from_stage(target: str, stage: str, keys: tuple[str, ...], assignments: Mapping[str, str], condition: str) -> str

Update matching target rows from a deduplicated stage using aliases t and s.

prepare_update_stage ¶

prepare_update_stage(stage: str, keys: tuple[str, ...]) -> tuple[str, ...]

Statements that prepare a stage for joined updates on its merge keys.

all_dialects `cached` ¶

all_dialects() -> tuple[WarehouseDialect, ...]

Return one instance of every warehouse dialect.

for_key ¶

for_key(key: str) -> WarehouseDialect

Return the dialect registered under key, such as trino_iceberg.

resolve ¶

resolve(sqlalchemy_name: str, table_format: str) -> WarehouseDialect

Choose the dialect for a connection's SQLAlchemy name and a table format.

Trino is Iceberg whichever format is asked for, because its catalog decides; PostgreSQL has no Iceberg tables; an unsupported database is a ConfigurationError.