Fix/incremental column precision changes (#5395)

* Only consider schema change when column cannot be expanded

* Add test for column shortening

* Add changelog entry

* Move test from integration to adapter tests

* Remove print statement

* add on_schema_change

.changes/unreleased/Fixes-20220617-193731.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: 'Resolves #5351 - Do not consider shorter varchar cols as schema changes'
+time: 2022-06-17T19:37:31.885484-07:00
+custom:
+  Author: epapineau
+  Issue: "5351"
+  PR: "5395"

core/dbt/include/global_project/macros/materializations/models/incremental/column_helpers.sql

@@ -41,7 +41,7 @@
   {% for sc in source_columns %}
     {% set tc = target_columns | selectattr("name", "equalto", sc.name) | list | first %}
     {% if tc %}
-      {% if sc.data_type != tc.data_type %}
+      {% if sc.data_type != tc.data_type and not sc.can_expand_to(other_column=tc) %}
         {{ result.append( { 'column_name': tc.name, 'new_type': sc.data_type } ) }}
       {% endif %}
     {% endif %}

tests/adapter/dbt/tests/adapter/basic/files.py

@@ -198,6 +198,17 @@
   select * from {{ ref('ephemeral') }}
 """
 
+incremental_not_schema_change_sql = """
+{{ config(materialized="incremental", unique_key="user_id_current_time",on_schema_change="sync_all_columns") }}
+select
+    1 || '-' || current_timestamp as user_id_current_time,
+    {% if is_incremental() %}
+        'thisis18characters' as platform
+    {% else %}
+        'okthisis20characters' as platform
+    {% endif %}
+"""
+
 base_materialized_var_sql = config_materialized_var + model_base
 base_table_sql = config_materialized_table + model_base
 base_view_sql = config_materialized_view + model_base

tests/adapter/dbt/tests/adapter/basic/test_incremental.py

@@ -1,10 +1,12 @@
 import pytest
 from dbt.tests.util import run_dbt, check_relations_equal, relation_from_name
+from dbt.contracts.results import RunStatus
 from dbt.tests.adapter.basic.files import (
     seeds_base_csv,
     seeds_added_csv,
     schema_base_yml,
     incremental_sql,
+    incremental_not_schema_change_sql,
 )
 
 
@@ -58,5 +60,28 @@ def test_incremental(self, project):
         assert len(catalog.sources) == 1
 
 
+class BaseIncrementalNotSchemaChange:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"name": "incremental"}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"incremental_not_schema_change.sql": incremental_not_schema_change_sql}
+
+    def test_incremental_not_schema_change(self, project):
+        # Schema change is not evaluated on first run, so two are needed
+        run_dbt(["run", "--select", "incremental_not_schema_change"])
+        run_result = (
+            run_dbt(["run", "--select", "incremental_not_schema_change"]).results[0].status
+        )
+
+        assert run_result == RunStatus.Success
+
+
 class Testincremental(BaseIncremental):
     pass
+
+
+class TestBaseIncrementalNotSchemaChange(BaseIncrementalNotSchemaChange):
+    pass