Switch from environment variable to behavior flag for gating microbatch functionality

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

@@ -0,0 +1,7 @@
+kind: Features
+body: Use a behavior flag to gate microbatch functionality (instead of an environment
+  variable)
+time: 2024-10-01T16:54:06.121016-05:00
+custom:
+  Author: QMalcolm
+  Issue: "327"

dbt-tests-adapter/dbt/tests/adapter/incremental/test_incremental_microbatch.py

@@ -1,6 +1,4 @@
-import os
 from pprint import pformat
-from unittest import mock
 
 import pytest
 
@@ -63,7 +61,14 @@ def assert_row_count(self, project, relation_name: str, expected_row_count: int)
 
         assert len(result) == expected_row_count, f"{relation_name}:{pformat(result)}"
 
-    @mock.patch.dict(os.environ, {"DBT_EXPERIMENTAL_MICROBATCH": "True"})
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "require_builtin_microbatch_strategy": True,
+            }
+        }
+
     def test_run_with_event_time(self, project, insert_two_rows_sql):
         # initial run -- backfills all data
         with patch_microbatch_end_time("2020-01-03 13:57:00"):

dbt/adapters/base/impl.py

@@ -22,7 +22,6 @@
     Union,
     TYPE_CHECKING,
 )
-import os
 import pytz
 from dbt_common.behavior_flags import Behavior, BehaviorFlag
 from dbt_common.clients.jinja import CallableMacroGenerator
@@ -312,7 +311,13 @@ def _behavior_flags(self) -> List[BehaviorFlag]:
         """
         This method should be overwritten by adapter maintainers to provide platform-specific flags
         """
-        return []
+        return [
+            {
+                "name": "require_builtin_microbatch_strategy",
+                "default": False,
+                "description": "If True, then a builtin `microbatch` materialization strategy can be expected to exist.",
+            }
+        ]
 
     ###
     # Methods that pass through to the connection manager
@@ -1570,7 +1575,7 @@ def valid_incremental_strategies(self):
 
     def builtin_incremental_strategies(self):
         builtin_strategies = ["append", "delete+insert", "merge", "insert_overwrite"]
-        if os.environ.get("DBT_EXPERIMENTAL_MICROBATCH"):
+        if self.behavior.require_builtin_microbatch_strategy:
             builtin_strategies.append("microbatch")
 
         return builtin_strategies