Skip to content

diff_resolver

Domain-specific walker that produces typed diffs from two DatabaseState trees.

Compares two state trees in parallel by name at each level. Emits per-object typed diffs and cascades full add/remove down to leaf objects. Renames are not detected: a renamed object surfaces as *Removed followed by *Added.

DiffResolver

Walk two DatabaseState trees and emit typed diffs.

Emission order is top-down: parent diffs appear before their children in the returned list. Execution ordering for downstream stages is handled by :class:DiffDependencyResolver via depends_on.

Source code in src/cloe_delta_table_manager/diff/diff_resolver.py
class DiffResolver:
    """Walk two DatabaseState trees and emit typed diffs.

    Emission order is top-down: parent diffs appear before their children in
    the returned list. Execution ordering for downstream stages is handled by
    :class:`DiffDependencyResolver` via ``depends_on``.
    """

    def diff(self, current: DatabaseState, desired: DatabaseState) -> list[Diff]:
        """Return the typed diff list to transform ``current`` into ``desired``."""
        diffs: list[Diff] = []
        self._diff_databases(current, desired, diffs)
        return diffs

    def _diff_databases(self, current: Databases, desired: Databases, out: list[Diff]) -> None:
        """Descend into every database present in either state.

        Catalogs (databases) are assumed to exist outside dtm's control — see
        ``docs/decisions/migration-generation.md``. The walker therefore never
        emits ``DatabaseAdded`` / ``DatabaseRemoved`` / ``DatabaseChanged``;
        it only walks into the schemas/tables/columns inside each catalog.
        """
        current_by_name = {db.name: db for db in current.databases}
        desired_by_name = {db.name: db for db in desired.databases}

        for name in sorted(current_by_name.keys() | desired_by_name.keys()):
            current_db = current_by_name.get(name)
            desired_db = desired_by_name.get(name)
            # Use whichever Database is present as the parent reference for
            # children — preferring `desired` since it represents the goal
            # state used in downstream rendering.
            parent_db = desired_db or current_db
            assert parent_db is not None  # name came from the union of the two maps
            current_schemas = current_db.schemas if current_db else []
            desired_schemas = desired_db.schemas if desired_db else []
            self._diff_schemas(parent_db, current_schemas, desired_schemas, out)

    def _diff_schemas(self, parent_db: Database, current: list[Schema], desired: list[Schema], out: list[Diff]) -> None:
        current_by_name = {s.name: s for s in current}
        desired_by_name = {s.name: s for s in desired}

        for name in sorted(desired_by_name.keys() - current_by_name.keys()):
            self._cascade_schema_added(desired_by_name[name], parent_db, out)
        for name in sorted(current_by_name.keys() - desired_by_name.keys()):
            self._cascade_schema_removed(current_by_name[name], parent_db, out)
        for name in sorted(current_by_name.keys() & desired_by_name.keys()):
            old, new = current_by_name[name], desired_by_name[name]
            if _differ(old, new, _SCHEMA_EXCLUDE):
                out.append(SchemaChanged(old=old, new=new, parent_db=parent_db))
            self._diff_schema_contents(parent_db, new, old, new, out)

    def _diff_schema_contents(
        self,
        parent_db: Database,
        parent_schema: Schema,
        current_schema: Schema,
        desired_schema: Schema,
        out: list[Diff],
    ) -> None:
        """Dispatch over peer object types within a schema.

        Today only tables exist at this level; views, materialized views and
        sequences extend this dispatch when ``cloe_metadata`` supports them.
        """
        self._diff_tables(parent_db, parent_schema, current_schema.tables, desired_schema.tables, out)

    def _diff_tables(
        self,
        parent_db: Database,
        parent_schema: Schema,
        current: list[Table],
        desired: list[Table],
        out: list[Diff],
    ) -> None:
        current_by_name = {t.name: t for t in current}
        desired_by_name = {t.name: t for t in desired}

        for name in sorted(desired_by_name.keys() - current_by_name.keys()):
            self._cascade_table_added(desired_by_name[name], parent_schema, parent_db, out)
        for name in sorted(current_by_name.keys() - desired_by_name.keys()):
            self._cascade_table_removed(current_by_name[name], parent_schema, parent_db, out)
        for name in sorted(current_by_name.keys() & desired_by_name.keys()):
            old, new = current_by_name[name], desired_by_name[name]
            if _differ(old, new, _TABLE_EXCLUDE):
                out.append(TableChanged(old=old, new=new, parent_schema=parent_schema, parent_db=parent_db))
            self._diff_columns(parent_db, parent_schema, new, old.columns, new.columns, out)
            self._diff_primary_key(parent_db, parent_schema, old, new, out)

    def _diff_columns(  # noqa: PLR0913
        self,
        parent_db: Database,
        parent_schema: Schema,
        parent_table: Table,
        current: list[Column],
        desired: list[Column],
        out: list[Diff],
    ) -> None:
        current_by_name = {c.name: c for c in current}
        desired_by_name = {c.name: c for c in desired}

        for name in sorted(desired_by_name.keys() - current_by_name.keys()):
            out.append(
                ColumnAdded(
                    column=desired_by_name[name],
                    parent_table=parent_table,
                    parent_schema=parent_schema,
                    parent_db=parent_db,
                )
            )
        for name in sorted(current_by_name.keys() - desired_by_name.keys()):
            out.append(
                ColumnRemoved(
                    column=current_by_name[name],
                    parent_table=parent_table,
                    parent_schema=parent_schema,
                    parent_db=parent_db,
                )
            )
        for name in sorted(current_by_name.keys() & desired_by_name.keys()):
            old, new = current_by_name[name], desired_by_name[name]
            if _differ(old, new, _COLUMN_EXCLUDE):
                out.append(
                    ColumnChanged(
                        old=old,
                        new=new,
                        parent_table=parent_table,
                        parent_schema=parent_schema,
                        parent_db=parent_db,
                    )
                )

    def _diff_primary_key(
        self,
        parent_db: Database,
        parent_schema: Schema,
        old_table: Table,
        new_table: Table,
        out: list[Diff],
    ) -> None:
        """Emit a single PrimaryKeyChanged when the table's key-column set changes."""
        old_pk = [c.name for c in old_table.columns if c.is_key]
        new_pk = [c.name for c in new_table.columns if c.is_key]
        if old_pk != new_pk:
            out.append(
                PrimaryKeyChanged(
                    old_key_columns=old_pk,
                    new_key_columns=new_pk,
                    parent_table=new_table,
                    parent_schema=parent_schema,
                    parent_db=parent_db,
                )
            )

    # ── cascade helpers ──────────────────────────────────────────────────────

    def _cascade_schema_added(self, schema: Schema, parent_db: Database, out: list[Diff]) -> None:
        out.append(SchemaAdded(schema_obj=schema, parent_db=parent_db))
        for table in schema.tables:
            self._cascade_table_added(table, schema, parent_db, out)

    def _cascade_schema_removed(self, schema: Schema, parent_db: Database, out: list[Diff]) -> None:
        out.append(SchemaRemoved(schema_obj=schema, parent_db=parent_db))
        for table in schema.tables:
            self._cascade_table_removed(table, schema, parent_db, out)

    def _cascade_table_added(self, table: Table, parent_schema: Schema, parent_db: Database, out: list[Diff]) -> None:
        # CREATE TABLE renders columns inline, so no ColumnAdded cascade.
        out.append(TableAdded(table=table, parent_schema=parent_schema, parent_db=parent_db))

    def _cascade_table_removed(self, table: Table, parent_schema: Schema, parent_db: Database, out: list[Diff]) -> None:
        # DROP TABLE removes columns implicitly, so no ColumnRemoved cascade.
        out.append(TableRemoved(table=table, parent_schema=parent_schema, parent_db=parent_db))

diff(current, desired)

Return the typed diff list to transform current into desired.

Source code in src/cloe_delta_table_manager/diff/diff_resolver.py
def diff(self, current: DatabaseState, desired: DatabaseState) -> list[Diff]:
    """Return the typed diff list to transform ``current`` into ``desired``."""
    diffs: list[Diff] = []
    self._diff_databases(current, desired, diffs)
    return diffs