Skip to content

perf: reuse columns from process_schema_changes in incremental materialization #1412

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
moomindani wants to merge 1 commit into
base: main
Choose a base branch
from
Open

perf: reuse columns from process_schema_changes in incremental materialization #1412

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,9 @@
## dbt-databricks next

### Under the Hood

- 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.

## dbt-databricks 1.11.7 (Apr 17, 2026)

### Features
Expand Down
26 changes: 20 additions & 6 deletions dbt/include/databricks/macros/materializations/incremental/incremental.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) }}
{% 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 -%}
Expand Down Expand Up @@ -139,13 +146,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' -%}
Expand Down Expand Up @@ -221,15 +235,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 %}
Expand Down
28 changes: 22 additions & 6 deletions dbt/include/databricks/macros/materializations/incremental/strategies.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) %}
Expand Down Expand Up @@ -138,7 +138,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 %}
Expand Down Expand Up @@ -219,10 +225,15 @@ 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. --#}
{%- if dest_columns is none -%}
{%- 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) %}
Expand Down Expand Up @@ -273,7 +284,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') -%}
Expand Down