Repository: dbt-labs/dbt-core
Pull request: Fix missing sql_header for incremental models
resolves #2136

### Description
This PR adds support for injecting a `sql_header` above `merge` statements generated for incremental models.


### Checklist
 - [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
 - [x] I have run this code in development and it appears to resolve the stated issue
 - [x] This PR includes tests, or tests are not required/relevant for this PR
 - [x] I have updated the `CHANGELOG.md` and added information about my change to the "dbt next" section.


Problem statement:
set_sql_header works well with 'table', but not 'incremental'
### Describe the bug
This test fails when you change 'table' for 'incremental':

test/integration/022_bigquery_test/models/sql_header_model.sql

### Steps To Reproduce

1. Change 'table' to 'incremental' in sql_header_model.sql
2. `dbt run` works well the first time (creates the table)
3. `dbt run` fails the next time (when doing incremental)

### Expected behavior

The incremental run should bring the same headers - the model will again need the temporal function.

### Screenshots and log output

"Function not found: yes_no_to_boolean at [14:8]"

### System information
**Which database are you using dbt with?**
- [ ] postgres
- [ ] redshift
- [x ] bigquery
- [ ] snowflake
- [ ] other (specify: ____________)


**The output of `dbt --version`:**
0.15.2

Hints from issue discussion:
Thanks for the report @fhoffa - we'll check it out!

Commit message: (#2136) Fix missing sql_header for incremental models

Patch to review:
diff --git a/core/dbt/include/global_project/macros/materializations/common/merge.sql b/core/dbt/include/global_project/macros/materializations/common/merge.sql
index dcbcc1a356d..8098483207e 100644
--- a/core/dbt/include/global_project/macros/materializations/common/merge.sql
+++ b/core/dbt/include/global_project/macros/materializations/common/merge.sql
@@ -18,6 +18,7 @@
 {% macro default__get_merge_sql(target, source, unique_key, dest_columns, predicates) -%}
     {%- set predicates = [] if predicates is none else [] + predicates -%}
     {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+    {%- set sql_header = config.get('sql_header', none) -%}
 
     {% if unique_key %}
         {% set unique_key_match %}
@@ -28,6 +29,8 @@
         {% do predicates.append('FALSE') %}
     {% endif %}
 
+    {{ sql_header if sql_header is not none }}
+
     merge into {{ target }} as DBT_INTERNAL_DEST
         using {{ source }} as DBT_INTERNAL_SOURCE
         on {{ predicates | join(' and ') }}
@@ -87,6 +90,9 @@
 {% macro default__get_insert_overwrite_merge_sql(target, source, dest_columns, predicates) -%}
     {%- set predicates = [] if predicates is none else [] + predicates -%}
     {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+    {%- set sql_header = config.get('sql_header', none) -%}
+
+    {{ sql_header if sql_header is not none }}
 
     merge into {{ target }} as DBT_INTERNAL_DEST
         using {{ source }} as DBT_INTERNAL_SOURCE
diff --git a/plugins/snowflake/dbt/include/snowflake/macros/materializations/merge.sql b/plugins/snowflake/dbt/include/snowflake/macros/materializations/merge.sql
index e3a5d5cd085..0c48eb9493a 100644
--- a/plugins/snowflake/dbt/include/snowflake/macros/materializations/merge.sql
+++ b/plugins/snowflake/dbt/include/snowflake/macros/materializations/merge.sql
@@ -7,9 +7,12 @@
     #}
 
     {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute='name')) -%}
+    {%- set sql_header = config.get('sql_header', none) -%}
 
     {%- if unique_key is none -%}
 
+        {{ sql_header if sql_header is not none }}
+
         insert into {{ target }} ({{ dest_cols_csv }})
         (
             select {{ dest_cols_csv }}
diff --git a/test/integration/022_bigquery_test/models/sql_header_model_incr.sql b/test/integration/022_bigquery_test/models/sql_header_model_incr.sql
new file mode 100644
index 00000000000..f93280a3bfd
--- /dev/null
+++ b/test/integration/022_bigquery_test/models/sql_header_model_incr.sql
@@ -0,0 +1,15 @@
+
+{{ config(materialized="incremental") }}
+
+{# This will fail if it is not extracted correctly #}
+{% call set_sql_header(config) %}
+  	CREATE TEMPORARY FUNCTION a_to_b(str STRING)
+	RETURNS STRING AS (
+	  CASE
+	  WHEN LOWER(str) = 'a' THEN 'b'
+	  ELSE str
+	  END
+	);
+{% endcall %}
+
+select a_to_b(dupe) as dupe from {{ ref('view_model') }}
diff --git a/test/integration/022_bigquery_test/test_simple_bigquery_view.py b/test/integration/022_bigquery_test/test_simple_bigquery_view.py
index e25704ba1b4..f17ca9bcfcc 100644
--- a/test/integration/022_bigquery_test/test_simple_bigquery_view.py
+++ b/test/integration/022_bigquery_test/test_simple_bigquery_view.py
@@ -53,7 +53,7 @@ def test__bigquery_simple_run(self):
         self.run_dbt(['seed', '--full-refresh'])
         results = self.run_dbt()
         # Bump expected number of results when adding new model
-        self.assertEqual(len(results), 8)
+        self.assertEqual(len(results), 9)
         self.assert_nondupes_pass()
 
 
@@ -64,7 +64,7 @@ class TestUnderscoreBigQueryRun(TestBaseBigQueryRun):
     def test_bigquery_run_twice(self):
         self.run_dbt(['seed'])
         results = self.run_dbt()
-        self.assertEqual(len(results), 8)
+        self.assertEqual(len(results), 9)
         results = self.run_dbt()
-        self.assertEqual(len(results), 8)
+        self.assertEqual(len(results), 9)
         self.assert_nondupes_pass()


Code review comments:
diff_hunk: @@ -87,6 +90,9 @@
 {% macro default__get_insert_overwrite_merge_sql(target, source, dest_columns, predicates) -%}
     {%- set predicates = [] if predicates is none else [] + predicates -%}
     {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+    {%- set sql_header = config.get('sql_header', none) -%}
+
+    {{ sql_header if sql_header is not none }}
path: core/dbt/include/global_project/macros/materializations/common/merge.sql
text: @user1:
I think, if we do this, then the `sql_header` will run twice for "dynamic" (scripting) `insert_overwrite` materializations. (The first time will be when creating the temp table.) I assume that's ok?

@author:
This is a good call. Temporary UDFs can't be defined twice in the same BQ script. Think we're going to need to find an alternative approach here.

I kicked around the idea of making this a materialization-level thing, and while I think that might be a good approach long-term, I don't want to touch every dbt materialization in support of this change. Going to try to figure out a nice way to make this sensible for a very specific use case (insert_overwrite on BigQuery incremental models) and localize the jank as much as possible