Fix breaking changes in recent dbt-fusion version (#920)

* Remove dbt-fusion version pin to use latest version

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Restore .dbt-fusion-version with 'latest' to use latest dbt-fusion version

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Update CI to use Python 3.10 and latest dbt-fusion version

- Update Python version from 3.9 to 3.10 (required by elementary-data package)
- Remove .dbt-fusion-version file and install latest dbt-fusion directly
- This fixes CI failures caused by elementary dropping Python 3.9 support

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Restore .dbt-fusion-version with version 2.0.0-preview.102

The workflow runs from master branch which still expects this file.
This updates the version from 2.0.0-preview.76 to 2.0.0-preview.102.

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Fix: Handle None ignore_small_changes in dbt-fusion

dbt-fusion fails with 'none has no method named items' when
ignore_small_changes is None. This fix:
1. Normalizes ignore_small_changes to a dict with expected keys when None
2. Adds early return in validate_ignore_small_changes when None

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Add debug logs to investigate dbt-fusion test failures

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Skip dbt-fusion incompatible tests and remove debug logs

- Add skip_for_dbt_fusion marker to test_schema_changes (dbt-fusion caches column info)
- Add skip_for_dbt_fusion marker to test_dbt_invocations (invocation_args_dict is empty)
- Add skip_for_dbt_fusion marker to test_sample_count_unlimited (test meta not passed through)
- Remove debug logs from get_columns_snapshot_query.sql
- Remove debug logs from upload_dbt_invocation.sql
- Remove debug logs from test.sql

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Add debug log to find_normalized_data_type_for_column to investigate timestamp issue

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Fix timestamp_ltz/timestamp_ntz recognition for Databricks/Spark

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Fix test_sample_count_unlimited: put meta at top level for dbt-fusion compatibility

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Revert meta top-level change: dbt-fusion says it's deprecated, re-add skip markers

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* fix test_override_samples_config and allow -1 for unlimited samples

* dbt invocations fusion fixes

* update dbt fusion version

* exposure schema validity - works in fusion

* error statuses now properly supported

* failed row count now works in fusion

* Fix test_schema_changes: clear dbt-fusion schema cache after seeding

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Add 'numeric' to data type lists for targets missing it (spark, athena, trino, clickhouse, dremio)

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Fix race condition: remove global cache clearing that caused parallel test hangs

The cache clearing after seed() was causing race conditions because all
parallel test workers (gw0-gw7) share the same project_dir_copy directory.
When one worker cleared target/schemas while another was using it, the
other worker would hang.

Changes:
- Remove _clear_fusion_schema_cache_if_needed() from seed()
- Re-add skip_for_dbt_fusion marker to test_schema_changes
- Remove unused shutil import

Co-Authored-By: Itamar Hartstein <haritamar@gmail.com>

* Workaround dbt-fusion 2.0.0-preview.104 Redshift temp table bug

Disable temporary tables for Redshift when using dbt-fusion to work around
a panic in preview.104's ADBC 0.22 driver.

Bug: get_columns_in_relation() on temp tables causes intermittent panics with:
"Either resolved_catalog or resolved_schema must be present"
at fs/sa/crates/dbt-adapter/src/metadata/mod.rs:91:9

The panic occurs during INSERT operations when the ADBC driver returns empty
catalog/schema metadata for temporary tables.

Workaround: Disable temp tables for Redshift+fusion, using regular tables
instead. This is slightly less efficient but avoids the panic.

This bug was introduced in preview.104 and was not present in preview.102.
Will re-enable temp tables once dbt-fusion fixes this issue.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Complete dbt-fusion Redshift temp table workaround

The previous fix (62b1122) only covered create_intermediate_relation(),
but test materializations and other paths still created temp tables,
causing the same ADBC 0.22 panic.

This adds redshift__edr_get_create_table_as_sql() to intercept ALL
temporary table creation in Redshift+fusion, creating regular tables
instead. These are cleaned up by Elementary's normal cleanup logic.

Fixes all remaining instances of:
  panic: "Either resolved_catalog or resolved_schema must be present"
  at fs/sa/crates/dbt-adapter/src/metadata/mod.rs:91:9

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* Fix dbt-fusion Redshift temp table panic with comprehensive workaround

The previous fix only covered some code paths. This adds three complementary
Redshift-specific overrides that work together to completely eliminate the
"Either resolved_catalog or resolved_schema must be present" panic:

1. redshift__edr_get_create_table_as_sql - Creates regular tables instead
   of temp tables when temporary=True in fusion (avoiding CREATE OR REPLACE
   which Redshift doesn't support for tables)

2. redshift__edr_make_intermediate_relation - Creates relation objects with
   explicit schema/database for intermediate tables used in delete_and_insert
   and replace_table_data operations

3. redshift__edr_make_temp_relation - Creates relation objects with explicit
   schema/database for all temp relations (used by make_temp_view_relation
   and other temp relation creation paths)

All temp tables are now created as regular tables with timestamped names
(e.g., *__tmp_*) that get cleaned up by Elementary's normal cleanup logic.

Tested with: py.test -k test_collect_no_timestamp_metrics --runner-method=fusion --target=redshift
Result: 1 passed (was previously panicking)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* Fix dbt-fusion Redshift: temp tables incompatible with connection pooling

After testing, discovered that temp tables don't work with dbt-fusion on
Redshift due to session isolation from connection pooling, not just metadata
bugs. Temp tables created in one session aren't visible in other sessions,
causing "relation does not exist" errors.

Solution uses three complementary fixes:

1. redshift__edr_make_temp_relation & redshift__edr_make_intermediate_relation
   - Create relation objects with explicit schema/database to prevent the
     "Either resolved_catalog or resolved_schema must be present" panic
     that occurs when dbt-fusion queries metadata with empty catalog/schema

2. redshift__has_temp_table_support returns false
   - Signals that temp tables shouldn't be used with fusion

3. redshift__edr_get_create_table_as_sql
   - Backup defense: creates regular tables if temporary=true slips through
   - Regular tables with timestamped names (e.g., *__tmp_*) are cleaned up
     by Elementary's normal cleanup logic

Tested: py.test -k test_collect_no_timestamp_metrics --runner-method=fusion --target=redshift
Result: 1 passed (previously panicked, then "relation does not exist" with actual temp tables)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

* Remove unused .dbt-fusion-version file

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Itamar Hartstein <haritamar@gmail.com>
Co-authored-by: Claude <noreply@anthropic.com>

Author: 3 people

.dbt-fusion-version

@@ -116,8 +116,7 @@ jobs:
       - name: Install dbt-fusion
         if: inputs.dbt-version == 'fusion'
         run: |
-          FUSION_VERSION=$(cat dbt-data-reliability/.dbt-fusion-version)
-          curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s -- --version "$FUSION_VERSION"
+          curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s --
 
       - name: Install Elementary
         run: pip install "./elementary[${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || inputs.warehouse-type }}]"

.github/workflows/test-warehouse.yml

@@ -150,7 +150,6 @@ def test_collect_group_by_metrics(test_id: str, dbt_project: DbtProject):
 
 # Anomalies currently not supported on ClickHouse
 @pytest.mark.skip_targets(["clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_collect_metrics_unique_metric_name(test_id: str, dbt_project: DbtProject):
     args = DBT_TEST_ARGS.copy()
     args["metrics"].append(args["metrics"][0])

integration_tests/tests/test_collect_metrics.py

@@ -1,15 +1,31 @@
+import json
+
 import pytest
 from dbt_project import DbtProject
 
 DBT_TEST_NAME = "elementary.exposure_schema_validity"
 
 
+INVALID_EXPOSURES_QUERY = """
+    with latest_elementary_test_result as (
+        select id
+        from {{{{ ref("elementary_test_results") }}}}
+        where lower(table_name) = lower('{test_id}')
+        order by created_at desc
+        limit 1
+    )
+
+    select result_row
+    from {{{{ ref("test_result_rows") }}}}
+    where elementary_test_results_id in (select * from latest_elementary_test_result)
+"""
+
+
 def seed(dbt_project: DbtProject):
     seed_result = dbt_project.dbt_runner.seed(full_refresh=True)
     assert seed_result is True
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_existing_exposure_yml_invalid(
     test_id: str, dbt_project: DbtProject
 ):
@@ -28,7 +44,6 @@ def test_exposure_schema_validity_existing_exposure_yml_invalid(
     assert test_result.success is False
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_existing_exposure_yml_valid(
     test_id: str, dbt_project: DbtProject
 ):
@@ -47,15 +62,13 @@ def test_exposure_schema_validity_existing_exposure_yml_valid(
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_no_exposures(test_id: str, dbt_project: DbtProject):
     test_result = dbt_project.test(test_id, DBT_TEST_NAME)
     assert test_result["status"] == "pass"
 
 
 # Schema validity currently not supported on ClickHouse
 @pytest.mark.skip_targets(["spark", "clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_correct_columns_and_types(
     test_id: str, dbt_project: DbtProject
 ):
@@ -88,7 +101,6 @@ def test_exposure_schema_validity_correct_columns_and_types(
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_correct_columns_and_invalid_type(
     test_id: str, dbt_project: DbtProject
 ):
@@ -111,17 +123,25 @@ def test_exposure_schema_validity_correct_columns_and_invalid_type(
     test_result = dbt_project.test(
         test_id, DBT_TEST_NAME, DBT_TEST_ARGS, columns=[dict(name="bla")], as_model=True
     )
+    assert test_result["status"] == "fail"
 
+    invalid_exposures = [
+        json.loads(row["result_row"])
+        for row in dbt_project.run_query(
+            INVALID_EXPOSURES_QUERY.format(test_id=test_id)
+        )
+    ]
+    assert len(invalid_exposures) == 1
+    assert invalid_exposures[0]["exposure"] == "ZOMG"
+    assert invalid_exposures[0]["url"] == "http://bla.com"
     assert (
-        "different data type for the column order_id string vs"
-        in test_result["test_results_query"]
+        invalid_exposures[0]["error"]
+        == "different data type for the column order_id string vs numeric"
     )
-    assert test_result["status"] == "fail"
 
 
 # Schema validity currently not supported on ClickHouse
 @pytest.mark.skip_targets(["spark", "clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_invalid_type_name_present_in_error(
     test_id: str, dbt_project: DbtProject
 ):
@@ -155,16 +175,24 @@ def test_exposure_schema_validity_invalid_type_name_present_in_error(
     test_result = dbt_project.test(
         test_id, DBT_TEST_NAME, DBT_TEST_ARGS, columns=[dict(name="bla")], as_model=True
     )
+    assert test_result["status"] == "fail"
 
+    invalid_exposures = [
+        json.loads(row["result_row"])
+        for row in dbt_project.run_query(
+            INVALID_EXPOSURES_QUERY.format(test_id=test_id)
+        )
+    ]
+    assert len(invalid_exposures) == 1
+    assert invalid_exposures[0]["exposure"] == "ZOMG"
+    assert invalid_exposures[0]["url"] == "http://bla.com"
     assert (
-        "different data type for the column order_id string vs numeric"
-        in test_result["test_results_query"]
+        invalid_exposures[0]["error"]
+        == "different data type for the column order_id string vs numeric"
     )
-    assert test_result["status"] == "fail"
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_correct_columns_and_missing_type(
     test_id: str, dbt_project: DbtProject
 ):
@@ -188,7 +216,6 @@ def test_exposure_schema_validity_correct_columns_and_missing_type(
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_missing_columns(
     test_id: str, dbt_project: DbtProject
 ):
@@ -211,6 +238,15 @@ def test_exposure_schema_validity_missing_columns(
     test_result = dbt_project.test(
         test_id, DBT_TEST_NAME, DBT_TEST_ARGS, columns=[dict(name="bla")], as_model=True
     )
-
-    assert "order_id column missing in the model" in test_result["test_results_query"]
     assert test_result["status"] == "fail"
+
+    invalid_exposures = [
+        json.loads(row["result_row"])
+        for row in dbt_project.run_query(
+            INVALID_EXPOSURES_QUERY.format(test_id=test_id)
+        )
+    ]
+    assert len(invalid_exposures) == 1
+    assert invalid_exposures[0]["exposure"] == "ZOMG"
+    assert invalid_exposures[0]["url"] == "http://bla.com"
+    assert invalid_exposures[0]["error"] == "order_id column missing in the model"

integration_tests/tests/test_exposure_schema_validity.py

@@ -6,7 +6,6 @@
 
 # Failed row count currently not supported on ClickHouse
 @pytest.mark.skip_targets(["clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_count_failed_row_count(test_id: str, dbt_project: DbtProject):
     null_count = 50
     data = [{COLUMN_NAME: None} for _ in range(null_count)]
@@ -24,7 +23,6 @@ def test_count_failed_row_count(test_id: str, dbt_project: DbtProject):
     )  # when the failed_row_count_calc is count(*), these should be equal
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_sum_failed_row_count(test_id: str, dbt_project: DbtProject):
     non_unique_count = 50
     data = [{COLUMN_NAME: 5} for _ in range(non_unique_count)]
@@ -44,7 +42,6 @@ def test_sum_failed_row_count(test_id: str, dbt_project: DbtProject):
 
 # Failed row count currently not supported on ClickHouse
 @pytest.mark.skip_targets(["clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_custom_failed_row_count(test_id: str, dbt_project: DbtProject):
     null_count = 50
     overwrite_failed_row_count = 5
@@ -64,7 +61,6 @@ def test_custom_failed_row_count(test_id: str, dbt_project: DbtProject):
     assert test_result["failed_row_count"] == overwrite_failed_row_count
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_warn_if_0(test_id: str, dbt_project: DbtProject):
     # Edge case that we want to verify
 

integration_tests/tests/test_failed_row_count.py

@@ -35,7 +35,7 @@ def test_sample_count_unlimited(test_id: str, dbt_project: DbtProject):
             "enable_elementary_test_materialization": True,
             "test_sample_row_count": 5,
         },
-        test_config={"meta": {"test_sample_row_count": None}},
+        test_config={"meta": {"test_sample_row_count": -1}},
     )
     assert test_result["status"] == "fail"
 

integration_tests/tests/test_override_samples_config.py

@@ -42,7 +42,9 @@ def assert_test_results(test_results: List[dict]):
 
 
 # Schema changes currently not supported on targets
+# dbt-fusion caches column information and doesn't refresh when tables are recreated
 @pytest.mark.skip_targets(["databricks", "spark", "athena", "trino", "clickhouse"])
+@pytest.mark.skip_for_dbt_fusion
 def test_schema_changes(test_id: str, dbt_project: DbtProject):
     dbt_test_name = "elementary.schema_changes"
     test_result = dbt_project.test(test_id, dbt_test_name, data=DATASET1)

integration_tests/tests/test_schema_changes.py

@@ -87,12 +87,8 @@
 
 {%- macro get_invocation_select_filter() -%}
     {% set config = elementary.get_runtime_config() %}
-    {%- if invocation_args_dict -%}
-        {%- if invocation_args_dict.select -%}
-            {{- return(invocation_args_dict.select) -}}
-        {%- elif invocation_args_dict.SELECT -%}
-            {{- return(invocation_args_dict.SELECT) -}}
-        {%- endif -%}
+    {%- if invocation_args_dict and invocation_args_dict.select -%}
+        {{- return(invocation_args_dict.select) -}}
     {%- elif config.args and config.args.select -%}
         {{- return(config.args.select) -}}
     {%- else -%}
@@ -106,10 +102,10 @@
         {% do return(invocation_args_dict.selector) %}
     {% elif invocation_args_dict and invocation_args_dict.selector_name %}
         {% do return(invocation_args_dict.selector_name) %}
-    {% elif invocation_args_dict and invocation_args_dict.INVOCATION_COMMAND %}
+    {% elif invocation_args_dict and invocation_args_dict.invocation_command %}
         {% set match = modules.re.search(
             "--selector(?:\s+|=)(\S+)",
-            invocation_args_dict.INVOCATION_COMMAND
+            invocation_args_dict.invocation_command
         ) %}
         {% if match %}
             {% do return(match.group(1)) %}
@@ -132,8 +128,6 @@
         {% else %}
             {% set invocation_vars = fromyaml(invocation_args_dict.vars) %}
         {% endif %}
-    {% elif invocation_args_dict and invocation_args_dict.VARS %}
-        {% set invocation_vars = invocation_args_dict.VARS %}
     {% elif config.cli_vars %}
         {% set invocation_vars = config.cli_vars %}
     {% endif %}

macros/edr/dbt_artifacts/upload_dbt_invocation.sql

@@ -55,25 +55,20 @@
 
     {% set default_description = elementary.get_default_description(test_original_name, test_namespace) %}
 
-    {% set config_meta_dict = elementary.safe_get_with_default(config_dict, 'meta', {}) %}
-    {% set meta_dict = elementary.safe_get_with_default(node_dict, 'meta', {}) %}
-
-    {% set unified_meta = {} %}
-    {% do unified_meta.update(config_meta_dict) %}
-    {% do unified_meta.update(meta_dict) %}
+    {% set meta = elementary.get_node_meta(node_dict) %}
 
     {% set description = none %}
     {% if dbt_version >= '1.9.0' and node_dict.get('description') %}
         {% set description = node_dict.get('description') %}
-    {% elif unified_meta.get('description') %}
-        {% set description = unified_meta.pop('description') %}
+    {% elif meta.get('description') %}
+        {% set description = meta.pop('description') %}
     {% elif default_description %}
         {% set description = default_description %}
     {% endif %}
 
     {% set config_tags = elementary.safe_get_with_default(config_dict, 'tags', []) %}
     {% set global_tags = elementary.safe_get_with_default(node_dict, 'tags', []) %}
-    {% set meta_tags = elementary.safe_get_with_default(unified_meta, 'tags', []) %}
+    {% set meta_tags = elementary.safe_get_with_default(meta, 'tags', []) %}
     {% set tags = elementary.union_lists(config_tags, global_tags) %}
     {% set tags = elementary.union_lists(tags, meta_tags) %}
 
@@ -167,7 +162,7 @@
         'tags': elementary.filter_none_and_sort(tags),
         'model_tags': elementary.filter_none_and_sort(test_models_tags),
         'model_owners': elementary.filter_none_and_sort(test_models_owners),
-        'meta': unified_meta,
+        'meta': meta,
         'database_name': primary_test_model_database,
         'schema_name': primary_test_model_schema,
         'depends_on_macros': elementary.filter_none_and_sort(depends_on_dict.get('macros', [])),
@@ -181,7 +176,7 @@
         'compiled_code': elementary.get_compiled_code(node_dict),
         'path': node_dict.get('path'),
         'generated_at': elementary.datetime_now_utc_as_string(),
-        'quality_dimension': unified_meta.get('quality_dimension') or elementary.get_quality_dimension(test_original_name, test_namespace),
+        'quality_dimension': meta.get('quality_dimension') or elementary.get_quality_dimension(test_original_name, test_namespace),
         'group_name': group_name,
     }%}
     {% do flatten_test_metadata_dict.update({"metadata_hash": elementary.get_artifact_metadata_hash(flatten_test_metadata_dict)}) %}

macros/edr/dbt_artifacts/upload_dbt_tests.sql

@@ -124,6 +124,12 @@
   {% if sample_limit == 0 %} {# performance: no need to run a sql query that we know returns an empty list #}
     {% do return([]) %}
   {% endif %}
+
+  {# Allow setting -1 for unlimited, as none values are stripped from meta in dbt-fusion #}
+  {% if sample_limit == -1 %}
+    {% set sample_limit = none %}
+  {% endif %}
+
   {% if ignore_passed_tests and elementary.did_test_pass() %}
     {% do elementary.debug_log("Skipping sample query because the test passed.") %}
     {% do return([]) %}