feat(janitor): archive stale relations before deletion in direct mode

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

@@ -389,6 +389,14 @@ def relation_exists(
         )
         return cursor.fetchone() is not None
 
+    def with_relation_age_metadata(
+        self,
+        *,
+        connection: Any,
+        relations: tuple[RelationInfo, ...],
+    ) -> tuple[RelationInfo, ...]:
+        return relations
+
     def list_relations(
         self,
         *,
@@ -626,6 +634,9 @@ def render_drop_view(self, *, destination: str, if_exists: bool = True) -> tuple
     def render_rename(self, *, origin: str, destination: str) -> tuple[str, ...]:
         return (f"ALTER TABLE {origin} RENAME TO {destination}",)
 
+    def render_rename_view(self, *, origin: str, destination: str) -> tuple[str, ...]:
+        return (f"ALTER VIEW {origin} RENAME TO {destination}",)
+
     def render_swap(self, *, left: str, right: str) -> tuple[str, ...]:
         staging: str = self._with_replaced_relation_name(
             relation=left, name=f"{self._relation_name(left)}__swap_staging"
@@ -2342,6 +2353,21 @@ def render_create_audit_result_index_sqls(
         del database, schema
         return ()
 
+    def render_create_janitor_event_table_sql(self, *, database: str | None, schema: str) -> str:
+        """Render DDL that creates the janitor audit event table when it is missing."""
+
+        from sqlbuild.executor.janitor_events.main.create_table_sql import (
+            build_janitor_events_create_table_sql,
+        )
+
+        return build_janitor_events_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

@@ -97,6 +97,8 @@ def supports_durable_clone(self) -> bool:
         return False
 
     def supports_relation_age_metadata(self) -> bool:
+        """Return False because the DuckDB and MotherDuck catalogs record no relation timestamps."""
+
         return False
 
     def supports_table_freshness_metadata(self) -> bool:
@@ -1029,6 +1031,18 @@ def render_create_audit_result_index_sqls(
         del database, schema
         return ()
 
+    def render_create_janitor_event_table_sql(self, *, database: str | None, schema: str) -> str:
+        from sqlbuild.executor.janitor_events.main.create_table_sql import (
+            build_janitor_events_create_table_sql,
+        )
+
+        return build_janitor_events_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,
         *,
@@ -1311,6 +1325,14 @@ def list_relations(
             for row in rows
         )
 
+    def with_relation_age_metadata(
+        self,
+        *,
+        connection: Any,
+        relations: tuple[RelationInfo, ...],
+    ) -> tuple[RelationInfo, ...]:
+        return relations
+
     def list_functions(
         self,
         *,
@@ -1626,6 +1648,10 @@ def render_rename(self, *, origin: str, destination: str) -> tuple[str, ...]:
         unqualified_destination: str = destination.rsplit(".", 1)[-1]
         return (f"ALTER TABLE {origin} RENAME TO {unqualified_destination}",)
 
+    def render_rename_view(self, *, origin: str, destination: str) -> tuple[str, ...]:
+        unqualified_destination: str = destination.rsplit(".", 1)[-1]
+        return (f"ALTER VIEW {origin} RENAME TO {unqualified_destination}",)
+
     def render_swap(self, *, left: str, right: str) -> tuple[str, ...]:
         staging: str = self._with_replaced_relation_name(
             relation=left, name=f"{self._relation_name(left)}__swap_staging"

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

@@ -23,6 +23,16 @@ def list_relations(
         """Return all relations in the given database/schema scope."""
         ...
 
+    @abstractmethod
+    def with_relation_age_metadata(
+        self,
+        *,
+        connection: Any,
+        relations: tuple[RelationInfo, ...],
+    ) -> tuple[RelationInfo, ...]:
+        """Return listed relations with age timestamps that listing does not read inline."""
+        ...
+
     @abstractmethod
     def list_functions(
         self,

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

@@ -312,6 +312,11 @@ def render_rename(self, *, origin: str, destination: str) -> tuple[str, ...]:
         """Render SQL statements that rename a relation."""
         ...
 
+    @abstractmethod
+    def render_rename_view(self, *, origin: str, destination: str) -> tuple[str, ...]:
+        """Render SQL statements that rename a view relation."""
+        ...
+
     @abstractmethod
     def render_swap(self, *, left: str, right: str) -> tuple[str, ...]:
         """Render SQL statements that swap two relations."""
@@ -875,6 +880,11 @@ def render_create_audit_result_index_sqls(
         """Render optional audit result table index DDL statements."""
         ...
 
+    @abstractmethod
+    def render_create_janitor_event_table_sql(self, *, database: str | None, schema: str) -> str:
+        """Render DDL that creates the janitor audit event table when it is missing."""
+        ...
+
     @abstractmethod
     def render_prune_fingerprint_history_sql(
         self,

src/sqlbuild/adapter/relations/_helpers/identifier_fitting.py

@@ -0,0 +1,35 @@
+"""Identifier-limit fitting implementations for generated relation names."""
+
+from __future__ import annotations
+
+import hashlib
+
+from sqlbuild.adapter.relations.constants import SHORTENED_LOGICAL_NAME_HASH_LENGTH
+from sqlbuild.errors.contracts.exceptions import SharedInputError
+
+
+def fit_artifact_logical_name_impl(
+    *, logical_name: str, fixed_prefix: str, identifier_limit: int, artifact_label: str
+) -> str:
+    """Fit a readable logical component with a deterministic hash suffix."""
+
+    max_logical_length: int = identifier_limit - len(fixed_prefix)
+    if max_logical_length < 1:
+        raise SharedInputError(
+            f"{artifact_label} prefix '{fixed_prefix}' does not fit within identifier "
+            f"limit {identifier_limit}"
+        )
+    if len(logical_name) <= max_logical_length:
+        return logical_name
+
+    suffix_length: int = SHORTENED_LOGICAL_NAME_HASH_LENGTH + 1
+    if max_logical_length <= suffix_length:
+        raise SharedInputError(
+            f"{artifact_label} name for '{logical_name}' cannot fit within identifier "
+            f"limit {identifier_limit}"
+        )
+    logical_hash: str = hashlib.sha256(logical_name.encode("utf-8")).hexdigest()[
+        :SHORTENED_LOGICAL_NAME_HASH_LENGTH
+    ]
+    prefix_length: int = max_logical_length - suffix_length
+    return f"{logical_name[:prefix_length]}_{logical_hash}"

src/sqlbuild/adapter/relations/constants.py

@@ -1,3 +1,4 @@
 """Shared relation metadata limits."""
 
 METADATA_NAME_FILTER_LIMIT: int = 250
+SHORTENED_LOGICAL_NAME_HASH_LENGTH: int = 8

src/sqlbuild/adapter/relations/main/fit_artifact_logical_name.py

@@ -0,0 +1,18 @@
+"""Generated relation name fitting entrypoint."""
+
+from __future__ import annotations
+
+from sqlbuild.adapter.relations._helpers.identifier_fitting import fit_artifact_logical_name_impl
+
+
+def fit_artifact_logical_name(
+    *, logical_name: str, fixed_prefix: str, identifier_limit: int, artifact_label: str
+) -> str:
+    """Fit a readable logical component after a fixed prefix within an identifier limit."""
+
+    return fit_artifact_logical_name_impl(
+        logical_name=logical_name,
+        fixed_prefix=fixed_prefix,
+        identifier_limit=identifier_limit,
+        artifact_label=artifact_label,
+    )

src/sqlbuild/adapter/relations/main/relation_age_timestamp.py

@@ -0,0 +1,15 @@
+"""Relation catalog timestamp normalization entrypoint."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+
+def relation_age_timestamp_utc(value: object) -> datetime | None:
+    """Return one catalog timestamp as aware UTC, treating offset-free values as UTC."""
+
+    if not isinstance(value, datetime):
+        return None
+    if value.tzinfo is None or value.utcoffset() is None:
+        return value.replace(tzinfo=UTC)
+    return value.astimezone(UTC)

src/sqlbuild/adapter/state_sql/_helpers/table_ddl.py

@@ -0,0 +1,32 @@
+"""Portable append-only state table DDL implementations."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+
+from sqlbuild.adapter.contract.types import FrameworkType
+from sqlbuild.sql_values.types import StateSqlValueType
+
+
+def render_state_table_create_sql_impl(
+    *,
+    qualified_name: str,
+    columns: tuple[str, ...],
+    column_types: Mapping[str, StateSqlValueType],
+    required_columns: frozenset[str],
+    render_framework_type: Callable[[FrameworkType], str],
+    transient: bool,
+) -> str:
+    """Render create-if-missing DDL for string, integer, and text-timestamp state columns."""
+
+    rendered_types: dict[StateSqlValueType, str] = {
+        StateSqlValueType.STRING: render_framework_type(FrameworkType.STRING),
+        StateSqlValueType.INTEGER: render_framework_type(FrameworkType.INTEGER),
+        StateSqlValueType.TEXT_TIMESTAMP: render_framework_type(FrameworkType.TIMESTAMP),
+    }
+    definitions: list[str] = []
+    for column in columns:
+        required: str = " NOT NULL" if column in required_columns else ""
+        definitions.append(f"{column} {rendered_types[column_types[column]]}{required}")
+    table_kind: str = "TRANSIENT TABLE" if transient else "TABLE"
+    return f"CREATE {table_kind} IF NOT EXISTS {qualified_name} ({', '.join(definitions)})"

src/sqlbuild/adapter/state_sql/main/render_state_table_create_sql.py

@@ -0,0 +1,28 @@
+"""Public append-only state table DDL rendering operation."""
+
+from collections.abc import Callable, Mapping
+
+from sqlbuild.adapter.contract.types import FrameworkType
+from sqlbuild.adapter.state_sql._helpers.table_ddl import render_state_table_create_sql_impl
+from sqlbuild.sql_values.types import StateSqlValueType
+
+
+def render_state_table_create_sql(
+    *,
+    qualified_name: str,
+    columns: tuple[str, ...],
+    column_types: Mapping[str, StateSqlValueType],
+    required_columns: frozenset[str],
+    render_framework_type: Callable[[FrameworkType], str],
+    transient: bool = False,
+) -> str:
+    """Render create-if-missing DDL for one append-only state table."""
+
+    return render_state_table_create_sql_impl(
+        qualified_name=qualified_name,
+        columns=columns,
+        column_types=column_types,
+        required_columns=required_columns,
+        render_framework_type=render_framework_type,
+        transient=transient,
+    )

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

@@ -4,6 +4,7 @@
 
 import json
 import logging
+from dataclasses import replace
 from datetime import date, datetime
 from decimal import Decimal
 from pathlib import Path
@@ -64,6 +65,7 @@
 from sqlbuild.adapter.relations.main.get_columns_for_relations import (
     get_columns_for_relations_bulk,
 )
+from sqlbuild.adapter.relations.main.relation_age_timestamp import relation_age_timestamp_utc
 from sqlbuild.adapter.state_sql.main.render_insert_source_freshness_records_sql import (
     render_insert_source_freshness_records_sql,
 )
@@ -282,6 +284,18 @@ def render_create_audit_result_index_sqls(
         del database, schema
         return ()
 
+    def render_create_janitor_event_table_sql(self, *, database: str | None, schema: str) -> str:
+        from sqlbuild.executor.janitor_events.main.create_table_sql import (
+            build_janitor_events_create_table_sql,
+        )
+
+        return build_janitor_events_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,
         *,
@@ -350,7 +364,7 @@ def render_prune_source_freshness_history_sql(
         )
 
     def supports_relation_age_metadata(self) -> bool:
-        return False
+        return True
 
     def supports_table_freshness_metadata(self) -> bool:
         return True
@@ -1825,6 +1839,10 @@ def render_rename(self, *, origin: str, destination: str) -> tuple[str, ...]:
         destination_name: str = self._strip_identifier_quotes(destination).split(".")[-1]
         return (f"ALTER TABLE {self._quote_identifier_path(origin)} RENAME TO {destination_name}",)
 
+    def render_rename_view(self, *, origin: str, destination: str) -> tuple[str, ...]:
+        del origin, destination
+        raise AdapterUserError(message="BigQuery does not support renaming views")
+
     def render_swap(self, *, left: str, right: str) -> tuple[str, ...]:
         raise AdapterUserError(message="BigQuery does not support atomic table swap")
 
@@ -2030,6 +2048,59 @@ def list_relations(
                     raise
         return tuple(relations)
 
+    def with_relation_age_metadata(
+        self,
+        *,
+        connection: _BigQueryConnection,
+        relations: tuple[RelationInfo, ...],
+    ) -> tuple[RelationInfo, ...]:
+        timestamps: dict[tuple[str | None, str, str], tuple[datetime | None, datetime | None]] = {}
+        dataset_key: tuple[str | None, str]
+        for dataset_key in sorted(
+            {(relation.database, str(relation.schema)) for relation in relations},
+            key=lambda key: (key[0] or "", key[1]),
+        ):
+            timestamps.update(
+                self._dataset_relation_timestamps(
+                    connection=connection, database=dataset_key[0], schema=dataset_key[1]
+                )
+            )
+        return tuple(
+            replace(
+                relation,
+                created_at=timestamps.get(self._age_key(relation), (None, None))[0],
+                last_altered_at=timestamps.get(self._age_key(relation), (None, None))[1],
+            )
+            for relation in relations
+        )
+
+    def _dataset_relation_timestamps(
+        self, *, connection: _BigQueryConnection, database: str | None, schema: str
+    ) -> dict[tuple[str | None, str, str], tuple[datetime | None, datetime | None]]:
+        dataset_id: str = self._build_dataset_id(database=database, schema=schema)
+        try:
+            cursor: _BigQueryCursor = self.execute(
+                connection=connection,
+                sql="SELECT table_id, TIMESTAMP_MILLIS(creation_time), "
+                f"TIMESTAMP_MILLIS(last_modified_time) FROM `{dataset_id}.__TABLES__`",
+            )
+            rows: list[tuple[Any, ...]] = cursor.fetchall()
+        except Exception as error:
+            if not self._is_google_not_found(error):
+                raise
+            return {}
+        return {
+            (database, schema, str(row[0])): (
+                relation_age_timestamp_utc(row[1]),
+                relation_age_timestamp_utc(row[2]),
+            )
+            for row in rows
+        }
+
+    @staticmethod
+    def _age_key(relation: RelationInfo) -> tuple[str | None, str, str]:
+        return (relation.database, str(relation.schema), relation.name)
+
     def list_functions(
         self,
         *,