Fix failing dbt-fusion tests when run in parallel in CI

.github/workflows/test.yml

@@ -477,6 +477,15 @@ jobs:
           uv pip install --system hatch
           hatch -e tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }}-${{ matrix.dbt-version }} run pip freeze
 
+      - name: Set RESOURCE_PREFIX without periods
+        id: set-resource-prefix
+        run: |
+          PYTHON_VER=$(echo "${{ matrix.python-version }}" | tr -d '.')
+          AIRFLOW_VER=$(echo "${{ matrix.airflow-version }}" | tr -d '.')
+          DBT_VER=$(echo "${{ matrix.dbt-version }}" | tr -d '.')
+          PREFIX="COSMOS_${{ github.run_id }}_PY${PYTHON_VER}_AF${AIRFLOW_VER}_DBT${DBT_VER}"
+          echo "prefix=${PREFIX}" >> $GITHUB_OUTPUT
+
       - name: Test Cosmos against Airflow ${{ matrix.airflow-version }}, Python ${{ matrix.python-version }} and dbt ${{ matrix.dbt-version }}
         run: |
           hatch run tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }}-${{ matrix.dbt-version }}:test-integration-dbtf-setup
@@ -492,6 +501,7 @@ jobs:
           SNOWFLAKE_SCHEMA: ${{ secrets.SNOWFLAKE_SCHEMA }}
           SNOWFLAKE_WAREHOUSE: ${{ secrets.SNOWFLAKE_WAREHOUSE }}
           SNOWFLAKE_DATABASE: ${{ secrets.SNOWFLAKE_DATABASE }}
+          RESOURCE_PREFIX: ${{ steps.set-resource-prefix.outputs.prefix }}
 
       - name: Upload coverage to Github
         uses: actions/upload-artifact@v4
@@ -500,6 +510,22 @@ jobs:
           path: .coverage
           include-hidden-files: true
 
+      - name: Clean up Snowflake resources
+        if: always()
+        run: |
+         pip install snowflake-connector-python
+         # Trigger a python script to delete the resources
+         python scripts/ci_dbtf_delete_snowflake_resources.py
+        env:
+          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
+          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
+          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}
+          SNOWFLAKE_SCHEMA: ${{ secrets.SNOWFLAKE_SCHEMA }}
+          SNOWFLAKE_WAREHOUSE: ${{ secrets.SNOWFLAKE_WAREHOUSE }}
+          SNOWFLAKE_DATABASE: ${{ secrets.SNOWFLAKE_DATABASE }}
+          RESOURCE_PREFIX: ${{ steps.set-resource-prefix.outputs.prefix }}
+
+
   Run-Performance-Tests:
     needs: Authorize
     runs-on: ubuntu-latest

dev/dags/dbt/jaffle_shop/macros/generate_alias_name.sql

@@ -0,0 +1,8 @@
+{% macro generate_alias_name(custom_alias_name=None, node=None) -%}
+  {%- set base = custom_alias_name if custom_alias_name else node.name -%}
+  {%- if env_var('RESOURCE_PREFIX', '') -%}
+    {{ return(env_var('RESOURCE_PREFIX') ~ '_' ~ base) }}
+  {%- else -%}
+    {{ base }}
+  {%- endif -%}
+{%- endmacro %}

scripts/ci_dbtf_delete_snowflake_resources.py

@@ -0,0 +1,59 @@
+import os
+import re
+
+import snowflake.connector
+
+
+def delete_snowflake_resource():
+    """
+    Delete Snowflake resources with a given prefix(set as an environment variable).
+    """
+    conn = snowflake.connector.connect(
+        user=os.environ["SNOWFLAKE_USER"],
+        password=os.environ["SNOWFLAKE_PASSWORD"],
+        account=os.environ["SNOWFLAKE_ACCOUNT"],
+        warehouse=os.environ["SNOWFLAKE_WAREHOUSE"],
+        database=os.environ["SNOWFLAKE_DATABASE"],
+        schema=os.environ["SNOWFLAKE_SCHEMA"],
+    )
+    prefix = os.getenv("RESOURCE_PREFIX", "")
+    if prefix:
+        cursor = conn.cursor()
+        # Use parameterized query for the SELECT (safe for values)
+        query = """
+        SELECT table_name, table_type
+        FROM information_schema.tables
+        WHERE table_schema = %s
+        AND table_name LIKE %s
+        """
+        cursor.execute(query, (os.environ["SNOWFLAKE_SCHEMA"], f"{prefix}_%"))
+        resources = cursor.fetchall()
+
+        for resource_name, resource_type in resources:
+            # Validate table name contains only safe characters (prevent injection)
+            if not re.match(r"^[A-Z0-9_]+$", resource_name):
+                print(f"Skipping potentially unsafe table name: {resource_name}")
+                continue
+
+            try:
+                if resource_type == "BASE TABLE":
+                    # Table names must be part of SQL string, not parameterized
+                    drop_sql = f"DROP TABLE IF EXISTS {resource_name}"
+                    print(f"Executing: {drop_sql}")
+                    cursor.execute(drop_sql)
+                elif resource_type == "VIEW":
+                    drop_sql = f"DROP VIEW IF EXISTS {resource_name}"
+                    print(f"Executing: {drop_sql}")
+                    cursor.execute(drop_sql)
+            except Exception as e:
+                print(f"Failed to drop {resource_name}: {e}")
+
+        cursor.close()
+        print(f"Processed {len(resources)} resources")
+    else:
+        print("No RESOURCE_PREFIX set, skipping cleanup")
+    conn.close()
+
+
+if __name__ == "__main__":
+    delete_snowflake_resource()

tests/dbt/test_graph.py

@@ -1908,9 +1908,9 @@ def test_save_dbt_ls_cache(mock_variable_set, mock_datetime, tmp_dbt_project_dir
     assert hash_args == "d41d8cd98f00b204e9800998ecf8427e"
     if sys.platform == "darwin":
         # We faced inconsistent hashing versions depending on the version of MacOS/Linux - the following line aims to address these.
-        assert hash_dir in ("c2c47529eaec412281bdb243a479b734", "fa93914ecb491cc40a17ab956397359a")
+        assert hash_dir in ("c2c47529eaec412281bdb243a479b734", "71bbf303ad4e06a7b1e2be20e0b73c0d")
     else:
-        assert hash_dir == "fa93914ecb491cc40a17ab956397359a"
+        assert hash_dir == "71bbf303ad4e06a7b1e2be20e0b73c0d"
 
 
 @pytest.mark.integration