diff --git a/CHANGELOG.md b/CHANGELOG.md index 9709ccffb..a801b9e64 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/dbt/include/databricks/macros/materializations/incremental/strategies.sql b/dbt/include/databricks/macros/materializations/incremental/strategies.sql index c1184b60d..cd2805096 100644 --- a/dbt/include/databricks/macros/materializations/incremental/strategies.sql +++ b/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 ') }} diff --git a/tests/functional/adapter/microbatch/fixtures.py b/tests/functional/adapter/microbatch/fixtures.py index 9a6dc900d..c09ca9bf3 100644 --- a/tests/functional/adapter/microbatch/fixtures.py +++ b/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 +""" diff --git a/tests/functional/adapter/microbatch/test_microbatch_insert_by_name.py b/tests/functional/adapter/microbatch/test_microbatch_insert_by_name.py new file mode 100644 index 000000000..8622c1762 --- /dev/null +++ b/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) diff --git a/tests/unit/macros/materializations/incremental/test_microbatch_macros.py b/tests/unit/macros/materializations/incremental/test_microbatch_macros.py index 390b9a868..183c2c1a0 100644 --- a/tests/unit/macros/materializations/incremental/test_microbatch_macros.py +++ b/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) diff --git a/tests/unit/macros/materializations/incremental/test_replace_where_macros.py b/tests/unit/macros/materializations/incremental/test_replace_where_macros.py index 1c984969f..54dcb7b6a 100644 --- a/tests/unit/macros/materializations/incremental/test_replace_where_macros.py +++ b/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 = {