Merge branch 'main' into bug/vector-dtype-parsing

.changes/unreleased/Features-20241202-095136.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support MicrobatchConcurrency
+time: 2024-12-02T09:51:36.606097-05:00
+custom:
+  Author: michelleark
+  Issue: "1260"

.changes/unreleased/Fixes-20241209-131530.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: AUTO should no longer lead to rebuilds of dynamic tables.
+time: 2024-12-09T13:15:30.554566-08:00
+custom:
+    Author: versusfacit
+    Issue: "1267"

.changes/unreleased/Under the Hood-20241205-144036.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Use new `batch` context variables over `node.config.__dbt_internal` ones
+time: 2024-12-05T14:40:36.373637-05:00
+custom:
+  Author: michelleark
+  Issue: "1263"

.changes/unreleased/Under the Hood-20241211-170831.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Update default Python version for python models
+time: 2024-12-11T17:08:31.842063-05:00
+custom:
+  Author: mikealfare
+  Issue: "1203"

dbt/adapters/snowflake/constants.py

@@ -0,0 +1 @@
+DEFAULT_PYTHON_VERSION_FOR_PYTHON_MODELS = "3.9"

dbt/adapters/snowflake/impl.py

@@ -22,6 +22,7 @@
 from dbt_common.exceptions import CompilationError, DbtDatabaseError, DbtRuntimeError
 from dbt_common.utils import filter_null_values
 
+from dbt.adapters.snowflake import constants
 from dbt.adapters.snowflake.relation_configs import (
     SnowflakeRelationType,
     TableFormat,
@@ -77,6 +78,7 @@ class SnowflakeAdapter(SQLAdapter):
             Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
             Capability.TableLastModifiedMetadataBatch: CapabilitySupport(support=Support.Full),
             Capability.GetCatalogForSingleRelation: CapabilitySupport(support=Support.Full),
+            Capability.MicrobatchConcurrency: CapabilitySupport(support=Support.Full),
         }
     )
 
@@ -344,7 +346,9 @@ def submit_python_job(self, parsed_model: dict, compiled_code: str):
         schema = parsed_model["schema"]
         database = parsed_model["database"]
         identifier = parsed_model["alias"]
-        python_version = parsed_model["config"].get("python_version", "3.8")
+        python_version = parsed_model["config"].get(
+            "python_version", constants.DEFAULT_PYTHON_VERSION_FOR_PYTHON_MODELS
+        )
 
         packages = parsed_model["config"].get("packages", [])
         imports = parsed_model["config"].get("imports", [])

dbt/adapters/snowflake/relation.py

@@ -17,6 +17,7 @@
 from dbt_common.events.functions import fire_event, warn_or_error
 
 from dbt.adapters.snowflake.relation_configs import (
+    RefreshMode,
     SnowflakeCatalogConfigChange,
     SnowflakeDynamicTableConfig,
     SnowflakeDynamicTableConfigChangeset,
@@ -109,7 +110,10 @@ def dynamic_table_config_changeset(
                 )
             )
 
-        if new_dynamic_table.refresh_mode != existing_dynamic_table.refresh_mode:
+        if (
+            new_dynamic_table.refresh_mode != RefreshMode.AUTO
+            and new_dynamic_table.refresh_mode != existing_dynamic_table.refresh_mode
+        ):
             config_change_collection.refresh_mode = SnowflakeDynamicTableRefreshModeConfigChange(
                 action=RelationConfigChangeAction.create,
                 context=new_dynamic_table.refresh_mode,

dbt/adapters/snowflake/relation_configs/__init__.py

@@ -3,6 +3,7 @@
     SnowflakeCatalogConfigChange,
 )
 from dbt.adapters.snowflake.relation_configs.dynamic_table import (
+    RefreshMode,
     SnowflakeDynamicTableConfig,
     SnowflakeDynamicTableConfigChangeset,
     SnowflakeDynamicTableRefreshModeConfigChange,

dbt/include/snowflake/macros/materializations/merge.sql

@@ -57,10 +57,10 @@
     {%- set incremental_predicates = [] if arg_dict.get('incremental_predicates') is none else arg_dict.get('incremental_predicates') -%}
 
     {#-- Add additional incremental_predicates to filter for batch --#}
-    {% if model.config.get("__dbt_internal_microbatch_event_time_start") -%}
+    {% if model.batch and model.batch.event_time_start -%}
         {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " >= to_timestamp_tz('" ~ model.config.__dbt_internal_microbatch_event_time_start ~ "')") %}
     {% endif %}
-    {% if model.config.__dbt_internal_microbatch_event_time_end -%}
+    {% if model.batch and model.batch.event_time_end -%}
         {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " < to_timestamp_tz('" ~ model.config.__dbt_internal_microbatch_event_time_end ~ "')") %}
     {% endif %}
     {% do arg_dict.update({'incremental_predicates': incremental_predicates}) %}

docker/Dockerfile

@@ -9,7 +9,7 @@ RUN apt-get update \
     build-essential=12.9 \
     ca-certificates=20210119 \
     git=1:2.30.2-1+deb11u2 \
-    libpq-dev=13.14-0+deb11u1 \
+    libpq-dev=13.18-0+deb11u1 \
     make=4.3-4.1 \
     openssh-client=1:8.4p1-5+deb11u3 \
     software-properties-common=0.96.20.2-2.1 \

setup.py

@@ -58,7 +58,7 @@ def _plugin_version() -> str:
     include_package_data=True,
     install_requires=[
         "dbt-common>=1.10,<2.0",
-        "dbt-adapters>=1.7,<2.0",
+        "dbt-adapters>=1.10.4,<2.0",
         "snowflake-connector-python[secure-local-storage]~=3.0",
         # add dbt-core to ensure backwards compatibility of installation, this is not a functional dependency
         "dbt-core>=1.8.0",

tests/functional/relation_tests/dynamic_table_tests/models.py

@@ -17,6 +17,26 @@
 """
 
 
+EXPLICIT_AUTO_DYNAMIC_TABLE = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+    refresh_mode='AUTO',
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+IMPLICIT_AUTO_DYNAMIC_TABLE = """
+{{ config(
+    materialized='dynamic_table',
+    snowflake_warehouse='DBT_TESTING',
+    target_lag='2 minutes',
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+
 DYNAMIC_TABLE_DOWNSTREAM = """
 {{ config(
     materialized='dynamic_table',

tests/functional/relation_tests/dynamic_table_tests/test_basic.py

@@ -1,6 +1,6 @@
 import pytest
 
-from dbt.tests.util import run_dbt
+from dbt.tests.util import assert_message_in_logs, run_dbt, run_dbt_and_capture
 
 from tests.functional.relation_tests.dynamic_table_tests import models
 from tests.functional.utils import query_relation_type
@@ -46,3 +46,44 @@ class TestBasicIcebergOn(TestBasic):
     @pytest.fixture(scope="class")
     def project_config_update(self):
         return {"flags": {"enable_iceberg_materializations": True}}
+
+
+class TestAutoConfigDoesntFullRefresh:
+    """
+    AUTO refresh_strategy will be compared accurately with both INCREMENTAL and FULL.
+    https://github.com/dbt-labs/dbt-snowflake/issues/1267
+    """
+
+    DT_NAME = "my_dynamic_table"
+
+    @pytest.fixture(scope="class", autouse=True)
+    def seeds(self):
+        return {"my_seed.csv": models.SEED}
+
+    @pytest.fixture(scope="class", autouse=True)
+    def models(self):
+        yield {
+            f"explicit_{self.DT_NAME}.sql": models.EXPLICIT_AUTO_DYNAMIC_TABLE,
+            f"implicit_{self.DT_NAME}.sql": models.IMPLICIT_AUTO_DYNAMIC_TABLE,
+        }
+
+    @pytest.mark.parametrize("test_dt", [f"explicit_{DT_NAME}", f"implicit_{DT_NAME}"])
+    def test_auto_config_doesnt_full_refresh(self, project, test_dt):
+        model_qualified_name = f"{project.database}.{project.test_schema}.{test_dt}"
+
+        run_dbt(["seed"])
+        _, logs = run_dbt_and_capture(["--debug", "run", "--select", f"{test_dt}.sql"])
+        assert_message_in_logs(f"create dynamic table {model_qualified_name}", logs)
+        assert_message_in_logs("refresh_mode = AUTO", logs)
+
+        _, logs = run_dbt_and_capture(["--debug", "run", "--select", f"{test_dt}.sql"])
+
+        assert_message_in_logs(f"create dynamic table {model_qualified_name}", logs, False)
+        assert_message_in_logs(
+            f"create or replace dynamic table {model_qualified_name}", logs, False
+        )
+        assert_message_in_logs("refresh_mode = AUTO", logs, False)
+        assert_message_in_logs(
+            f"No configuration changes were identified on: `{model_qualified_name}`. Continuing.",
+            logs,
+        )