feat(migrations): keep incremental and snapshot history when models are renamed

Makefile

@@ -130,6 +130,7 @@ E2E_DUCKDB_CLI_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/init \
 	tests/e2e/src/sqlbuild/cli/commands/main/rules \
 	tests/e2e/src/sqlbuild/cli/commands/main/lineage \
+	tests/e2e/src/sqlbuild/cli/commands/main/model_migrations \
 	tests/e2e/src/sqlbuild/cli/commands/main/motherduck \
 	tests/e2e/src/sqlbuild/cli/commands/main/playground \
 	tests/e2e/src/sqlbuild/cli/commands/main/postgres \

src/sqlbuild/adapter/contract/classes/base_adapter.py

@@ -36,6 +36,7 @@
     ExpressionInferenceProfile,
     FunctionDefinition,
     FunctionInfo,
+    MigrationStagePlan,
     QueryResult,
     RelationInfo,
     RowDiffCoverage,
@@ -57,6 +58,7 @@
     HistoricalSnapshotCloseStyle,
     HistoricalSnapshotInsertStyle,
     LoaderLogicalType,
+    MigrationTransfer,
     PromotionStrategy,
     SnapshotLatestVersionStyle,
     SnapshotUpdateStyle,
@@ -702,6 +704,29 @@ def render_durable_clone(
         del origin_is_transient
         return self.render_create_table_as(destination=destination, sql=f"SELECT * FROM {origin}")
 
+    def render_migration_stage(
+        self,
+        *,
+        origin: str,
+        stage: str,
+        origin_is_transient: bool = False,
+        stage_is_transient: bool | None = None,
+    ) -> MigrationStagePlan:
+        del origin_is_transient, stage_is_transient
+        return MigrationStagePlan(
+            transfer=MigrationTransfer.COPY,
+            statements=(f"CREATE TABLE {stage} AS SELECT * FROM {origin}",),
+        )
+
+    def capture_dependent_view_rebinds(
+        self, *, connection: Any, database: str | None, schema: str, name: str
+    ) -> tuple[str, ...]:
+        del connection, database, schema, name
+        return ()
+
+    def supports_transactional_ddl(self) -> bool:
+        return False
+
     def render_query_with_cursor_bounds(
         self,
         *,
@@ -2195,6 +2220,21 @@ def render_create_janitor_event_table_sql(self, *, database: str | None, schema:
             transient=self.state_tables_transient,
         )
 
+    def render_create_migration_state_table_sql(self, *, database: str | None, schema: str) -> str:
+        """Render DDL that creates the model migration event table when it is missing."""
+
+        from sqlbuild.compiler.migrations.main.create_table_sql import (
+            build_migration_state_create_table_sql,
+        )
+
+        return build_migration_state_create_table_sql(
+            database=database,
+            schema=schema,
+            render_qualified_name=self.render_qualified_name,
+            render_framework_type=self.render_framework_type,
+            transient=self.state_tables_transient,
+        )
+
     def render_prune_fingerprint_history_sql(
         self,
         *,

src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py

@@ -41,6 +41,7 @@
     ExpressionInferenceProfile,
     FunctionDefinition,
     FunctionInfo,
+    MigrationStagePlan,
     QueryResult,
     RelationInfo,
     RowDiffColumnResult,
@@ -61,6 +62,7 @@
     CursorKind,
     FrameworkType,
     LoaderLogicalType,
+    MigrationTransfer,
     PromotionStrategy,
     TablePromotionMode,
 )
@@ -851,6 +853,18 @@ def render_create_janitor_event_table_sql(self, *, database: str | None, schema:
             render_framework_type=self.render_framework_type,
         )
 
+    def render_create_migration_state_table_sql(self, *, database: str | None, schema: str) -> str:
+        from sqlbuild.compiler.migrations.main.create_table_sql import (
+            build_migration_state_create_table_sql,
+        )
+
+        return build_migration_state_create_table_sql(
+            database=database,
+            schema=schema,
+            render_qualified_name=self.render_qualified_name,
+            render_framework_type=self.render_framework_type,
+        )
+
     def render_prune_fingerprint_history_sql(
         self,
         *,
@@ -1524,6 +1538,29 @@ def render_durable_clone(
         del origin_is_transient
         return self.render_create_table_as(destination=destination, sql=f"SELECT * FROM {origin}")
 
+    def render_migration_stage(
+        self,
+        *,
+        origin: str,
+        stage: str,
+        origin_is_transient: bool = False,
+        stage_is_transient: bool | None = None,
+    ) -> MigrationStagePlan:
+        del origin_is_transient, stage_is_transient
+        return MigrationStagePlan(
+            transfer=MigrationTransfer.COPY,
+            statements=(f"CREATE TABLE {stage} AS SELECT * FROM {origin}",),
+        )
+
+    def capture_dependent_view_rebinds(
+        self, *, connection: Any, database: str | None, schema: str, name: str
+    ) -> tuple[str, ...]:
+        del connection, database, schema, name
+        return ()
+
+    def supports_transactional_ddl(self) -> bool:
+        return True
+
     def render_query_with_cursor_bounds(
         self,
         *,

src/sqlbuild/adapter/contract/classes/strict_adapter.py

@@ -15,6 +15,7 @@
     ColumnInfo,
     CursorValue,
     ExpressionInferenceProfile,
+    MigrationStagePlan,
     RowDiffTolerance,
     RowDiffTolerances,
     SnapshotChangeTarget,
@@ -341,6 +342,30 @@ def render_durable_clone(
         """Render SQL statements that clone/copy into a durable independent destination."""
         ...
 
+    @abstractmethod
+    def render_migration_stage(
+        self,
+        *,
+        origin: str,
+        stage: str,
+        origin_is_transient: bool = False,
+        stage_is_transient: bool | None = None,
+    ) -> MigrationStagePlan:
+        """Render statements that create a fresh, independent stage holding the origin's data."""
+        ...
+
+    @abstractmethod
+    def capture_dependent_view_rebinds(
+        self, *, connection: Any, database: str | None, schema: str, name: str
+    ) -> tuple[str, ...]:
+        """Capture statements that re-point identity-bound dependent views at this table name."""
+        ...
+
+    @abstractmethod
+    def supports_transactional_ddl(self) -> bool:
+        """Return whether renames and state inserts can commit or roll back together."""
+        ...
+
     @abstractmethod
     def render_query_with_cursor_bounds(
         self,
@@ -885,6 +910,11 @@ def render_create_janitor_event_table_sql(self, *, database: str | None, schema:
         """Render DDL that creates the janitor audit event table when it is missing."""
         ...
 
+    @abstractmethod
+    def render_create_migration_state_table_sql(self, *, database: str | None, schema: str) -> str:
+        """Render DDL that creates the model migration event table when it is missing."""
+        ...
+
     @abstractmethod
     def render_prune_fingerprint_history_sql(
         self,

src/sqlbuild/adapter/contract/models.py

@@ -15,6 +15,7 @@
     HistoricalSnapshotCloseStyle,
     HistoricalSnapshotInsertStyle,
     LifeCycleEventKind,
+    MigrationTransfer,
     RetentionChangePhase,
     RetentionScope,
     SnapshotLatestVersionStyle,
@@ -184,6 +185,16 @@ class RenderedRetentionChange:
     statements: tuple[str, ...]
 
 
+@dataclass(frozen=True)
+class MigrationStagePlan:
+    """Fresh migration stage statements, plus a copy used only for recognized clone refusals."""
+
+    transfer: MigrationTransfer
+    statements: tuple[str, ...]
+    fallback_statements: tuple[str, ...] = ()
+    is_clone_refusal: Callable[[BaseException], bool] | None = None
+
+
 @dataclass(frozen=True)
 class FunctionInfo:
     """Metadata for one discovered warehouse function or routine."""

src/sqlbuild/adapter/contract/types.py

@@ -36,6 +36,19 @@ class PromotionStrategy(StrEnum):
     CREATE_NEW = "create_new"
 
 
+class MigrationTransfer(StrEnum):
+    """How a model migration stage receives the origin's data."""
+
+    CLONE = "clone"
+    COPY = "copy"
+
+    @property
+    def label(self) -> str:
+        """Return the human-readable plan label for this transfer."""
+
+        return "zero-copy clone" if self == MigrationTransfer.CLONE else "physical copy"
+
+
 class TablePromotionMode(StrEnum):
     IMMEDIATE = "immediate"
     STAGED = "staged"

src/sqlbuild/adapters/bigquery/_helpers/clone_refusal.py

@@ -0,0 +1,40 @@
+"""Recognize BigQuery refusals to clone a table where a physical copy can still succeed."""
+
+from __future__ import annotations
+
+from sqlbuild.adapters.bigquery.constants import (
+    CLONE_REFUSAL_ERROR_CODE,
+    CLONE_REFUSAL_PHRASES,
+    CLONE_REFUSAL_REASONS,
+    MAX_CLONE_REFUSAL_ERROR_CHAIN,
+)
+
+
+def is_bigquery_clone_refusal(error: BaseException) -> bool:
+    """Return whether a driver error in the cause chain is a clone-capability refusal."""
+
+    current: BaseException | None = error
+    depth: int = 0
+    while current is not None and depth < MAX_CLONE_REFUSAL_ERROR_CHAIN:
+        if _is_refusal(current):
+            return True
+        current = current.__cause__ or current.__context__
+        depth += 1
+    return False
+
+
+def _is_refusal(error: BaseException) -> bool:
+    details: object | None = getattr(error, "errors", None)
+    if getattr(error, "code", None) != CLONE_REFUSAL_ERROR_CODE or not isinstance(details, list):
+        return False
+    reasons: set[str] = {
+        str(detail.get("reason", "")) for detail in details if isinstance(detail, dict)
+    }
+    messages: str = " ".join(
+        str(detail.get("message", "")) for detail in details if isinstance(detail, dict)
+    ).lower()
+    return (
+        bool(reasons)
+        and reasons <= CLONE_REFUSAL_REASONS
+        and any(phrase in messages for phrase in CLONE_REFUSAL_PHRASES)
+    )

src/sqlbuild/adapters/bigquery/classes/bigquery_adapter.py

@@ -46,6 +46,7 @@
     ExpressionInferenceProfile,
     FunctionDefinition,
     FunctionInfo,
+    MigrationStagePlan,
     QueryResult,
     RelationInfo,
     RenderedRetentionChange,
@@ -73,6 +74,7 @@
     HistoricalSnapshotCloseStyle,
     HistoricalSnapshotInsertStyle,
     LoaderLogicalType,
+    MigrationTransfer,
     PromotionStrategy,
     RetentionChangePhase,
     RetentionScope,
@@ -93,6 +95,7 @@
 from sqlbuild.adapter.type_system.main.first_arg_nullability import first_arg_nullability
 from sqlbuild.adapter.type_system.main.normalize_numeric_family import normalize_numeric_family
 from sqlbuild.adapter.type_system.main.types_equal import types_equal
+from sqlbuild.adapters.bigquery._helpers.clone_refusal import is_bigquery_clone_refusal
 from sqlbuild.adapters.bigquery._helpers.statement_telemetry import affected_rows
 from sqlbuild.adapters.bigquery.classes.bigquery_connection import _BigQueryConnection
 from sqlbuild.adapters.bigquery.classes.bigquery_cursor import _BigQueryCursor
@@ -321,6 +324,18 @@ def render_create_janitor_event_table_sql(self, *, database: str | None, schema:
             render_framework_type=self.render_framework_type,
         )
 
+    def render_create_migration_state_table_sql(self, *, database: str | None, schema: str) -> str:
+        from sqlbuild.compiler.migrations.main.create_table_sql import (
+            build_migration_state_create_table_sql,
+        )
+
+        return build_migration_state_create_table_sql(
+            database=database,
+            schema=schema,
+            render_qualified_name=self.render_qualified_name,
+            render_framework_type=self.render_framework_type,
+        )
+
     def render_prune_fingerprint_history_sql(
         self,
         *,
@@ -1712,6 +1727,33 @@ def render_durable_clone(
             f"CLONE {self._quote_identifier_path(origin)}",
         )
 
+    def render_migration_stage(
+        self,
+        *,
+        origin: str,
+        stage: str,
+        origin_is_transient: bool = False,
+        stage_is_transient: bool | None = None,
+    ) -> MigrationStagePlan:
+        del origin_is_transient, stage_is_transient
+        quoted_stage: str = self._quote_identifier_path(stage)
+        quoted_origin: str = self._quote_identifier_path(origin)
+        return MigrationStagePlan(
+            transfer=MigrationTransfer.CLONE,
+            statements=(f"CREATE TABLE {quoted_stage} CLONE {quoted_origin}",),
+            fallback_statements=(f"CREATE TABLE {quoted_stage} COPY {quoted_origin}",),
+            is_clone_refusal=is_bigquery_clone_refusal,
+        )
+
+    def capture_dependent_view_rebinds(
+        self, *, connection: Any, database: str | None, schema: str, name: str
+    ) -> tuple[str, ...]:
+        del connection, database, schema, name
+        return ()
+
+    def supports_transactional_ddl(self) -> bool:
+        return False
+
     def render_query_with_cursor_bounds(
         self,
         *,

src/sqlbuild/adapters/bigquery/constants.py

@@ -17,3 +17,17 @@
 SELECT_STATEMENT_TYPE: str = "SELECT"
 TABLE_NAME_WILDCARD: str = "*"
 TIMESTAMP_TYPE_TOKEN: str = "TIMESTAMP"
+
+CLONE_REFUSAL_ERROR_CODE: int = 400
+CLONE_REFUSAL_REASONS: frozenset[str] = frozenset({"invalid", "invalidQuery"})
+CLONE_REFUSAL_PHRASES: tuple[str, ...] = (
+    "cannot clone",
+    "can't clone",
+    "clone is not supported",
+    "clones are not supported",
+    "cloning is not supported",
+    "not supported for clone",
+    "clones and snapshots",
+    "clone or snapshot",
+)
+MAX_CLONE_REFUSAL_ERROR_CHAIN: int = 8