Replace environment variable with a project flag to gate microbatch functionality

.changes/unreleased/Features-20241001-161422.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Change gating of microbatch feature to be behind project flag / behavior flag
+time: 2024-10-01T16:14:22.267253-05:00
+custom:
+  Author: MichelleArk QMalcolm
+  Issue: "10798"

core/dbt/context/providers.py

@@ -238,12 +238,12 @@ def resolve_limit(self) -> Optional[int]:
     def resolve_event_time_filter(self, target: ManifestNode) -> Optional[EventTimeFilter]:
         event_time_filter = None
         if (
-            os.environ.get("DBT_EXPERIMENTAL_MICROBATCH")
-            and (isinstance(target.config, NodeConfig) or isinstance(target.config, SourceConfig))
+            (isinstance(target.config, NodeConfig) or isinstance(target.config, SourceConfig))
             and target.config.event_time
             and isinstance(self.model, ModelNode)
             and self.model.config.materialized == "incremental"
             and self.model.config.incremental_strategy == "microbatch"
+            and self.manifest.use_microbatch_batches(project_name=self.config.project_name)
         ):
             start = self.model.config.get("__dbt_internal_microbatch_event_time_start")
             end = self.model.config.get("__dbt_internal_microbatch_event_time_end")

core/dbt/contracts/graph/manifest.py

@@ -714,10 +714,10 @@ def __init__(self):
         self._macros_by_name = {}
         self._macros_by_package = {}
 
-    def find_macro_by_name(
+    def find_macro_candidate_by_name(
         self, name: str, root_project_name: str, package: Optional[str]
-    ) -> Optional[Macro]:
-        """Find a macro in the graph by its name and package name, or None for
+    ) -> Optional[MacroCandidate]:
+        """Find a MacroCandidate in the graph by its name and package name, or None for
         any package. The root project name is used to determine priority:
          - locally defined macros come first
          - then imported macros
@@ -735,7 +735,15 @@ def filter(candidate: MacroCandidate) -> bool:
             filter=filter,
         )
 
-        return candidates.last()
+        return candidates.last_candidate()
+
+    def find_macro_by_name(
+        self, name: str, root_project_name: str, package: Optional[str]
+    ) -> Optional[Macro]:
+        macro_candidate = self.find_macro_candidate_by_name(
+            name=name, root_project_name=root_project_name, package=package
+        )
+        return macro_candidate.macro if macro_candidate else None
 
     def find_generate_macro_by_name(
         self, component: str, root_project_name: str, imported_package: Optional[str] = None
@@ -1747,6 +1755,24 @@ def __reduce_ex__(self, protocol):
         )
         return self.__class__, args
 
+    def _microbatch_macro_is_core(self, project_name: str) -> bool:
+        microbatch_is_core = False
+        candidate = self.find_macro_candidate_by_name(
+            name="get_incremental_microbatch_sql", root_project_name=project_name, package=None
+        )
+
+        # We want to check for "Core", because "Core" basically means "builtin"
+        if candidate is not None and candidate.locality == Locality.Core:
+            microbatch_is_core = True
+
+        return microbatch_is_core
+
+    def use_microbatch_batches(self, project_name: str) -> bool:
+        return (
+            get_flags().require_batched_execution_for_custom_microbatch_strategy
+            or self._microbatch_macro_is_core(project_name=project_name)
+        )
+
 
 class MacroManifest(MacroMethods):
     def __init__(self, macros) -> None:

core/dbt/contracts/project.py

@@ -338,6 +338,7 @@ class ProjectFlags(ExtensibleDbtClassMixin):
     write_json: Optional[bool] = None
 
     # legacy behaviors - https://github.com/dbt-labs/dbt-core/blob/main/docs/guides/behavior-change-flags.md
+    require_batched_execution_for_custom_microbatch_strategy: bool = False
     require_explicit_package_overrides_for_builtin_materializations: bool = True
     require_resource_names_without_spaces: bool = False
     source_freshness_run_project_hooks: bool = False
@@ -350,6 +351,7 @@ class ProjectFlags(ExtensibleDbtClassMixin):
     @property
     def project_only_flags(self) -> Dict[str, Any]:
         return {
+            "require_batched_execution_for_custom_microbatch_strategy": self.require_batched_execution_for_custom_microbatch_strategy,
             "require_explicit_package_overrides_for_builtin_materializations": self.require_explicit_package_overrides_for_builtin_materializations,
             "require_resource_names_without_spaces": self.require_resource_names_without_spaces,
             "source_freshness_run_project_hooks": self.source_freshness_run_project_hooks,

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 @@
 
         self.check_for_model_deprecations()
         self.check_for_spaces_in_resource_names()
+        self.check_for_microbatch_deprecations()
 
         return self.manifest
 
@@ -649,6 +650,23 @@
             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:
@@ -1390,7 +1408,7 @@
             node.config.final_validate()
 
     def check_valid_microbatch_config(self):
-        if os.environ.get("DBT_EXPERIMENTAL_MICROBATCH"):
+        if self.manifest.use_microbatch_batches(project_name=self.root_project.project_name):
             for node in self.manifest.nodes.values():
                 if (
                     node.config.materialized == "incremental"

core/dbt/task/run.py

@@ -1,5 +1,4 @@
 import functools
-import os
 import threading
 import time
 from copy import deepcopy
@@ -482,11 +481,10 @@ def execute(self, model, manifest):
             )
 
         hook_ctx = self.adapter.pre_model_hook(context_config)
-
         if (
-            os.environ.get("DBT_EXPERIMENTAL_MICROBATCH")
-            and model.config.materialized == "incremental"
+            model.config.materialized == "incremental"
             and model.config.incremental_strategy == "microbatch"
+            and manifest.use_microbatch_batches(project_name=self.config.project_name)
         ):
             return self._execute_microbatch_model(
                 hook_ctx, context_config, model, manifest, context, materialization_macro