feat: add weekly cleanup workflow for stale CI schemas

.github/workflows/cleanup-stale-schemas.yml

@@ -0,0 +1,73 @@
+name: Cleanup stale CI schemas
+
+on:
+  schedule:
+    # Every Sunday at 03:00 UTC
+    - cron: "0 3 * * 0"
+  workflow_dispatch:
+    inputs:
+      max-age-hours:
+        type: string
+        required: false
+        default: "24"
+        description: Drop schemas older than this many hours
+
+env:
+  TESTS_DIR: ${{ github.workspace }}/dbt-data-reliability/integration_tests
+
+jobs:
+  cleanup:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        warehouse-type:
+          - snowflake
+          - bigquery
+          - redshift
+          - databricks_catalog
+          - athena
+    steps:
+      - name: Checkout dbt package
+        uses: actions/checkout@v4
+        with:
+          path: dbt-data-reliability
+
+      - name: Setup Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.10"
+          cache: "pip"
+
+      - name: Install dbt
+        run: >
+          pip install
+          "dbt-core"
+          "dbt-${{ (matrix.warehouse-type == 'databricks_catalog' && 'databricks') || (matrix.warehouse-type == 'athena' && 'athena-community') || matrix.warehouse-type }}"
+
+      - name: Write dbt profiles
+        env:
+          CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }}
+        run: |
+          # The cleanup job doesn't create schemas, but generate_profiles.py
+          # requires --schema-name.  Use a dummy value.
+          python "${{ github.workspace }}/dbt-data-reliability/integration_tests/profiles/generate_profiles.py" \
+            --template "${{ github.workspace }}/dbt-data-reliability/integration_tests/profiles/profiles.yml.j2" \
+            --output ~/.dbt/profiles.yml \
+            --schema-name "cleanup_placeholder"
+
+      - name: Install dbt deps
+        working-directory: ${{ env.TESTS_DIR }}/dbt_project
+        run: dbt deps
+
+      - name: Symlink local elementary package
+        run: ln -sfn ${{ github.workspace }}/dbt-data-reliability ${{ env.TESTS_DIR }}/dbt_project/dbt_packages/elementary
+
+      - name: Drop stale CI schemas
+        working-directory: ${{ env.TESTS_DIR }}/dbt_project
+        # Only dbt_ prefixed schemas are created in this repo's CI.
+        # The elementary repo has its own workflow for py_ prefixed schemas.
+        run: >
+          dbt run-operation elementary.drop_stale_ci_schemas
+          --args '{prefixes: ["dbt_"], max_age_hours: ${{ inputs.max-age-hours || '24' }}}'
+          -t "${{ matrix.warehouse-type }}"

integration_tests/dbt_project/macros/test_drop_stale_ci_schemas.sql

@@ -0,0 +1,75 @@
+{#
+  Integration-test helper for elementary.drop_stale_ci_schemas.
+
+  Creates two CI-style schemas (one with an old timestamp, one recent),
+  runs the cleanup macro, checks which schemas survived, cleans up,
+  and returns a JSON result dict.
+#}
+
+{% macro test_drop_stale_ci_schemas() %}
+  {% set database = elementary.target_database() %}
+  {% set now = modules.datetime.datetime.utcnow() %}
+
+  {# Old schema: timestamp in the past (2020-01-01 00:00:00) #}
+  {% set old_schema = 'dbt_200101_000000_citest_00000000' %}
+  {# Recent schema: timestamp = now #}
+  {% set recent_ts = now.strftime('%y%m%d_%H%M%S') %}
+  {% set recent_schema = 'dbt_' ~ recent_ts ~ '_citest_11111111' %}
+
+  {{ log("TEST: creating old schema: " ~ old_schema, info=true) }}
+  {{ log("TEST: creating recent schema: " ~ recent_schema, info=true) }}
+
+  {# ── Create both schemas ───────────────────────────────────────────── #}
+  {% do elementary_tests.edr_create_schema(database, old_schema) %}
+  {% do elementary_tests.edr_create_schema(database, recent_schema) %}
+
+  {# ── Verify both exist before running cleanup ──────────────────────── #}
+  {% set old_exists_before = elementary.ci_schema_exists(database, old_schema) %}
+  {% set recent_exists_before = elementary.ci_schema_exists(database, recent_schema) %}
+  {{ log("TEST: old_exists_before=" ~ old_exists_before ~ ", recent_exists_before=" ~ recent_exists_before, info=true) }}
+
+  {# ── Run cleanup with a large threshold so only the artificially old
+       schema (year 2020) is caught, and real CI schemas from parallel
+       workers are safely below the threshold. ──────────────────────────── #}
+  {% do elementary.drop_stale_ci_schemas(prefixes=['dbt_'], max_age_hours=8760) %}
+
+  {# ── Check which schemas survived ─────────────────────────────────── #}
+  {% set old_exists_after = elementary.ci_schema_exists(database, old_schema) %}
+  {% set recent_exists_after = elementary.ci_schema_exists(database, recent_schema) %}
+  {{ log("TEST: old_exists_after=" ~ old_exists_after ~ ", recent_exists_after=" ~ recent_exists_after, info=true) }}
+
+  {# ── Cleanup: drop any remaining test schemas ─────────────────────── #}
+  {% if old_exists_after is true %}
+    {% do elementary.drop_ci_schema(database, old_schema) %}
+  {% endif %}
+  {% if recent_exists_after %}
+    {% do elementary.drop_ci_schema(database, recent_schema) %}
+  {% endif %}
+
+  {# ── Return results ────────────────────────────────────────────────── #}
+  {% set results = {
+    "old_exists_before": old_exists_before,
+    "recent_exists_before": recent_exists_before,
+    "old_dropped": not old_exists_after,
+    "recent_kept": recent_exists_after
+  } %}
+  {% do return(results) %}
+{% endmacro %}
+
+
+{# ── Per-adapter schema creation ─────────────────────────────────────── #}
+
+{% macro edr_create_schema(database, schema_name) %}
+  {% do return(adapter.dispatch('edr_create_schema', 'elementary_tests')(database, schema_name)) %}
+{% endmacro %}
+
+{% macro default__edr_create_schema(database, schema_name) %}
+  {% set schema_relation = api.Relation.create(database=database, schema=schema_name) %}
+  {% do dbt.create_schema(schema_relation) %}
+  {% do adapter.commit() %}
+{% endmacro %}
+
+{% macro clickhouse__edr_create_schema(database, schema_name) %}
+  {% do run_query("CREATE DATABASE IF NOT EXISTS `" ~ schema_name ~ "`") %}
+  {% do adapter.commit() %}
+{% endmacro %}

integration_tests/tests/test_ci_cleanup.py

@@ -0,0 +1,21 @@
+"""Integration test for elementary.drop_stale_ci_schemas macro."""
+
+import json
+
+import pytest
+from dbt_project import DbtProject
+
+
+@pytest.mark.skip_targets(["dremio"])
+def test_drop_stale_ci_schemas(dbt_project: DbtProject):
+    """Verify that old CI schemas are dropped and recent ones are kept."""
+    result = dbt_project.dbt_runner.run_operation(
+        "elementary_tests.test_drop_stale_ci_schemas",
+    )
+    assert result, "run_operation returned no output"
+    data = json.loads(result[0])
+
+    assert data["old_exists_before"], "Setup failed: old schema was not created"
+    assert data["recent_exists_before"], "Setup failed: recent schema was not created"
+    assert data["old_dropped"], "Old schema should have been dropped by cleanup"
+    assert data["recent_kept"], "Recent schema should have been kept by cleanup"

macros/utils/ci_cleanup/drop_stale_ci_schemas.sql

@@ -0,0 +1,156 @@
+{#
+  drop_stale_ci_schemas – clean up timestamped CI schemas.
+
+  Schema naming convention produced by CI:
+      <prefix><YYMMDD_HHMMSS>_<branch>_<hash>
+  Examples:
+      dbt_260228_112345_master_abcd1234
+      py_260228_112345_master_abcd1234
+      dbt_260228_112345_master_abcd1234_elementary   (suffixed variant)
+
+  Call from a GitHub Actions workflow via:
+      dbt run-operation elementary.drop_stale_ci_schemas \
+          --args '{prefixes: ["dbt_", "py_"], max_age_hours: 24}'
+#}
+
+{% macro drop_stale_ci_schemas(prefixes=None, max_age_hours=24) %}
+  {% if prefixes is none %}
+    {% set prefixes = ['dbt_', 'py_'] %}
+  {% endif %}
+
+  {% set max_age_hours = max_age_hours | int %}
+  {% set database = elementary.target_database() %}
+  {% set all_schemas = elementary.list_ci_schemas(database) %}
+  {# utcnow() is deprecated in Python 3.12+ but modules.datetime.timezone is not
+     available in dbt's Jinja context. Both now and strptime produce naive datetimes
+     so comparisons are safe. #}
+  {% set now = modules.datetime.datetime.utcnow() %}
+  {% set max_age_seconds = max_age_hours * 3600 %}
+  {% set ns = namespace(dropped=0) %}
+
+  {{ log("CI schema cleanup: scanning " ~ all_schemas | length ~ " schema(s) in database '" ~ database ~ "' for prefixes " ~ prefixes | string, info=true) }}
+
+  {% for schema_name in all_schemas | sort %}
+    {% set schema_lower = schema_name.lower() %}
+    {% for prefix in prefixes %}
+      {% if schema_lower.startswith(prefix.lower()) %}
+        {% set remainder = schema_lower[prefix | length :] %}
+        {# Timestamp format: YYMMDD_HHMMSS (13 chars) followed by _ #}
+        {% if remainder | length >= 14 and remainder[6:7] == '_' and remainder[13:14] == '_' %}
+          {% set ts_str = remainder[:13] %}
+          {# Validate: positions 0-5 and 7-12 must be digits #}
+          {% set digits = ts_str[:6] ~ ts_str[7:] %}
+          {% set ns_valid = namespace(ok=true) %}
+          {% for c in digits %}
+            {% if c not in '0123456789' %}
+              {% set ns_valid.ok = false %}
+            {% endif %}
+          {% endfor %}
+          {% if ns_valid.ok %}
+            {# Validate date component ranges before strptime to avoid ValueError #}
+            {% set mm = ts_str[2:4] | int %}
+            {% set dd = ts_str[4:6] | int %}
+            {% set hh = ts_str[7:9] | int %}
+            {% set mi = ts_str[9:11] | int %}
+            {% set ss = ts_str[11:13] | int %}
+            {% if 1 <= mm <= 12 and 1 <= dd <= 31 and 0 <= hh <= 23 and 0 <= mi <= 59 and 0 <= ss <= 59 %}
+              {% set schema_ts = modules.datetime.datetime.strptime(ts_str, '%y%m%d_%H%M%S') %}
+              {% set age_seconds = (now - schema_ts).total_seconds() %}
+              {% if age_seconds > max_age_seconds %}
+                {{ log("  DROP " ~ schema_name ~ "  (age: " ~ (age_seconds / 3600) | round(1) ~ " h)", info=true) }}
+                {% do elementary.drop_ci_schema(database, schema_name) %}
+                {% set ns.dropped = ns.dropped + 1 %}
+              {% else %}
+                {{ log("  keep " ~ schema_name ~ "  (age: " ~ (age_seconds / 3600) | round(1) ~ " h)", info=true) }}
+              {% endif %}
+            {% else %}
+              {{ log("  skip " ~ schema_name ~ "  (invalid timestamp: " ~ ts_str ~ ")", info=true) }}
+            {% endif %}
+          {% endif %}
+        {% endif %}
+      {% endif %}
+    {% endfor %}
+  {% endfor %}
+
+  {{ log("CI schema cleanup complete. Dropped " ~ ns.dropped ~ " stale schema(s).", info=true) }}
+{% endmacro %}
+
+
+{# ── Per-adapter schema drop ─────────────────────────────────────────── #}
+
+{# ── Per-adapter schema listing ─────────────────────────────────────── #}
+
+{% macro list_ci_schemas(database) %}
+  {% do return(adapter.dispatch('list_ci_schemas', 'elementary')(database)) %}
+{% endmacro %}
+
+{# adapter.list_schemas() is not available in run-operation context
+   (RuntimeDatabaseWrapper does not expose it). Use information_schema instead. #}
+{% macro default__list_ci_schemas(database) %}
+  {% set results = run_query("SELECT schema_name FROM information_schema.schemata") %}
+  {% set schemas = [] %}
+  {% for row in results %}
+    {% do schemas.append(row[0]) %}
+  {% endfor %}
+  {% do return(schemas) %}
+{% endmacro %}
+
+{% macro bigquery__list_ci_schemas(database) %}
+  {% set results = run_query("SELECT schema_name FROM `" ~ database ~ "`.INFORMATION_SCHEMA.SCHEMATA") %}
+  {% set schemas = [] %}
+  {% for row in results %}
+    {% do schemas.append(row[0]) %}
+  {% endfor %}
+  {% do return(schemas) %}
+{% endmacro %}
+
+{% macro clickhouse__list_ci_schemas(database) %}
+  {% set results = run_query('SHOW DATABASES') %}
+  {% set schemas = [] %}
+  {% for row in results %}
+    {% do schemas.append(row[0]) %}
+  {% endfor %}
+  {% do return(schemas) %}
+{% endmacro %}
+
+
+{# ── Per-adapter schema drop ─────────────────────────────────────────── #}
+
+{% macro drop_ci_schema(database, schema_name) %}
+  {% do return(adapter.dispatch('drop_ci_schema', 'elementary')(database, schema_name)) %}
+{% endmacro %}
+
+{% macro default__drop_ci_schema(database, schema_name) %}
+  {% set schema_relation = api.Relation.create(database=database, schema=schema_name) %}
+  {% do dbt.drop_schema(schema_relation) %}
+  {% do adapter.commit() %}
+{% endmacro %}
+
+{% macro clickhouse__drop_ci_schema(database, schema_name) %}
+  {% do run_query("DROP DATABASE IF EXISTS `" ~ schema_name ~ "`") %}
+  {% do adapter.commit() %}
+{% endmacro %}
+
+
+{# ── Per-adapter schema existence check (run-operation safe) ──────── #}
+{# adapter.check_schema_exists() is not available in run-operation context.
+   This is only used by the integration test; the main macro does not need it. #}
+
+{% macro ci_schema_exists(database, schema_name) %}
+  {% do return(adapter.dispatch('ci_schema_exists', 'elementary')(database, schema_name)) %}
+{% endmacro %}
+
+{% macro default__ci_schema_exists(database, schema_name) %}
+  {% set result = run_query("SELECT schema_name FROM information_schema.schemata WHERE lower(schema_name) = lower('" ~ schema_name ~ "')") %}
+  {% do return(result | length > 0) %}
+{% endmacro %}
+
+{% macro bigquery__ci_schema_exists(database, schema_name) %}
+  {% set result = run_query("SELECT schema_name FROM `" ~ database ~ "`.INFORMATION_SCHEMA.SCHEMATA WHERE lower(schema_name) = lower('" ~ schema_name ~ "')") %}
+  {% do return(result | length > 0) %}
+{% endmacro %}
+
+{% macro clickhouse__ci_schema_exists(database, schema_name) %}
+  {% set result = run_query("SELECT 1 FROM system.databases WHERE name = '" ~ schema_name ~ "' LIMIT 1") %}
+  {% do return(result | length > 0) %}
+{% endmacro %}

macros/utils/cross_db_utils/schema_exists.sql

@@ -29,6 +29,11 @@
   {% do return(adapter.check_schema_exists(database, schema)) %}
 {% endmacro %}
 
+{% macro clickhouse__schema_exists(database, schema) %}
+  {% set result = run_query("SELECT 1 FROM system.databases WHERE name = '" ~ schema ~ "' LIMIT 1") %}
+  {% do return(result | length > 0) %}
+{% endmacro %}
+
 {% macro default__schema_exists(database, schema) %}
   {% do return(adapter.check_schema_exists(database, schema)) %}
 {% endmacro %}