Gate MicrobatchConcurrency behind use_concurrent_microbatch behavior flag

Makes concurrent microbatch opt-in (default=False) so it can ship safely
in a patch release. Users enable it via `flags: {use_concurrent_microbatch: true}`
in dbt_project.yml. The default can be flipped to True in 1.12.

Overrides supports() as an instance method rather than adding to the
class-level _capabilities dict, ensuring the behavior flag is always
checked regardless of how supports() is called.

Author: sd-db

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### Features
 - Add `query_id` to `SQLQueryStatus` events to improve query tracing and debugging
 - Add support for Row Filters ([#1294](https://github.com/databricks/dbt-databricks/pull/1294))
+- Enable concurrent microbatch execution via `use_concurrent_microbatch` behavior flag ([#914](https://github.com/databricks/dbt-databricks/issues/914))
 
 ### Fixes
 - Fix `hard_deletes: invalidate` incorrectly invalidating active records in snapshots (thanks @Zurbste!) ([#1281](https://github.com/databricks/dbt-databricks/issues/1281)) 

dbt/adapters/databricks/impl.py

@@ -141,6 +141,16 @@
     ),
 )  # type: ignore[typeddict-item]
 
+USE_CONCURRENT_MICROBATCH = BehaviorFlag(
+    name="use_concurrent_microbatch",
+    default=False,
+    description=(
+        "Enable concurrent execution of microbatch incremental batches. "
+        "When enabled, dbt will run microbatch batches in parallel threads "
+        "instead of sequentially. Requires dbt-core 1.9+."
+    ),
+)  # type: ignore[typeddict-item]
+
 
 class DatabricksRelationInfo(NamedTuple):
     table_name: str
@@ -246,8 +256,14 @@ def _behavior_flags(self) -> list[BehaviorFlag]:
             USE_MATERIALIZATION_V2,
             USE_REPLACE_ON_FOR_INSERT_OVERWRITE,
             USE_MANAGED_ICEBERG,
+            USE_CONCURRENT_MICROBATCH,
         ]
 
+    def supports(self, capability: Capability) -> bool:
+        if capability == Capability.MicrobatchConcurrency:
+            return bool(self.behavior.use_concurrent_microbatch)
+        return super().supports(capability)
+
     def quote(self, identifier):  # type: ignore[override,no-untyped-def]
         """Override base adapter's quote method to prevent double quoting."""
         return quote(identifier)

tests/unit/test_adapter_capabilities.py

@@ -136,6 +136,41 @@ def test_capability_caching_in_adapter(self, adapter):
             # Connection should have been asked twice (adapter doesn't cache)
             assert mock_conn.has_capability.call_count == 2
 
+    def test_microbatch_concurrency_disabled_by_default(self, adapter):
+        """Test that MicrobatchConcurrency is disabled by default (behavior flag off)."""
+        from dbt.adapters.capability import Capability
+
+        assert not adapter.supports(Capability.MicrobatchConcurrency)
+
+    def test_microbatch_concurrency_enabled_with_flag(self):
+        """Test that MicrobatchConcurrency is enabled when behavior flag is on."""
+        from dbt.adapters.capability import Capability
+
+        project_cfg = {
+            "name": "test_project",
+            "version": "0.1",
+            "profile": "test",
+            "project-root": "/tmp/dbt/does-not-exist",
+            "config-version": 2,
+            "flags": {"use_concurrent_microbatch": True},
+        }
+        profile_cfg = {
+            "outputs": {
+                "test": {
+                    "type": "databricks",
+                    "catalog": "main",
+                    "schema": "analytics",
+                    "host": "test.databricks.com",
+                    "http_path": "sql/protocolv1/o/1234567890123456/1234-567890-test123",
+                    "token": "dapi" + "X" * 32,
+                }
+            },
+            "target": "test",
+        }
+        config = config_from_parts_or_dicts(project_cfg, profile_cfg)
+        adapter = DatabricksAdapter(config, get_context("spawn"))
+        assert adapter.supports(Capability.MicrobatchConcurrency)
+
     def test_insert_by_name_capability_spec(self):
         """Test INSERT_BY_NAME capability specification and requirements."""
         # Verify the capability exists