perf: reuse columns from process_schema_changes in incremental materialization

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### Under the Hood
 
 - Document serverless environment configuration for Python models (thanks @TangoEnSkai!) ([#1649](https://github.com/databricks/dbt-databricks/pull/1649) resolves [#1055](https://github.com/databricks/dbt-databricks/issues/1055))
+- Reuse the columns returned by `process_schema_changes` in the incremental materialization so the downstream strategy macros (`merge`, `append`, `delete+insert`) no longer re-issue `DESCRIBE TABLE EXTENDED` on the target. Saves one metadata round-trip per incremental model. Mirrors the existing `dbt-snowflake` pattern. ([#1412](https://github.com/databricks/dbt-databricks/pull/1412) resolves [#1411](https://github.com/databricks/dbt-databricks/issues/1411))
 
 ## dbt-databricks 1.12.5 (Sep 1, 2026)
 

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

@@ -58,9 +58,16 @@
         {{ set_overwrite_mode('DYNAMIC') }}
       {%- endif -%}
       {#-- Relation must be merged --#}
-      {%- do process_schema_changes(on_schema_change, intermediate_relation, existing_relation) -%}
+      {#-- Reuse the columns returned by `process_schema_changes` so the downstream
+           merge strategy macro doesn't have to re-issue DESCRIBE on the target.
+           When `on_schema_change == 'ignore'`, the macro returns `{}` and we fall
+           back to a single DESCRIBE on the existing relation. --#}
+      {%- set dest_columns = process_schema_changes(on_schema_change, intermediate_relation, existing_relation) -%}
+      {%- if not dest_columns -%}
+        {%- set dest_columns = adapter.get_columns_in_relation(existing_relation) -%}
+      {%- endif -%}
       {{ process_config_changes(target_relation, existing_relation) }}
-      {% set build_sql = get_build_sql(incremental_strategy, target_relation, intermediate_relation) %}
+      {% set build_sql = get_build_sql(incremental_strategy, target_relation, intermediate_relation, dest_columns) %}
       {%- if language == 'sql' -%}
         {#-- Check if build_sql is a list (multi-statement strategy) or a string (single statement) --#}
         {%- if build_sql is sequence and build_sql is not string -%}
@@ -150,13 +157,20 @@
       {%- call statement('create_temp_relation', language=language) -%}
         {{ create_table_as(True, temp_relation, compiled_code, language) }}
       {%- endcall -%}
-      {%- do process_schema_changes(on_schema_change, temp_relation, existing_relation) -%}
+      {#-- Reuse the columns returned by `process_schema_changes` so the downstream
+           merge strategy macro doesn't have to re-issue DESCRIBE on the target.
+           When `on_schema_change == 'ignore'`, the macro returns `{}` and we fall
+           back to a single DESCRIBE on the existing relation. --#}
+      {%- set dest_columns = process_schema_changes(on_schema_change, temp_relation, existing_relation) -%}
+      {%- if not dest_columns -%}
+        {%- set dest_columns = adapter.get_columns_in_relation(existing_relation) -%}
+      {%- endif -%}
       {%- set strategy_sql_macro_func = adapter.get_incremental_strategy_macro(context, incremental_strategy) -%}
       {%- set strategy_arg_dict = ({
               'target_relation': target_relation,
               'temp_relation': temp_relation,
               'unique_key': unique_key,
-              'dest_columns': none,
+              'dest_columns': dest_columns,
               'incremental_predicates': incremental_predicates}) -%}
       {%- set build_sql = strategy_sql_macro_func(strategy_arg_dict) -%}
       {%- if language == 'sql' -%}
@@ -240,15 +254,15 @@
   {% endif %}
 {% endmacro %}
 
-{% macro get_build_sql(incremental_strategy, target_relation, intermediate_relation) %}
+{% macro get_build_sql(incremental_strategy, target_relation, intermediate_relation, dest_columns=none) %}
   {%- set unique_key = config.get('unique_key') -%}
   {%- set incremental_predicates = config.get('predicates') or config.get('incremental_predicates') -%}
   {%- set strategy_sql_macro_func = adapter.get_incremental_strategy_macro(context, incremental_strategy) -%}
   {%- set strategy_arg_dict = ({
           'target_relation': target_relation,
           'temp_relation': intermediate_relation,
           'unique_key': unique_key,
-          'dest_columns': none,
+          'dest_columns': dest_columns,
           'incremental_predicates': incremental_predicates}) -%}
   {% do return(strategy_sql_macro_func(strategy_arg_dict)) %}
 {% endmacro %}

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

@@ -8,7 +8,7 @@
 {% endmacro %}
 
 {% macro databricks__get_incremental_append_sql(arg_dict) %}
-  {% do return(get_insert_into_sql(arg_dict["temp_relation"], arg_dict["target_relation"])) %}
+  {% do return(get_insert_into_sql(arg_dict["temp_relation"], arg_dict["target_relation"], arg_dict.get("dest_columns"))) %}
 {% endmacro %}
 
 {% macro databricks__get_incremental_replace_where_sql(arg_dict) %}
@@ -139,7 +139,13 @@ INSERT INTO {{ target_relation.render() }}
   {%- set source_relation = arg_dict.get('temp_relation') -%}
   {%- set target_relation = arg_dict.get('target_relation') -%}
   {%- set incremental_predicates = config.get('incremental_predicates') -%}
-  {%- set target_columns = (adapter.get_columns_in_relation(target_relation) | map(attribute='quoted') | list) -%}
+  {#-- Reuse dest_columns from the materialization (obtained via `process_schema_changes`)
+       when provided, otherwise fall back to a fresh DESCRIBE. --#}
+  {%- set dest_columns = arg_dict.get('dest_columns') -%}
+  {%- if dest_columns is none -%}
+    {%- set dest_columns = adapter.get_columns_in_relation(target_relation) -%}
+  {%- endif -%}
+  {%- set target_columns = (dest_columns | map(attribute='quoted') | list) -%}
   {%- set unique_key = config.require('unique_key') -%}
   {% do return(delete_insert_sql_impl(source_relation, target_relation, target_columns, unique_key, incremental_predicates)) %}
 {% endmacro %}
@@ -229,10 +235,20 @@ where {{ incremental_predicates }}
 {% endmacro %}
 
 
-{% macro get_insert_into_sql(source_relation, target_relation) %}
+{% macro get_insert_into_sql(source_relation, target_relation, dest_columns=none) %}
     {%- set source_columns = adapter.get_columns_in_relation(source_relation) | map(attribute="name") | list -%}
-    {%- set dest_columns = adapter.get_columns_in_relation(target_relation) | map(attribute="name") | list -%}
-    {{ insert_into_sql_impl(target_relation, dest_columns, source_relation, source_columns) }}
+    {#-- Reuse dest_columns from the materialization when provided; otherwise DESCRIBE.
+         `dest_columns` comes from `process_schema_changes`, which returns the *source*
+         columns, so it can omit a column the target still has (`on_schema_change:
+         append_new_columns`). That subset is only safe where the emitted statement
+         matches columns by name; without `insert_by_name` the matching-sets branch
+         degrades to a positional `select *`, so pay for the DESCRIBE instead. --#}
+    {%- if dest_columns is none or not adapter.has_dbr_capability('insert_by_name') -%}
+      {%- set dest_cols_list = adapter.get_columns_in_relation(target_relation) | map(attribute="name") | list -%}
+    {%- else -%}
+      {%- set dest_cols_list = dest_columns | map(attribute="name") | list -%}
+    {%- endif -%}
+    {{ insert_into_sql_impl(target_relation, dest_cols_list, source_relation, source_columns) }}
 {% endmacro %}
 
 {% macro insert_into_sql_impl(target_relation, dest_columns, source_relation, source_columns) %}
@@ -283,7 +299,12 @@ where {{ incremental_predicates }}
   {%- set source_alias = config.get('source_alias', 'DBT_INTERNAL_SOURCE') -%}
 
   {%- set predicates = [] if incremental_predicates is none else [] + incremental_predicates -%}
-  {%- set dest_columns = adapter.get_columns_in_relation(target) -%}
+  {#-- Prefer the `dest_columns` passed in by the materialization (obtained via
+       `process_schema_changes` or a single DESCRIBE on the existing relation).
+       Only issue a fresh DESCRIBE when no columns were provided. --#}
+  {%- if dest_columns is none -%}
+    {%- set dest_columns = adapter.get_columns_in_relation(target) -%}
+  {%- endif -%}
   {%- set source_columns = (adapter.get_columns_in_relation(source) | map(attribute='name') | list)-%}
   {%- set merge_update_columns = config.get('merge_update_columns') -%}
   {%- set merge_exclude_columns = config.get('merge_exclude_columns') -%}

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

@@ -1,4 +1,4 @@
-from unittest.mock import Mock
+from unittest.mock import MagicMock, Mock
 
 import pytest
 
@@ -262,3 +262,86 @@ def test_insert_into_sql_impl__case_insensitive_matching(self, template_bundle,
         assert "insert into" in clean_result
         assert "by name" in clean_result
         assert "select * from" in clean_result
+
+    def _column(self, name):
+        column = Mock()
+        column.name = name
+        return column
+
+    def _named_relation(self, rendered):
+        relation = MagicMock()
+        relation.__str__.return_value = rendered
+        relation.render.return_value = rendered
+        return relation
+
+    def test_get_insert_into_sql__no_dest_columns__describes_target(self, template_bundle):
+        """Without dest_columns the target is described, so a target-only column narrows
+        the INSERT to the intersection of target and source."""
+        source_relation = self._named_relation("source_table")
+        target_relation = self._named_relation("target_table")
+        template_bundle.context["adapter"].has_dbr_capability = Mock(return_value=True)
+        template_bundle.context["adapter"].get_columns_in_relation = Mock(
+            side_effect=lambda relation: (
+                [self._column("id"), self._column("name")]
+                if relation is source_relation
+                else [self._column("id"), self._column("name"), self._column("target_only")]
+            )
+        )
+
+        result = self.run_macro(
+            template_bundle.template,
+            "get_insert_into_sql",
+            source_relation,
+            target_relation,
+        )
+
+        expected = "insert into target_table (`id`, `name`) select `id`, `name` from source_table"
+        self.assert_sql_equal(result, expected)
+
+    def test_get_insert_into_sql__dest_columns_from_source__uses_by_name(self, template_bundle):
+        """`process_schema_changes` returns the *source* columns, so reusing them as
+        dest_columns makes the two sets match by construction and selects BY NAME even
+        though the target still holds a column the source does not have."""
+        source_relation = self._named_relation("source_table")
+        target_relation = self._named_relation("target_table")
+        template_bundle.context["adapter"].has_dbr_capability = Mock(return_value=True)
+        template_bundle.context["adapter"].get_columns_in_relation = Mock(
+            return_value=[self._column("id"), self._column("name")]
+        )
+
+        result = self.run_macro(
+            template_bundle.template,
+            "get_insert_into_sql",
+            source_relation,
+            target_relation,
+            [self._column("id"), self._column("name")],
+        )
+
+        expected = "insert into target_table by name select * from source_table"
+        self.assert_sql_equal(result, expected)
+
+    def test_get_insert_into_sql__dest_columns_ignored_without_by_name(self, template_bundle):
+        """DBR < 12.2 has no BY NAME, and the matching-sets branch there degrades to a
+        positional `select *`. dest_columns can omit a column the target still has, so the
+        target is described instead and the explicit column list is kept."""
+        source_relation = self._named_relation("source_table")
+        target_relation = self._named_relation("target_table")
+        template_bundle.context["adapter"].has_dbr_capability = Mock(return_value=False)
+        template_bundle.context["adapter"].get_columns_in_relation = Mock(
+            side_effect=lambda relation: (
+                [self._column("id"), self._column("name")]
+                if relation is source_relation
+                else [self._column("id"), self._column("name"), self._column("target_only")]
+            )
+        )
+
+        result = self.run_macro(
+            template_bundle.template,
+            "get_insert_into_sql",
+            source_relation,
+            target_relation,
+            [self._column("id"), self._column("name")],
+        )
+
+        expected = "insert into target_table (`id`, `name`) select `id`, `name` from source_table"
+        self.assert_sql_equal(result, expected)