Fire deprecation warning for custom microbatch macros

core/dbt/deprecations.py

@@ -128,6 +128,11 @@ class MFCumulativeTypeParamsDeprecation(DBTDeprecation):
     _event = "MFCumulativeTypeParamsDeprecation"
 
 
+class MicrobatchMacroOutsideOfBatchesDeprecation(DBTDeprecation):
+    _name = "microbatch-macro-outside-of-batches-deprecation"
+    _event = "MicrobatchMacroOutsideOfBatchesDeprecation"
+
+
 def renamed_env_var(old_name: str, new_name: str):
     class EnvironmentVariableRenamed(DBTDeprecation):
         _name = f"environment-variable-renamed:{old_name}"
@@ -178,6 +183,7 @@ def show_callback():
     SourceFreshnessProjectHooksNotRun(),
     MFTimespineWithoutYamlConfigurationDeprecation(),
     MFCumulativeTypeParamsDeprecation(),
+    MicrobatchMacroOutsideOfBatchesDeprecation(),
 ]
 
 deprecations: Dict[str, DBTDeprecation] = {d.name: d for d in deprecations_list}

core/dbt/events/core_types.proto

@@ -461,6 +461,14 @@ message MFCumulativeTypeParamsDeprecationMsg {
     MFCumulativeTypeParamsDeprecation data = 2;
 }
 
+// D020
+message MicrobatchMacroOutsideOfBatchesDeprecation {}
+
+message MicrobatchMacroOutsideOfBatchesDeprecationMsg {
+    CoreEventInfo info = 1;
+    MicrobatchMacroOutsideOfBatchesDeprecation data = 2;
+}
+
 // I065
 message DeprecatedModel {
     string model_name = 1;

core/dbt/events/types.py

@@ -486,6 +486,16 @@ def message(self) -> str:
         return line_wrap_message(warning_tag(description))
 
 
+class MicrobatchMacroOutsideOfBatchesDeprecation(WarnLevel):
+    def code(self) -> str:
+        return "D020"
+
+    def message(self) -> str:
+        description = "The use of a custom microbatch macro outside of batched execution is deprecated. To use it with batched execution, set `flags.require_batched_execution_for_custom_microbatch_strategy` to `True` in `dbt_project.yml`. In the future this will be the default behavior."
+
+        return line_wrap_message(warning_tag(description))
+
+
 # =======================================================
 # I - Project parsing
 # =======================================================

core/dbt/parser/manifest.py

@@ -509,6 +509,7 @@ def load(self) -> Manifest:
 
         self.check_for_model_deprecations()
         self.check_for_spaces_in_resource_names()
+        self.check_for_microbatch_deprecations()
 
         return self.manifest
 
@@ -649,6 +650,23 @@ def check_for_spaces_in_resource_names(self):
             else:  # ERROR level
                 raise DbtValidationError("Resource names cannot contain spaces")
 
+    def check_for_microbatch_deprecations(self) -> None:
+        if not get_flags().require_batched_execution_for_custom_microbatch_strategy:
+            has_microbatch_model = False
+            for _, node in self.manifest.nodes.items():
+                if (
+                    isinstance(node, ModelNode)
+                    and node.config.materialized == "incremental"
+                    and node.config.incremental_strategy == "microbatch"
+                ):
+                    has_microbatch_model = True
+                    break
+
+            if has_microbatch_model and self.manifest._microbatch_macro_is_core(
+                self.root_project.project_name
+            ):
+                dbt.deprecations.warn("microbatch-macro-outside-of-batches-deprecation")
+
     def load_and_parse_macros(self, project_parser_files):
         for project in self.all_projects.values():
             if project.project_name not in project_parser_files:

tests/functional/microbatch/test_microbatch.py

@@ -2,7 +2,11 @@
 
 import pytest
 
-from dbt.events.types import LogModelResult, MicrobatchModelNoEventTimeInputs
+from dbt.events.types import (
+    LogModelResult,
+    MicrobatchMacroOutsideOfBatchesDeprecation,
+    MicrobatchModelNoEventTimeInputs,
+)
 from dbt.tests.util import (
     get_artifact,
     patch_microbatch_end_time,
@@ -159,6 +163,10 @@ def project_config_update(self):
             }
         }
 
+    @pytest.fixture(scope="class")
+    def deprecation_catcher(self) -> EventCatcher:
+        return EventCatcher(MicrobatchMacroOutsideOfBatchesDeprecation)
+
 
 class TestMicrobatchCustomUserStrategyDefault(BaseMicrobatchCustomUserStrategy):
     @pytest.fixture(scope="class")
@@ -169,7 +177,11 @@ def project_config_update(self):
             }
         }
 
-    def test_use_custom_microbatch_strategy_by_default(self, project):
+    def test_use_custom_microbatch_strategy_by_default(
+        self,
+        project,
+        deprecation_catcher: EventCatcher,
+    ):
         with mock.patch.object(
             type(project.adapter), "valid_incremental_strategies", lambda _: []
         ):
@@ -181,11 +193,15 @@ def test_use_custom_microbatch_strategy_by_default(self, project):
             assert "custom microbatch strategy" in logs
             # The custom strategy wasn't used with batch functionality
             assert "START batch" not in logs
+            # Deprecation warning about custom microbatch macro fired
+            assert len(deprecation_catcher.caught_events) == 0
 
 
 class TestMicrobatchCustomUserStrategyProjectFlagTrueValid(BaseMicrobatchCustomUserStrategy):
     def test_use_custom_microbatch_strategy_project_flag_true_invalid_incremental_strategy(
-        self, project
+        self,
+        project,
+        deprecation_catcher: EventCatcher,
     ):
         with mock.patch.object(
             type(project.adapter), "valid_incremental_strategies", lambda _: ["microbatch"]
@@ -200,6 +216,8 @@ def test_use_custom_microbatch_strategy_project_flag_true_invalid_incremental_st
             assert "custom microbatch strategy" in logs
             # The custom strategy was used with batch functionality
             assert "START batch" in logs
+            # Deprecation warning about custom microbatch macro not fired
+            assert len(deprecation_catcher.caught_events) == 0
 
 
 class TestMicrobatchCustomUserStrategyProjectFlagTrueInvalid(BaseMicrobatchCustomUserStrategy):