Skip to content

Commit e851f86

Browse files
AidanHarveyNelsonAidanHarveyNelson
committed
Initial commmit with minimal testing and happy path confirmed to be working
1 parent 16c0ded commit e851f86

14 files changed

Lines changed: 339 additions & 51 deletions

File tree

‎integration_tests/macros/insert_data.sql‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{% macro insert_data() %}
22

3-
{% set test_table = [target.database, target.schema, 'testing']|join('.') %}
3+
{% set test_table = [target.database, target.schema, 'testing_source']|join('.') %}
44

55
{% set insert_data_sql %}
66
begin;

‎integration_tests/macros/prep_structure.sql‎

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,27 @@
11
{% macro prep_structure() %}
22

3-
{% set test_table = [target.database, target.schema, 'testing']|join('.') %}
4-
53
{% set create_schema_sql = "CREATE SCHEMA IF NOT EXISTS " ~ target.schema %}
64

7-
{% set create_testing_table_sql %}
8-
begin;
9-
create or replace table {{ test_table }} (
10-
id integer,
11-
data string
12-
);
13-
commit;
14-
{% endset %}
15-
165
{% do log('Creating schema ' ~ target.schema, info = true) %}
176
{% do run_query(create_schema_sql) %}
187

19-
{% do log('Creating testing table ' ~ test_table, info = true) %}
20-
{% do run_query(create_testing_table_sql) %}
8+
{% set test_table = [
9+
[target.database, target.schema, 'testing_source']|join('.'),
10+
[target.database, target.schema, 'testing_ref']|join('.'),
11+
] %}
12+
13+
{% for table in test_table %}
14+
{% set create_testing_table_sql %}
15+
begin;
16+
create or replace table {{ table }} (
17+
id integer,
18+
data string
19+
);
20+
commit;
21+
{% endset %}
22+
23+
{% do log('Creating testing table ' ~ table, info = true) %}
24+
{% do run_query(create_testing_table_sql) %}
25+
{% endfor %}
2126

2227
{% endmacro %}

‎integration_tests/models/customers/_models.yml‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,22 @@
11
version: 2
22

33
models:
4-
- name: transformed_testing
4+
- name: stream_from_source
5+
config:
6+
materialized: stream_task
7+
unique_key:
8+
- id
9+
stream:
10+
name: "stream_from_source_test"
11+
description: "Stream for the customers table"
12+
tags:
13+
- customers
14+
copy_grants: False
15+
append_only: True
16+
task:
17+
name: "transform_from_stream_test"
18+
description: "Task for the customers table"
19+
when: "SYSTEM$STREAM_HAS_DATA('stream_from_source_test')"
20+
schedule: "5 MINUTES"
21+
columns:
22+
- name: id

‎integration_tests/models/customers/_sources.yml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,4 +5,5 @@ sources:
55
database: "{{ target.database }}"
66
schema: "{{ target.schema }}"
77
tables:
8-
- name: testing
8+
- name: testing_source
9+
- name: testing_ref

‎integration_tests/models/customers/customers.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
select 1 as test

‎integration_tests/models/customers/stream_from_source.sql‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
select
2+
*,
3+
1 as new_column
4+
from {{ dbt_snowflake_streams_tasks.stream_source('customers', 'testing_source') }}

‎macros/helpers/create_stream.sql‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
{%- macro create_stream(source_table) -%}
2+
{%- do log('Source Table is: ' ~ source_table, info = true) -%}
3+
{%- set name = config.require('stream').name -%}
4+
{%- do log('Creating Stream: ' ~ name, info = true) -%}
5+
{%- set args_sql -%}
6+
APPEND_ONLY = {{ config.get('stream').append_only|upper or 'TRUE' }}
7+
COMMENT = '{{ config.get('stream').description or '' }}'
8+
{{- "COPY GRANTS" ~ config.get('stream').copy_grants if config.get('stream').copy_grants else '' -}}
9+
{%- endset -%}
10+
{%- set run_sql -%}
11+
CREATE OR REPLACE STREAM {{ name }} ON TABLE {{ source_table }}
12+
{{ args_sql }}
13+
{%- endset -%}
14+
{{ run_sql }}
15+
{%- endmacro -%}

‎macros/helpers/create_table.sql‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
{%- macro create_table(target_table, sql, tmp_relation, is_new) -%}
2+
{%- do log('Target Table is: ' ~ target_table, info = true) -%}
3+
{%- set run_sql -%}
4+
{%- if not is_new -%}
5+
{% set missing_columns = adapter.get_missing_columns(target_table, tmp_relation) %}
6+
{% do log('Missing columns are ' ~ missing_columns, info=true)%}
7+
{% if missing_columns | length > 0 %}
8+
alter table {{ target_table }}
9+
{%- for col in adapter.get_missing_columns(target_table, tmp_relation) -%}
10+
add column "{{- col.name -}}" {{- col.data_type -}}{{- ";" if loop.last else "\n" -}}
11+
{%- endfor -%}
12+
{% endif%}
13+
{%- else -%}
14+
create table {{ target_table }} as {{ sql }};
15+
{%- endif -%}
16+
{%- endset -%}
17+
{{ run_sql }}
18+
{%- endmacro -%}

‎macros/helpers/create_task.sql‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
{%- macro create_task(stream, target_table, sql, tmp_relation, unique_key) -%}
2+
{%- do log('Target Table is: ' ~ target_table, info = true) -%}
3+
{%- set name = config.require('task').name -%}
4+
{%- do log('Creating Task: ' ~ name, info = true) -%}
5+
{# This SQL has formatting issues think of a better way to populated and validate the config#}
6+
{%- set args_sql -%}
7+
COMMENT = '{{ config.get('stream').description or '' }}'
8+
{{ "WAREHOUSE = " ~ config.get('task').warehouse|upper if config.get('task').warehouse|upper else '' }}
9+
{{ "SCHEDULE = '" ~ config.get('task').schedule|upper ~ "'" if config.get('task').schedule|upper else '' -}}
10+
{{ "EXECUTE AS USER = '" ~ config.get('task').execute_as_user|upper ~ "'" if config.get('task').execute_as_user|upper else '' -}}
11+
{{ "ALLOW_OVERLAPPING_EXECUTION = " ~ config.get('task').allow_overlapping_execution|upper if config.get('task').allow_overlapping_execution else '' -}}
12+
{{ "USER_TASK_TIMEOUT_MS = " ~ config.get('task').user_task_timeout_ms if config.get('task').user_task_timeout_ms else '' -}}
13+
{{ "SUSPEND_TASK_AFTER_NUM_FAILURES = " ~ config.get('task').suspend_task_after_num_failures if config.get('task').suspend_task_after_num_failures else '' -}}
14+
{{ "ERROR_INTEGRATION = '" ~ config.get('task').error_integration|upper ~ "'" if config.get('task').error_integration|upper else '' -}}
15+
{{ "SUCCESS_INTEGRATION = '" ~ config.get('task').success_integration|upper ~ "'" if config.get('task').success_integration|upper else '' -}}
16+
{{ "LOG_LEVEL = '" ~ config.get('task').log_level|upper ~ "'" if config.get('task').log_level|upper else '' -}}
17+
{{ "FINALIZE = '" ~ config.get('task').finalize|upper ~ "'" if config.get('task').finalize|upper else '' -}}
18+
{{ "TASK_AUTO_RETRY_ATTEMPTS = " ~ config.get('task').task_auto_retry_attempts if config.get('task').task_auto_retry_attempts else '' -}}
19+
{{ "USER_TASK_MINIMUM_TRIGGER_INTERVAL_IN_SECONDS = " ~ config.get('task').user_task_minimum_trigger_interval_in_seconds if config.get('task').user_task_minimum_trigger_interval_in_seconds else '' -}}
20+
{{ "TARGET_COMPLETION_INTERVAL = '" ~ config.get('task').target_completion_interval|upper ~ "'" if config.get('task').target_completion_interval|upper else '' -}}
21+
{{ "SERVERLESS_TASK_MIN_STATEMENT_SIZE = '" ~ config.get('task').serverless_task_min_statement_size|upper ~ "'" if config.get('task').serverless_task_min_statement_size|upper else '' -}}
22+
{{ "SERVERLESS_TASK_MAX_STATEMENT_SIZE = '" ~ config.get('task').serverless_task_max_statement_size|upper ~ "'" if config.get('task').serverless_task_max_statement_size|upper else '' }}
23+
{{ "WHEN \n " ~ config.get('task').when|upper if config.get('task').when|upper else '' }}
24+
{%- endset -%}
25+
{%- do log('Printing Args SQL: ' ~ args_sql, info = true) -%}
26+
{%- set columns = adapter.get_columns_in_relation(tmp_relation) | rejectattr('name', 'equalto', 'METADATA$ACTION')
27+
| rejectattr('name', 'equalto', 'METADATA$ISUPDATE')
28+
| rejectattr('name', 'equalto', 'METADATA$ROW_ID')
29+
| list -%}
30+
{%- set merge_task -%}
31+
MERGE INTO {{target_table }} TARGET
32+
USING {{ stream }} SOURCE
33+
ON ({%- for key in unique_key -%}
34+
TARGET.{{ key }} = SOURCE.{{ key }}{%- if not loop.last -%} AND {%- endif -%}{%- endfor -%}
35+
)
36+
-- Mode DELETE
37+
WHEN MATCHED AND SOURCE.METADATA$ACTION = 'DELETE' AND SOURCE.METADATA$ISUPDATE = 'FALSE' THEN
38+
DELETE
39+
-- Mode UPDATE
40+
WHEN MATCHED AND SOURCE.METADATA$ACTION = 'INSERT' THEN
41+
UPDATE SET {% for col in columns %}
42+
TARGET.{{ col.name }} = SOURCE.{{ col.name }}{%- if not loop.last -%},{%- endif -%}{%- endfor -%}
43+
-- Mode INSERT
44+
WHEN NOT MATCHED AND SOURCE.METADATA$ACTION = 'INSERT' THEN INSERT
45+
( {%- for col in columns -%}
46+
{{- col.name -}}{%- if not loop.last -%},{%- endif -%}{%- endfor -%}
47+
)
48+
VALUES
49+
({%- for col in columns -%}
50+
SOURCE.{{- col.name -}}{%- if not loop.last -%},{%- endif -%}{%- endfor -%}
51+
)
52+
{%- endset -%}
53+
{%- set run_sql -%}
54+
CREATE OR ALTER TASK {{ name }}
55+
{{ args_sql }}
56+
AS
57+
{{ merge_task }}
58+
{%- endset -%}
59+
{{ run_sql }}
60+
{%- endmacro -%}

‎macros/helpers/stream.sql‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
{%- macro stream_source(source_name, table_name) -%}
2+
{{ dbt_snowflake_streams_tasks.stream('source', source_name, table_name) }}
3+
{%- endmacro -%}
4+
5+
{%- macro stream_ref(model_name) -%}
6+
{{ dbt_snowflake_streams_tasks.stream('model', none, model_name) }}
7+
{%- endmacro -%}
8+
9+
10+
{%- macro stream(model_type, source_name, model_name) -%}
11+
12+
{%- set full_refresh = flags.FULL_REFRESH == TRUE -%}
13+
{# Need to convert this to process a list of passed in streams #}
14+
15+
{%- set stream_name = '' -%}
16+
{%- set target_model = '' -%}
17+
18+
{%- if not execute -%}
19+
{% do log('Running During Parsing Mode', info=true)%}
20+
{% do log('Setting up Stream Configuration For Model Type: ' ~ model_type ~ ' Source: ' ~ source_name ~ ' Model Name: ' ~ model_name, info = true) %}
21+
22+
{% do config.set('src_table', {
23+
'model_type': model_type,
24+
'source_name': source_name,
25+
'model_name': model_name
26+
}) %}
27+
{%- endif -%}
28+
29+
{%- if model_type == 'model' -%}
30+
{%- set target_model = ref(model_name) -%}
31+
{% elif model_type == 'source' %}
32+
{%- set target_model = source(source_name, model_name) -%}
33+
{%- else -%}
34+
{%- do exceptions.raise_compiler_error("Invalid model_type provided to stream macro. Use 'model' or 'source'.") -%}
35+
{%- endif -%}
36+
37+
{%- if full_refresh -%}
38+
{%- do log('Target Model: ' ~ target_model, info=true ) -%}
39+
{{- target_model -}}
40+
{%- elif stream_name %}
41+
{{- stream_name -}}
42+
{%- else %}
43+
{% do log('Target Model: ' ~ target_model, info=true ) -%}
44+
{{- target_model -}}
45+
{%- endif -%}
46+
47+
{%- endmacro -%}

0 commit comments

Comments
 (0)