fix: add insert-by-name support to microbatch and replace_where strategies (#1338)

CHANGELOG.md

@@ -1,3 +1,9 @@
+## dbt-databricks 1.11.7 (TBD)
+
+### Fixes
+
+- Fix column order mismatch in microbatch and replace_where incremental strategies by using INSERT BY NAME syntax ([#1338](https://github.com/databricks/dbt-databricks/issues/1338))
+
 ## dbt-databricks 1.11.6 (Mar 10, 2026)
 
 ### Features

dbt/include/databricks/macros/materializations/incremental/strategies.sql

@@ -121,7 +121,9 @@
   {%- set predicates = args_dict['incremental_predicates'] -%}
   {%- set target_relation = args_dict['target_relation'] -%}
   {%- set temp_relation = args_dict['temp_relation'] -%}
+  {%- set has_insert_by_name = adapter.has_dbr_capability('insert_by_name') -%}
 INSERT INTO {{ target_relation.render() }}
+{%- if has_insert_by_name %} BY NAME{% endif %}
 {%- if predicates %}
   {%- if predicates is sequence and predicates is not string %}
  REPLACE WHERE {{ predicates | join(' and ') }}

tests/functional/adapter/microbatch/fixtures.py

@@ -14,3 +14,45 @@
       - name: event_time
         description: "Timestamp of the event"
 """
+
+microbatch_seeds_csv = """
+id,event_time,amount
+1,2023-01-01,100
+2,2023-01-01,200
+3,2023-01-02,300
+""".strip()
+
+# Initial model: columns in (id, event_time, amount) order
+microbatch_model_sql = """
+{{ config(
+    materialized='incremental',
+    incremental_strategy='microbatch',
+    event_time='event_time',
+    begin='2023-01-01',
+    batch_size='day'
+) }}
+select id, event_time, amount from {{ ref('microbatch_seeds') }}
+"""
+
+# Reordered model: columns in (amount, id, event_time) order — this is the key scenario
+# Without BY NAME, positional INSERT would silently corrupt data here
+microbatch_model_reordered_sql = """
+{{ config(
+    materialized='incremental',
+    incremental_strategy='microbatch',
+    event_time='event_time',
+    begin='2023-01-01',
+    batch_size='day'
+) }}
+select amount, id, event_time from {{ ref('microbatch_seeds') }}
+"""
+
+schema_yml = """
+version: 2
+models:
+  - name: microbatch_model
+    columns:
+      - name: id
+      - name: event_time
+      - name: amount
+"""

tests/functional/adapter/microbatch/test_microbatch_insert_by_name.py

@@ -0,0 +1,100 @@
+import pytest
+from dbt.tests import util
+
+from tests.functional.adapter.microbatch.fixtures import (
+    microbatch_seeds_csv,
+    schema_yml,
+)
+
+# Uses replace_where strategy (same macro as microbatch) to verify that
+# column reordering between runs does not corrupt data.
+_initial_model_sql = """
+{{ config(
+    materialized='incremental',
+    incremental_strategy='replace_where',
+    incremental_predicates="id >= 2",
+) }}
+
+{% if not is_incremental() %}
+select id, event_time, amount from {{ ref('microbatch_seeds') }}
+{% else %}
+select id, event_time, amount from {{ ref('microbatch_seeds') }} where id >= 2
+{% endif %}
+"""
+
+# Same logic but columns in a different order (amount, id, event_time)
+_reordered_model_sql = """
+{{ config(
+    materialized='incremental',
+    incremental_strategy='replace_where',
+    incremental_predicates="id >= 2",
+) }}
+
+{% if not is_incremental() %}
+select amount, id, event_time from {{ ref('microbatch_seeds') }}
+{% else %}
+select amount, id, event_time from {{ ref('microbatch_seeds') }} where id >= 2
+{% endif %}
+"""
+
+
+class TestReplaceWhereColumnOrder:
+    """
+    Verifies that the replace_where strategy (shared with microbatch) preserves
+    data integrity when column order changes between the temp and target tables.
+
+    The fix uses INSERT BY NAME so that columns are matched by name, not position.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "replace_where_col_order.sql": _initial_model_sql,
+            "schema.yml": schema_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {
+            "microbatch_seeds.csv": microbatch_seeds_csv,
+        }
+
+    def test_replace_where_column_order(self, project):
+        # Seed and initial run — creates the table with original column order
+        util.run_dbt(["seed"])
+        util.run_dbt(["run"])
+
+        # Verify initial data
+        actual_initial = project.run_sql(
+            f"select id, amount from {project.database}.{project.test_schema}"
+            ".replace_where_col_order order by id",
+            fetch="all",
+        )
+        assert len(actual_initial) == 3
+        assert actual_initial[0] == (1, 100)
+        assert actual_initial[1] == (2, 200)
+        assert actual_initial[2] == (3, 300)
+
+        # Swap to the reordered model (amount, id, event_time)
+        util.write_file(
+            _reordered_model_sql,
+            str(project.project_root) + "/models/replace_where_col_order.sql",
+        )
+
+        # Incremental run with reordered columns
+        util.run_dbt(["run"])
+
+        # Verify data integrity — columns must match by name, not position
+        actual_final = project.run_sql(
+            f"select id, amount from {project.database}.{project.test_schema}"
+            ".replace_where_col_order order by id",
+            fetch="all",
+        )
+
+        assert len(actual_final) == 3
+        # id=1 was not replaced (predicate is id >= 2), so it stays the same
+        assert actual_final[0] == (1, 100)
+        # These rows were replaced — if column order was wrong, amount and id
+        # would be swapped (e.g., id=200 amount=2 instead of id=2 amount=200)
+        assert actual_final[1] == (2, 200)
+        assert actual_final[2] == (3, 300)

tests/unit/macros/materializations/incremental/test_microbatch_macros.py

@@ -14,6 +14,17 @@ def template_name(self) -> str:
     def macro_folders_to_load(self) -> list:
         return ["macros", "macros/materializations/incremental"]
 
+    @pytest.fixture(autouse=True)
+    def setup_mock_capability(self, context):
+        """Mock the adapter's has_dbr_capability to support insert_by_name by default"""
+
+        def has_dbr_capability_side_effect(capability_name):
+            if capability_name == "insert_by_name":
+                return True  # Default to DBR 12.2+
+            return False
+
+        context["adapter"].has_dbr_capability = Mock(side_effect=has_dbr_capability_side_effect)
+
     @pytest.fixture
     def mock_temp(self):
         relation = Mock()
@@ -50,8 +61,9 @@ def test_databricks__get_incremental_microbatch_sql_with_start_and_end(
 
         expected = """
         insert into `some_database`.`some_schema`.`some_table`
-        replace where cast(event_timestamp as TIMESTAMP) >= '2023-01-01 00:00:00'
-            and cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00' table `temp_table`
+        by name replace where cast(event_timestamp as TIMESTAMP) >= '2023-01-01 00:00:00'
+            and cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00'
+            table `temp_table`
 """
 
         self.assert_sql_equal(result, expected)
@@ -75,7 +87,8 @@ def test_databricks__get_incremental_microbatch_sql_with_start_only(
 
         expected = """
         insert into `some_database`.`some_schema`.`some_table`
-        replace where cast(event_timestamp as TIMESTAMP) >= '2023-01-01 00:00:00' table `temp_table`
+        by name replace where cast(event_timestamp as TIMESTAMP) >= '2023-01-01 00:00:00'
+            table `temp_table`
 """
 
         self.assert_sql_equal(result, expected)
@@ -99,7 +112,8 @@ def test_databricks__get_incremental_microbatch_sql_with_end_only(
 
         expected = """
         insert into `some_database`.`some_schema`.`some_table`
-        replace where cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00' table `temp_table`
+        by name replace where cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00'
+            table `temp_table`
 """
 
         self.assert_sql_equal(result, expected)
@@ -125,9 +139,10 @@ def test_databricks__get_incremental_microbatch_sql_with_existing_predicates(
 
         expected = """
         insert into `some_database`.`some_schema`.`some_table`
-        replace where col1 = 'value'
+        by name replace where col1 = 'value'
             and col2 > 100 and cast(event_timestamp as TIMESTAMP) >= '2023-01-01 00:00:00'
-            and cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00' table `temp_table`
+            and cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00'
+            table `temp_table`
 """
 
         self.assert_sql_equal(result, expected)
@@ -151,7 +166,37 @@ def test_databricks__get_incremental_microbatch_sql_without_start_or_end(
 
         expected = """
         insert into `some_database`.`some_schema`.`some_table`
-        table `temp_table`
+        by name table `temp_table`
+"""
+
+        self.assert_sql_equal(result, expected)
+
+    def test_databricks__get_incremental_microbatch_sql_without_insert_by_name(
+        self, template_bundle, context, config, mock_arg_dict
+    ):
+        """Test that BY NAME is omitted when the DBR version doesn't support insert_by_name"""
+
+        context["adapter"].has_dbr_capability = Mock(return_value=False)
+
+        context["model"] = {"config": {"event_time": "event_timestamp"}}
+        config["__dbt_internal_microbatch_event_time_start"] = "2023-01-01 00:00:00"
+        config["__dbt_internal_microbatch_event_time_end"] = "2023-01-02 00:00:00"
+
+        context["return"] = Mock()
+
+        self.run_macro_raw(
+            template_bundle.template,
+            "databricks__get_incremental_microbatch_sql",
+            mock_arg_dict,
+        )
+
+        result = context["return"].call_args[0][0]
+
+        expected = """
+        insert into `some_database`.`some_schema`.`some_table`
+        replace where cast(event_timestamp as TIMESTAMP) >= '2023-01-01 00:00:00'
+            and cast(event_timestamp as TIMESTAMP) < '2023-01-02 00:00:00'
+            table `temp_table`
 """
 
         self.assert_sql_equal(result, expected)

tests/unit/macros/materializations/incremental/test_replace_where_macros.py

@@ -14,6 +14,17 @@ def template_name(self) -> str:
     def macro_folders_to_load(self) -> list:
         return ["macros", "macros/materializations/incremental"]
 
+    @pytest.fixture(autouse=True)
+    def setup_mock_capability(self, context):
+        """Mock the adapter's has_dbr_capability to support insert_by_name by default"""
+
+        def has_dbr_capability_side_effect(capability_name):
+            if capability_name == "insert_by_name":
+                return True  # Default to DBR 12.2+
+            return False
+
+        context["adapter"].has_dbr_capability = Mock(side_effect=has_dbr_capability_side_effect)
+
     @pytest.fixture
     def mock_relations(self):
         target_relation = Mock()
@@ -37,7 +48,7 @@ def test_get_replace_where_sql_with_string_predicate(self, template_bundle, mock
 
         expected = """
         INSERT INTO schema.target_table
-        REPLACE WHERE date_col > '2023-01-01'
+        BY NAME REPLACE WHERE date_col > '2023-01-01'
         TABLE schema.temp_table
         """
 
@@ -59,7 +70,7 @@ def test_get_replace_where_sql_with_predicate_list(self, template_bundle, mock_r
 
         expected = """
         INSERT INTO schema.target_table
-        REPLACE WHERE date_col > '2023-01-01' and another_col != 'value'
+        BY NAME REPLACE WHERE date_col > '2023-01-01' and another_col != 'value'
         TABLE schema.temp_table
         """
 
@@ -78,7 +89,7 @@ def test_get_replace_where_sql_without_predicates(self, template_bundle, mock_re
 
         expected = """
         INSERT INTO schema.target_table
-        TABLE schema.temp_table
+        BY NAME TABLE schema.temp_table
         """
 
         self.assert_sql_equal(result, expected)
@@ -96,7 +107,7 @@ def test_get_replace_where_sql_with_empty_predicate_list(self, template_bundle,
 
         expected = """
         INSERT INTO schema.target_table
-        TABLE schema.temp_table
+        BY NAME TABLE schema.temp_table
         """
 
         self.assert_sql_equal(result, expected)
@@ -118,14 +129,18 @@ def test_get_replace_where_sql_with_complex_predicates(self, template_bundle, mo
 
         expected = """
         INSERT INTO schema.target_table
-        REPLACE WHERE date_col BETWEEN '2023-01-01' AND '2023-01-31' and status IN ('active', 'pending') and amount > 1000
+        BY NAME REPLACE WHERE date_col BETWEEN '2023-01-01' AND '2023-01-31' and status IN ('active', 'pending') and amount > 1000
         TABLE schema.temp_table
         """  # noqa
 
         self.assert_sql_equal(result, expected)
 
-    def test_get_replace_where_sql__by_name_with_predicates(self, template_bundle, mock_relations):
-        """Test that get_replace_where_sql does not use BY NAME (removed as of recent change)"""
+    def test_get_replace_where_sql__without_insert_by_name(
+        self, template_bundle, context, mock_relations
+    ):
+        """Test that BY NAME is omitted when the DBR version doesn't support insert_by_name"""
+        context["adapter"].has_dbr_capability = Mock(return_value=False)
+
         target_relation, temp_relation = mock_relations
 
         args_dict = {