Merge pull request #3 from AidanHarveyNelson/feature/improved-integration-tests

Cleaned up Materialization Code and helper code. Added and improved integration tests

Author: AidanHarveyNelson

.github/workflows/ci.yml

@@ -22,9 +22,9 @@ jobs:
       with:
         # snowflake
         SNOWFLAKE_USER: ${{ vars.SNOWFLAKE_USER }}
-        SNOWFLAKE_WAREHOUSE: ${{ vars.SNOWFLAKE_WAREHOUSE }}
         SNOWFLAKE_ROLE: ${{ vars.SNOWFLAKE_ROLE }}
         SNOWFLAKE_DATABASE: ${{ vars.SNOWFLAKE_DATABASE }}
+        SNOWFLAKE_WAREHOUSE: ${{ vars.SNOWFLAKE_WAREHOUSE }}
         SNOWFLAKE_SCHEMA: "integration_tests_snowflake_${{ github.run_number }}"
 
       secrets:

.pre-commit-config.yaml

@@ -0,0 +1,17 @@
+# Configuration for pre-commit hooks (see https://pre-commit.com/).
+
+# Force all unspecified python hooks to run python 3.12
+default_language_version:
+  python: python3.12
+
+repos:
+- repo: https://github.com/pre-commit/pre-commit-hooks
+  rev: v3.2.0
+  hooks:
+  - id: check-yaml
+    args: [--unsafe]
+  - id: end-of-file-fixer
+  - id: trailing-whitespace
+    exclude_types:
+      - "markdown"
+  - id: check-case-conflict

CHANGLOG.md

@@ -1,7 +1,17 @@
 # Changelog
 
-## v0.0.2 (TBA)
+## v0.0.2 (2026-09-03)
 
+- Feature: Rewrite stream_task.sql materialization with proper step ordering: temp view → target table → stream → task → resume task
+- Feature: Simplify stream.sql to always set src_table config and return the source relation for column discovery
+- Feature: Fix create_table.sql get_missing_columns argument order
+- Feature: Rewrite create_task.sql with clean MERGE formatting and proper parameter handling
+- Feature: Add 8 integration tests covering table/stream/task existence and data validation
+- Feature: Add prep_structure, insert_data, and cleanup helper macros
+- Feature: Update sample models to use correct config structure
+- Feature: Update run_test.sh with --full-refresh and clear step logging
+- Feature: Remove non-Snowflake adapter dependencies from pyproject.toml
+- Feature: Update README with usage guide and config reference
 - Feature: Expanded integration tests to cover all use cases
 
 

README.md

@@ -1,2 +1,126 @@
 # dbt-snowflake-streams-tasks
-A DBT Snowflake Plugin that enables users to create streams and tasks in DBT to build better versions of incremental models that don't need to be scheduled.
+
+A dbt package that enables creating Snowflake [streams](https://docs.snowflake.com/en/sql-reference/sql/create-stream) and [tasks](https://docs.snowflake.com/en/sql-reference/sql/create-task) through dbt, providing an alternative to incremental models that leverages Snowflake's native change data capture.
+
+## How It Works
+
+Instead of scheduling dbt runs to process incremental data, this package creates:
+
+1. **A target table** — with the schema defined by your SQL
+2. **A Snowflake stream** — to capture changes on your source table/view
+3. **A Snowflake task** — with a MERGE statement that automatically processes stream data into your target table
+
+Once deployed, the stream and task run autonomously in Snowflake — no dbt scheduling required.
+
+## Installation
+
+Add to your `packages.yml`:
+
+```yaml
+packages:
+  - git: "https://github.com/<your-org>/dbt-snowflake-streams-tasks.git"
+    revision: v0.0.1
+```
+
+## Usage
+
+### 1. Write Your Model SQL
+
+Use `stream_ref()` or `stream_source()` to specify the stream source:
+
+```sql
+-- models/my_model.sql
+select
+    id,
+    name,
+    email,
+    created_at
+from {{ dbt_streams_tasks.stream_ref('upstream_model') }}
+```
+
+Or from a source:
+
+```sql
+-- models/my_model.sql
+select *
+from {{ dbt_streams_tasks.stream_source('my_source', 'my_table') }}
+```
+
+### 2. Configure the Model
+
+In your YAML file:
+
+```yaml
+# models/_models.yml
+version: 2
+
+models:
+  - name: my_model
+    config:
+      materialized: stream_task
+      unique_key:
+        - id
+      stream:
+        name: "my_stream"
+        description: "Captures changes from the source"
+        append_only: true
+        show_initial_rows: true
+        copy_grants: false
+      task:
+        name: "my_task"
+        description: "Merges stream data into target"
+        warehouse: "MY_WAREHOUSE"
+        schedule: "1 MINUTE"
+        when: "SYSTEM$STREAM_HAS_DATA('my_stream')"
+```
+
+### 3. Run dbt
+
+```bash
+dbt run            # Creates table, stream, and task
+dbt run --full-refresh  # Recreates everything from scratch
+```
+
+## Configuration Reference
+
+### Stream Config
+
+| Parameter | Required | Default | Description |
+|-----------|----------|---------|-------------|
+| `name` | Yes | - | Name of the Snowflake stream |
+| `description` | No | `''` | Comment on the stream |
+| `append_only` | No | `true` | Only track inserts (not updates/deletes) |
+| `show_initial_rows` | No | `true` | Include existing rows on stream creation |
+| `copy_grants` | No | `false` | Copy grants when replacing the stream |
+
+### Task Config
+
+| Parameter | Required | Default | Description |
+|-----------|----------|---------|-------------|
+| `name` | Yes | - | Name of the Snowflake task |
+| `description` | No | `''` | Comment on the task |
+| `warehouse` | No* | - | Warehouse to execute the task (*required if not serverless) |
+| `schedule` | No | - | How often to run (e.g., `'1 MINUTE'`) |
+| `when` | No | - | Condition to check before running |
+| `allow_overlapping_execution` | No | - | Allow concurrent executions |
+| `user_task_timeout_ms` | No | - | Task timeout in milliseconds |
+| `suspend_task_after_num_failures` | No | - | Auto-suspend after N failures |
+| `task_auto_retry_attempts` | No | - | Number of auto-retry attempts |
+
+## Running Integration Tests
+
+```bash
+# Set up environment variables
+cp integration_tests/test.env.sample integration_tests/test.env
+cp integration_tests/vars.env.sample integration_tests/vars.env
+# Edit both files with your Snowflake credentials
+
+# Run tests
+./run_test.sh
+```
+
+## Requirements
+
+- dbt-core >= 1.9.1
+- dbt-snowflake >= 1.9.0
+- Snowflake account with CHANGE_TRACKING enabled on source tables

dbt_project.yml

@@ -1,5 +1,5 @@
-name: 'dbt_snowflake_streams_tasks'
-version: '0.0.1'
+name: 'dbt_streams_tasks'
+version: '0.0.2'
 config-version: 2
 require-dbt-version: [">=1.0.0", "<2.0.0"]
 macro-paths: ["macros"]

integration_tests/dbt_project.yml

@@ -1,5 +1,5 @@
 
-name: 'dbt_snowflake_streams_tasks_integration_tests'
+name: 'dbt_streams_tasks_integration_tests'
 version: '1.0'
 
 profile: 'integration_tests'
@@ -22,13 +22,10 @@ flags:
   use_colors: True
 
 models:
-  dbt_snowflake_streams_tasks_integration_tests:
+  dbt_streams_tasks_integration_tests:
     customers:
       +enabled: True
 
-
-
-tests:
-  dbt_snowflake_streams_tasks_integration_tests:
-    customers:
-      +enabled: True
+dispatch:
+  - macro_namespace: 'dbt_streams_tasks'
+    search_order: ['dbt_streams_tasks_integration_tests', 'dbt_streams_tasks']

integration_tests/macros/cleanup.sql

@@ -0,0 +1,22 @@
+{% macro cleanup() %}
+    {# Clean up all test objects created during integration tests #}
+
+    {% set objects_to_drop = [
+        "TASK transform_from_stream_test",
+        "TASK transform_from_ref_test",
+        "STREAM stream_from_source_test",
+        "STREAM stream_from_ref_test",
+        "TABLE " ~ [target.database, target.schema, 'STREAM_FROM_SOURCE']|join('.'),
+        "TABLE " ~ [target.database, target.schema, 'STREAM_FROM_REF']|join('.'),
+        "VIEW " ~ [target.database, target.schema, 'VIEW_ON_SOURCE']|join('.'),
+        "TABLE " ~ [target.database, target.schema, 'testing_source']|join('.'),
+        "TABLE " ~ [target.database, target.schema, 'testing_ref']|join('.')
+    ] %}
+
+    {% for obj in objects_to_drop %}
+        {% set drop_sql = "DROP " ~ obj ~ " IF EXISTS" %}
+        {% do log('Dropping: ' ~ obj, info=true) %}
+        {% do run_query(drop_sql) %}
+    {% endfor %}
+
+{% endmacro %}

integration_tests/macros/insert_data.sql

@@ -1,14 +1,17 @@
 {% macro insert_data() %}
 
-    {% set test_table  = [target.database, target.schema, 'testing_source']|join('.') %}
+    {% set tables = [
+        [target.database, target.schema, 'testing_source']|join('.'),
+        [target.database, target.schema, 'testing_ref']|join('.')
+    ] %}
 
-    {% set insert_data_sql %}
-        begin;
-        insert into {{test_table}} values (1, 'test data');
-        commit;
-    {% endset %}
+    {% for table in tables %}
+        {% set insert_data_sql %}
+            INSERT INTO {{ table }} VALUES (2, 'new stream data')
+        {% endset %}
 
-    {% do log('Inserting data into testing table ' ~ test_table, info = true) %}
-    {% do run_query(insert_data_sql) %}
+        {% do log('Inserting data into ' ~ table, info = true) %}
+        {% do run_query(insert_data_sql) %}
+    {% endfor %}
 
 {% endmacro %}

integration_tests/macros/prep_structure.sql

@@ -1,42 +1,36 @@
 {% macro prep_structure() %}
 
-    {% set create_schema_sql = "CREATE SCHEMA IF NOT EXISTS " ~ target.schema %}
+    {% set create_schema_sql = "CREATE SCHEMA IF NOT EXISTS " ~ target.database ~ "." ~ target.schema %}
 
     {% do log('Creating schema ' ~ target.schema, info = true) %}
     {% do run_query(create_schema_sql) %}
 
-    {% set test_table  = [
+    {% set test_tables  = [
         [target.database, target.schema, 'testing_source']|join('.'),
         [target.database, target.schema, 'testing_ref']|join('.'),
     ] %}
 
-    {% for table in test_table %}
+    {% for table in test_tables %}
         {% set create_testing_table_sql %}
-            begin;
-            create or replace table {{ table }} (
-                id integer,
-                data string
+            CREATE OR REPLACE TABLE {{ table }} (
+                id INTEGER,
+                data STRING
             )
             CHANGE_TRACKING = TRUE
-            ;
-            commit;
         {% endset %}
 
         {% do log('Creating testing table ' ~ table, info = true) %}
         {% do run_query(create_testing_table_sql) %}
     {% endfor %}
 
+    {# Insert initial data into both source tables #}
+    {% for table in test_tables %}
+        {% set insert_data_sql %}
+            INSERT INTO {{ table }} VALUES (1, 'initial data')
+        {% endset %}
 
-    {% set test_table  = [target.database, target.schema, 'testing_ref']|join('.') %}
-
-    {% set insert_data_sql %}
-        begin;
-        insert into {{test_table}} values (1, 'test data');
-        commit;
-    {% endset %}
-
-    {% do log('Inserting data into testing table ' ~ test_table, info = true) %}
-    {% do run_query(insert_data_sql) %}
-
+        {% do log('Inserting initial data into ' ~ table, info = true) %}
+        {% do run_query(insert_data_sql) %}
+    {% endfor %}
 
 {% endmacro %}

integration_tests/models/customers/_models.yml

@@ -8,16 +8,16 @@ models:
         - id
       stream:
         name: "stream_from_source_test"
-        description: "Stream for the customers table"
-        tags:
-          - customers
-        copy_grants: False
-        append_only: True
+        description: "Stream for the customers source table"
+        copy_grants: false
+        append_only: true
+        show_initial_rows: true
       task:
         name: "transform_from_stream_test"
-        description: "Task for the customers table"
+        description: "Task to merge stream data into stream_from_source"
+        warehouse: "{{ env_var('SNOWFLAKE_WAREHOUSE', 'COMPUTE_WH') }}"
         when: "SYSTEM$STREAM_HAS_DATA('stream_from_source_test')"
-        schedule: "1 MINUTES"
+        schedule: "1 MINUTE"
     columns:
       - name: id
 
@@ -28,13 +28,13 @@ models:
         - id
       stream:
         name: "stream_from_ref_test"
-        description: "Stream for the customers table"
-        tags:
-          - customers
-        copy_grants: True
-        append_only: False
+        description: "Stream for the view_on_source view"
+        copy_grants: false
+        append_only: true
+        show_initial_rows: true
       task:
         name: "transform_from_ref_test"
-        description: "Task for the customers table"
+        description: "Task to merge stream data into stream_from_ref"
+        warehouse: "{{ env_var('SNOWFLAKE_WAREHOUSE', 'COMPUTE_WH') }}"
         when: "SYSTEM$STREAM_HAS_DATA('stream_from_ref_test')"
-        schedule: "1 MINUTES"
+        schedule: "1 MINUTE"