Skip to content

databricks_statement_generator

Render typed diffs into Databricks (Unity Catalog) SQL statements.

Three-level naming: catalog.schema.table. One diff may produce zero, one, or several ALTER statements depending on the change kind and the number of delta fields.

AzureDatabricksOptions

Bases: BaseModel

Azure Databricks-specific render-time options grouped by object type.

These options are not tracked in state and are not crawled from the database. They influence rendering only — currently only the CREATE TABLE statement (via tables.properties). As a result, changes to these options apply to newly created objects only; existing objects are left untouched. Future object groups (views etc.) will be added as sibling sub-blocks.

Parameters:

Name Type Description Default
tables

Options applied to table-creation statements.

 required
Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class AzureDatabricksOptions(BaseModel):
    """Azure Databricks-specific render-time options grouped by object type.

    These options are *not* tracked in state and are not crawled from the
    database. They influence rendering only — currently only the
    ``CREATE TABLE`` statement (via ``tables.properties``). As a result,
    changes to these options apply to newly created objects only; existing
    objects are left untouched. Future object groups (``views`` etc.) will
    be added as sibling sub-blocks.

    Args:
        tables: Options applied to table-creation statements.

    """

    model_config = ConfigDict(extra="forbid")

    tables: TableOptions = Field(default_factory=TableOptions)

DatabricksStatementGenerator

Render typed diffs to Databricks SQL via in-code match-dispatch.

Unsupported deltas are detected generically: a per-class registry of handled fields runs as a pre-check before the match dispatch. Any *Changed diff that touches a field outside its allow-list is reported as unsupported — no silent no-ops, even for diff types added in the future. To add support for a new field on an existing *Changed class, register it in :attr:_HANDLED_CHANGED_FIELDS and extend the renderer.

Parameters:

Name Type Description Default
on_unsupported OnUnsupported

"fail" (default) raises :class:UnsupportedDiffError for diffs without a SQL representation. "skip" returns an empty command list for those diffs; pipeline-level cascading of the skip to dependent diffs is handled by the caller.

 'fail'
Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class DatabricksStatementGenerator:
    """Render typed diffs to Databricks SQL via in-code match-dispatch.

    Unsupported deltas are detected *generically*: a per-class registry of
    handled fields runs as a pre-check before the match dispatch. Any
    ``*Changed`` diff that touches a field outside its allow-list is reported
    as unsupported — no silent no-ops, even for diff types added in the
    future. To add support for a new field on an existing ``*Changed`` class,
    register it in :attr:`_HANDLED_CHANGED_FIELDS` and extend the renderer.

    Args:
        on_unsupported: ``"fail"`` (default) raises :class:`UnsupportedDiffError`
            for diffs without a SQL representation. ``"skip"`` returns an empty
            command list for those diffs; pipeline-level cascading of the skip
            to dependent diffs is handled by the caller.

    """

    # Per-class allow-list of column/table/schema fields whose changes this
    # generator can express in SQL. Anything outside this set surfaces as
    # `UnsupportedDiffError`. Adding a new ``*Changed`` diff class with no
    # entry here means it is always unsupported until rendering is added.
    _HANDLED_CHANGED_FIELDS: ClassVar[dict[type[Diff], frozenset[str]]] = {
        ColumnChanged: frozenset(
            {
                "data_type",
                "data_type_length",
                "data_type_precision",
                "data_type_numeric_scale",
                "is_nullable",
                "comment",
                "constraints",
            }
        ),
        # `is_key` is intentionally absent: primary keys are table-level and
        # rendered from PrimaryKeyChanged, not from a per-column ColumnChanged.
        # TableChanged and SchemaChanged: no fields handled today — any change
        # surfaces as unsupported. Add fields here as renderers are added.
    }

    def __init__(
        self,
        *,
        on_unsupported: OnUnsupported = "fail",
        options: AzureDatabricksOptions | None = None,
    ) -> None:
        """Initialize the generator with an unsupported-diff handling mode and options."""
        self._on_unsupported = on_unsupported
        self._options = options or AzureDatabricksOptions()

    def generate(self, diff: Diff) -> list[str]:  # noqa: PLR0911
        """Return the SQL commands for a single diff (zero, one, or many)."""
        catalog_reason = (
            "catalogs are assumed to exist and are not managed by dtm (no CREATE/ALTER/DROP CATALOG is emitted)"
        )

        # Generic deny-by-default guard for any *Changed-style diff (anything
        # carrying `old`/`new` Pydantic models). Field deltas outside the
        # per-class allow-list surface as unsupported regardless of whether
        # the diff type is known to this generator.
        unsupported = self._check_changed_fields(diff)
        if unsupported is not None:
            return unsupported

        match diff:
            case DatabaseAdded() | DatabaseRemoved() | DatabaseChanged():
                return self._handle_unsupported(diff, catalog_reason)

            case SchemaAdded(schema_obj=sch, parent_db=db):
                return [f"CREATE SCHEMA IF NOT EXISTS {db.name}.{sch.name};"]
            case SchemaRemoved(schema_obj=sch, parent_db=db):
                return [f"DROP SCHEMA IF EXISTS {db.name}.{sch.name} CASCADE;"]

            case TableAdded(table=tbl, parent_schema=sch, parent_db=db):
                return [
                    _render_create_table(
                        db.name,
                        sch.name,
                        tbl.name,
                        tbl.columns,
                        table_properties=self._options.tables.properties,
                    )
                ]
            case TableRemoved(table=tbl, parent_schema=sch, parent_db=db):
                return [f"DROP TABLE IF EXISTS {db.name}.{sch.name}.{tbl.name};"]

            case ColumnAdded(column=col, parent_table=tbl, parent_schema=sch, parent_db=db):
                return _render_add_column(db.name, sch.name, tbl.name, col)
            case ColumnRemoved(column=col, parent_table=tbl, parent_schema=sch, parent_db=db):
                return [f"ALTER TABLE {db.name}.{sch.name}.{tbl.name} DROP COLUMN IF EXISTS {col.name};"]
            case ColumnChanged(old=old, new=new, parent_table=tbl, parent_schema=sch, parent_db=db):
                return _render_alter_column(db.name, sch.name, tbl.name, old, new)

            case PrimaryKeyChanged(
                old_key_columns=old_pk, new_key_columns=new_pk, parent_table=tbl, parent_schema=sch, parent_db=db
            ):
                return _render_primary_key_change(db.name, sch.name, tbl.name, old_pk, new_pk)

            case _:
                return self._handle_unsupported(diff, f"no handler for {type(diff).__name__}")

    def _check_changed_fields(self, diff: Diff) -> list[str] | None:
        """Return the unsupported result if `diff` carries unhandled field deltas.

        Operates on any diff with ``old``/``new`` attributes (the *Changed
        family). Subclasses with no entry in :attr:`_HANDLED_CHANGED_FIELDS`
        are treated as having no handled fields — any delta is unsupported.
        Returns ``None`` if all field deltas are within the allow-list (or
        the diff is not a ``*Changed`` diff).
        """
        old = getattr(diff, "old", None)
        new = getattr(diff, "new", None)
        if not isinstance(old, BaseModel) or not isinstance(new, BaseModel):
            return None
        handled = self._HANDLED_CHANGED_FIELDS.get(type(diff), frozenset())
        unhandled = sorted(_differing_fields(old, new) - handled)
        if not unhandled:
            return None
        return self._handle_unsupported(
            diff,
            f"{type(diff).__name__} touches fields with no SQL renderer: {unhandled}",
        )

    def _handle_unsupported(self, diff: Diff, reason: str) -> list[str]:
        if self._on_unsupported == "fail":
            raise UnsupportedDiffError(diff, reason)
        return []

__init__(*, on_unsupported='fail', options=None)

Initialize the generator with an unsupported-diff handling mode and options.

Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
121
122
123
124
125
126
127
128
129
def __init__(
    self,
    *,
    on_unsupported: OnUnsupported = "fail",
    options: AzureDatabricksOptions | None = None,
) -> None:
    """Initialize the generator with an unsupported-diff handling mode and options."""
    self._on_unsupported = on_unsupported
    self._options = options or AzureDatabricksOptions()

generate(diff)

Return the SQL commands for a single diff (zero, one, or many).

Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def generate(self, diff: Diff) -> list[str]:  # noqa: PLR0911
    """Return the SQL commands for a single diff (zero, one, or many)."""
    catalog_reason = (
        "catalogs are assumed to exist and are not managed by dtm (no CREATE/ALTER/DROP CATALOG is emitted)"
    )

    # Generic deny-by-default guard for any *Changed-style diff (anything
    # carrying `old`/`new` Pydantic models). Field deltas outside the
    # per-class allow-list surface as unsupported regardless of whether
    # the diff type is known to this generator.
    unsupported = self._check_changed_fields(diff)
    if unsupported is not None:
        return unsupported

    match diff:
        case DatabaseAdded() | DatabaseRemoved() | DatabaseChanged():
            return self._handle_unsupported(diff, catalog_reason)

        case SchemaAdded(schema_obj=sch, parent_db=db):
            return [f"CREATE SCHEMA IF NOT EXISTS {db.name}.{sch.name};"]
        case SchemaRemoved(schema_obj=sch, parent_db=db):
            return [f"DROP SCHEMA IF EXISTS {db.name}.{sch.name} CASCADE;"]

        case TableAdded(table=tbl, parent_schema=sch, parent_db=db):
            return [
                _render_create_table(
                    db.name,
                    sch.name,
                    tbl.name,
                    tbl.columns,
                    table_properties=self._options.tables.properties,
                )
            ]
        case TableRemoved(table=tbl, parent_schema=sch, parent_db=db):
            return [f"DROP TABLE IF EXISTS {db.name}.{sch.name}.{tbl.name};"]

        case ColumnAdded(column=col, parent_table=tbl, parent_schema=sch, parent_db=db):
            return _render_add_column(db.name, sch.name, tbl.name, col)
        case ColumnRemoved(column=col, parent_table=tbl, parent_schema=sch, parent_db=db):
            return [f"ALTER TABLE {db.name}.{sch.name}.{tbl.name} DROP COLUMN IF EXISTS {col.name};"]
        case ColumnChanged(old=old, new=new, parent_table=tbl, parent_schema=sch, parent_db=db):
            return _render_alter_column(db.name, sch.name, tbl.name, old, new)

        case PrimaryKeyChanged(
            old_key_columns=old_pk, new_key_columns=new_pk, parent_table=tbl, parent_schema=sch, parent_db=db
        ):
            return _render_primary_key_change(db.name, sch.name, tbl.name, old_pk, new_pk)

        case _:
            return self._handle_unsupported(diff, f"no handler for {type(diff).__name__}")

TableOptions

Bases: BaseModel

Render-time options that apply to CREATE TABLE statements.

Parameters:

Name Type Description Default
properties

Key-value pairs emitted as TBLPROPERTIES on every generated CREATE TABLE statement. Values are quoted as strings.

 required
Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
47
48
49
50
51
52
53
54
55
56
57
58
class TableOptions(BaseModel):
    """Render-time options that apply to ``CREATE TABLE`` statements.

    Args:
        properties: Key-value pairs emitted as ``TBLPROPERTIES`` on every
            generated ``CREATE TABLE`` statement. Values are quoted as strings.

    """

    model_config = ConfigDict(extra="forbid")

    properties: dict[str, str] = Field(default_factory=dict)

UnsupportedDiffError

Bases: Exception

Raised when a diff has no Databricks SQL representation.

Carries the offending diff so the caller can decide how to surface it.

Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
31
32
33
34
35
36
37
38
39
40
41
class UnsupportedDiffError(Exception):
    """Raised when a diff has no Databricks SQL representation.

    Carries the offending diff so the caller can decide how to surface it.
    """

    def __init__(self, diff: Diff, reason: str) -> None:
        """Build the error with the offending diff and a human-readable reason."""
        self.diff = diff
        self.reason = reason
        super().__init__(f"Unsupported diff {type(diff).__name__}: {reason}")

__init__(diff, reason)

Build the error with the offending diff and a human-readable reason.

Source code in src/cloe_delta_table_manager/diff/databricks_statement_generator.py 
37
38
39
40
41
def __init__(self, diff: Diff, reason: str) -> None:
    """Build the error with the offending diff and a human-readable reason."""
    self.diff = diff
    self.reason = reason
    super().__init__(f"Unsupported diff {type(diff).__name__}: {reason}")