Move the incremental macro lookup higher to trigger errors

Add tests that check for the new error string

dbt/include/spark/macros/materializations/incremental/incremental.sql

@@ -6,6 +6,8 @@
 
   {%- set file_format = dbt_spark_validate_get_file_format(raw_file_format) -%}
   {%- set strategy = dbt_spark_validate_get_incremental_strategy(raw_strategy, file_format) -%}
+  {#-- Get the incremental_strategy, the macro to use for the strategy, and build the sql --#}
+  {% set strategy_sql_macro_func = adapter.get_incremental_strategy_macro(context, strategy) %}
 
   {#-- Set vars --#}
 
@@ -56,12 +58,9 @@
       {{ create_table_as(True, tmp_relation, compiled_code, language) }}
     {%- endcall -%}
     {%- do process_schema_changes(on_schema_change, tmp_relation, existing_relation) -%}
-
-    {#-- Get the incremental_strategy, the macro to use for the strategy, and build the sql --#}
-    {% set strategy_sql_macro_func = adapter.get_incremental_strategy_macro(context, strategy) %}
+    {#-- call the incremental strategy macro --#}
     {% set strategy_arg_dict = ({'target_relation': target_relation, 'temp_relation': tmp_relation, 'unique_key': unique_key, 'dest_columns': dest_columns, 'predicates': incremental_predicates }) %}
     {% set build_sql = strategy_sql_macro_func(strategy_arg_dict) %}
-
     {%- call statement('main') -%}
       {{ build_sql }}
     {%- endcall -%}

tests/functional/adapter/incremental_strategies/test_incremental_strategies.py

@@ -128,7 +128,6 @@ def models(self):
         return {
             "bad_file_format.sql": bad_file_format_sql,
             "bad_merge_not_delta.sql": bad_merge_not_delta_sql,
-            "bad_strategy.sql": bad_strategy_sql,
         }
 
     @staticmethod
@@ -142,3 +141,23 @@ def run_and_test():
     @pytest.mark.skip_profile("databricks_http_cluster", "spark_session")
     def test_bad_strategies(self, project):
         self.run_and_test()
+
+
+class TestBadCustomStrategy(BaseIncrementalStrategies):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "bad_strategy.sql": bad_strategy_sql,
+        }
+
+    @staticmethod
+    def run_and_test():
+        run_results = run_dbt(["run"], expect_pass=False)
+        # assert all models fail with compilation errors
+        for result in run_results:
+            assert result.status == "error"
+            assert "dbt could not find an incremental strategy macro" in result.message
+
+    @pytest.mark.skip_profile("databricks_http_cluster", "spark_session")
+    def test_bad_custom_strategies(self, project):
+        self.run_and_test()