support duckdb

macros/internal/metadata_processing/escape_column_names.sql

@@ -132,6 +132,10 @@
     {%- do return (('`', '`')) -%}
 {%- endmacro %}
 
+{%- macro duckdb__get_escape_characters() %}
+    {%- do return (('', '')) -%}
+{%- endmacro %}
+
 {%- macro sqlserver__get_escape_characters() %}
     {%- do return (('"', '"')) -%}
 {%- endmacro %}

macros/materialisations/period_mat_helpers/get_period_of_load.sql

@@ -73,3 +73,18 @@
 
     {% do return(period_of_load) %}
 {%- endmacro -%}
+
+{%- macro duckdb__get_period_of_load(period, offset, start_timestamp) -%}
+    {# Postgres uses different DateTime arithmetic #}
+    {% set period_of_load_sql -%}
+        SELECT DATE_TRUNC('{{ period }}',
+               TO_TIMESTAMP('{{ start_timestamp }}', 'YYYY-MM-DD HH24:MI:SS') + interval '{{ offset }} {{ period }}'
+        ) AS period_of_load
+    {%- endset %}
+
+    {% set period_of_load_dict = dbtvault.get_query_results_as_dict(period_of_load_sql) %}
+
+    {% set period_of_load = period_of_load_dict['PERIOD_OF_LOAD'][0] | string %}
+
+    {% do return(period_of_load) %}
+{%- endmacro -%}

macros/staging/null_columns.sql

@@ -86,4 +86,13 @@
     {{ col_name_esc }} AS {{ col_name_orig_esc }},
     COALESCE({{ col_name_esc }}, '{{ default_value }}') AS {{ col_name_esc }}
 
+{%- endmacro -%}
+
+{%- macro duckdb__null_column_sql(col_name, default_value) -%}
+
+    {%- set col_name_esc = dbtvault.escape_column_names(col_name) -%}
+    {%- set col_name_orig_esc = dbtvault.escape_column_names(col_name ~ "_ORIGINAL") -%}
+    {{ col_name_esc }} AS {{ col_name_orig_esc }},
+    COALESCE({{ col_name_esc }}, '{{ default_value }}') AS {{ col_name_esc }}
+
 {%- endmacro -%}

macros/supporting/cast_date.sql

@@ -77,4 +77,10 @@
 
     {{ dbtvault.snowflake__cast_date(column_str=column_str, as_string=as_string, datetime=datetime, alias=alias)}}
 
+{%- endmacro -%}
+
+{%- macro duckdb__cast_date(column_str, as_string=false, datetime=false, alias=none) -%}
+
+    {{ dbtvault.snowflake__cast_date(column_str=column_str, as_string=as_string, datetime=datetime, alias=alias)}}
+
 {%- endmacro -%}

macros/supporting/hash.sql

@@ -303,6 +303,94 @@
 
 {%- endmacro -%}
 
+{%- macro duckdb__hash(columns, alias, is_hashdiff) -%}
+
+{%- set hash = var('hash', 'MD5') -%}
+{%- set concat_string = var('concat_string', '||') -%}
+{%- set null_placeholder_string = var('null_placeholder_string', '^^') -%}
+
+{#- Select hashing algorithm -#}
+{%- if hash == 'MD5' -%}
+    {%- set hash_alg = 'MD5' -%}
+{%- elif hash == 'SHA' -%}
+    {%- set hash_alg = 'SHA256' -%}
+{%- else -%}
+    {%- set hash_alg = 'MD5' -%}
+{%- endif -%}
+
+{#- Select hashing expression (left and right sides) -#}
+{#- * MD5 is simple function call to md5(val) -#}
+{#- * SHA256 needs input cast to BYTEA and then its BYTEA result encoded as hex text output -#}
+{#-   e.g. ENCODE(SHA256(CAST(val AS BYTEA)), 'hex') -#}
+{#- Ref: https://www.postgresql.org/docs/11/functions-binarystring.html  -#}
+{%- if hash_alg == 'MD5' -%}
+    {%- set hash_expr_left = 'MD5(' -%}
+    {%- set hash_expr_right = ')' -%}
+{%- elif hash_alg == 'SHA256' -%}
+    {%- set hash_expr_left = 'ENCODE(SHA256(CAST(' -%}
+    {%- set hash_expr_right = " AS BYTEA)), 'hex')" -%}
+{%- endif -%}
+
+{%- set standardise = "NULLIF(UPPER(TRIM(CAST([EXPRESSION] AS VARCHAR))), '')" -%}
+
+{#- Alpha sort columns before hashing if a hashdiff -#}
+{%- if is_hashdiff and dbtvault.is_list(columns) -%}
+    {%- set columns = columns|sort -%}
+{%- endif -%}
+
+{#- If single column to hash -#}
+{%- if columns is string -%}
+    {%- set column_str = dbtvault.as_constant(columns) -%}
+    {%- if dbtvault.is_expression(column_str) -%}
+        {%- set escaped_column_str = column_str -%}
+    {%- else -%}
+        {%- set escaped_column_str = dbtvault.escape_column_names(column_str) -%}
+    {%- endif -%}
+
+    {{- "CAST(UPPER({}{}{}) AS BYTEA) AS {}".format(hash_expr_left, standardise | replace('[EXPRESSION]', escaped_column_str), hash_expr_right, dbtvault.escape_column_names(alias)) | indent(4) -}}
+
+{#- Else a list of columns to hash -#}
+{%- else -%}
+    {%- set all_null = [] -%}
+
+    {%- if is_hashdiff -%}
+        {{- "CAST(UPPER({}CONCAT_WS('{}',".format(hash_expr_left, concat_string) | indent(4) -}}
+    {%- else -%}
+        {{- "CAST(UPPER({}NULLIF(CONCAT_WS('{}',".format(hash_expr_left, concat_string) | indent(4) -}}
+    {%- endif -%}
+
+    {%- for column in columns -%}
+
+        {%- do all_null.append(null_placeholder_string) -%}
+
+        {%- set column_str = dbtvault.as_constant(column) -%}
+        {%- if dbtvault.is_expression(column_str) -%}
+            {%- set escaped_column_str = column_str -%}
+        {%- else -%}
+            {%- set escaped_column_str = dbtvault.escape_column_names(column_str) -%}
+        {%- endif -%}
+
+        {{- "\nCOALESCE({}, '{}')".format(standardise | replace('[EXPRESSION]', escaped_column_str), null_placeholder_string) | indent(4) -}}
+        {{- "," if not loop.last -}}
+
+        {%- if loop.last -%}
+
+            {% if is_hashdiff %}
+                {{- "\n){}) AS BYTEA) AS {}".format(hash_expr_right, dbtvault.escape_column_names(alias)) -}}
+            {%- else -%}
+                {{- "\n), '{}'){}) AS BYTEA) AS {}".format(all_null | join(""), hash_expr_right, dbtvault.escape_column_names(alias)) -}}
+            {%- endif -%}
+        {%- else -%}
+
+            {%- do all_null.append(concat_string) -%}
+
+        {%- endif -%}
+    {%- endfor -%}
+
+{%- endif -%}
+
+{%- endmacro -%}
+
 {%- macro databricks__hash(columns, alias, is_hashdiff) -%}
 
 {%- set hash = var('hash', 'MD5') -%}

macros/tables/duckdb/hub.sql

@@ -0,0 +1,89 @@
+{%- macro duckdb__hub(src_pk, src_nk, src_extra_columns, src_ldts, src_source, source_model) -%}
+
+{%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_nk, src_extra_columns, src_ldts, src_source]) -%}
+
+{%- if model.config.materialized == 'vault_insert_by_rank' %}
+    {%- set source_cols_with_rank = source_cols + dbtvault.escape_column_names([config.get('rank_column')]) -%}
+{%- endif -%}
+
+{{ dbtvault.prepend_generated_by() }}
+
+{{ 'WITH ' -}}
+
+{%- if not (source_model is iterable and source_model is not string) -%}
+    {%- set source_model = [source_model] -%}
+{%- endif -%}
+
+{%- set ns = namespace(last_cte= "") -%}
+
+{%- for src in source_model -%}
+
+{%- set source_number = loop.index | string -%}
+
+row_rank_{{ source_number }} AS (
+{#- PostgreSQL has DISTINCT ON which should be more performant than the
+    strategy used by Snowflake ROW_NUMBER() OVER( PARTITION BY ...
+-#}
+    {%- if model.config.materialized == 'vault_insert_by_rank' %}
+    SELECT DISTINCT ON ({{ dbtvault.prefix([src_pk], 'rr') }}) {{ dbtvault.prefix(source_cols_with_rank, 'rr') }}
+    {%- else %}
+    SELECT DISTINCT ON ({{ dbtvault.prefix([src_pk], 'rr') }}) {{ dbtvault.prefix(source_cols, 'rr') }}
+    {%- endif %}
+    FROM {{ ref(src) }} AS rr
+    WHERE {{ dbtvault.multikey(src_pk, prefix='rr', condition='IS NOT NULL') }}
+    ORDER BY {{ dbtvault.prefix([src_pk], 'rr') }}, {{ dbtvault.prefix([src_ldts], 'rr') }}
+    {%- set ns.last_cte = "row_rank_{}".format(source_number) %}
+),{{ "\n" if not loop.last }}
+{% endfor -%}
+{% if source_model | length > 1 %}
+stage_union AS (
+    {%- for src in source_model %}
+    SELECT * FROM row_rank_{{ loop.index | string }}
+    {%- if not loop.last %}
+    UNION ALL
+    {%- endif %}
+    {%- endfor %}
+    {%- set ns.last_cte = "stage_union" %}
+),
+{%- endif -%}
+{%- if model.config.materialized == 'vault_insert_by_period' %}
+stage_mat_filter AS (
+    SELECT *
+    FROM {{ ns.last_cte }}
+    WHERE __PERIOD_FILTER__
+    {%- set ns.last_cte = "stage_mat_filter" %}
+),
+{%- elif model.config.materialized == 'vault_insert_by_rank' %}
+stage_mat_filter AS (
+    SELECT *
+    FROM {{ ns.last_cte }}
+    WHERE __RANK_FILTER__
+    {%- set ns.last_cte = "stage_mat_filter" %}
+),
+{%- endif -%}
+{%- if source_model | length > 1 %}
+
+row_rank_union AS (
+{#- PostgreSQL has DISTINCT ON which should be more performant than the
+    strategy used by Snowflake ROW_NUMBER() OVER( PARTITION BY ...
+-#}
+    SELECT DISTINCT ON ({{ dbtvault.prefix([src_pk], 'ru') }}) ru.*
+    FROM {{ ns.last_cte }} AS ru
+    WHERE {{ dbtvault.multikey(src_pk, prefix='ru', condition='IS NOT NULL') }}
+    ORDER BY {{ dbtvault.prefix([src_pk], 'ru') }}, {{ dbtvault.prefix([src_ldts], 'ru') }}, {{ dbtvault.prefix([src_source], 'ru') }} ASC
+    {%- set ns.last_cte = "row_rank_union" %}
+),
+{% endif %}
+records_to_insert AS (
+    SELECT {{ dbtvault.prefix(source_cols, 'a', alias_target='target') }}
+    FROM {{ ns.last_cte }} AS a
+    {%- if dbtvault.is_any_incremental() %}
+    LEFT JOIN {{ this }} AS d
+    ON {{ dbtvault.multikey(src_pk, prefix=['a','d'], condition='=') }}
+    WHERE {{ dbtvault.multikey(src_pk, prefix='d', condition='IS NULL') }}
+    {%- endif %}
+)
+
+SELECT * FROM records_to_insert
+
+{%- endmacro -%}

macros/tables/duckdb/link.sql

@@ -0,0 +1,100 @@
+{%- macro duckdb__link(src_pk, src_fk, src_extra_columns, src_ldts, src_source, source_model) -%}
+
+{%- set source_cols = dbtvault.expand_column_list(columns=[src_pk, src_fk, src_extra_columns, src_ldts, src_source]) -%}
+{%- set fk_cols = dbtvault.expand_column_list([src_fk]) -%}
+
+{%- if model.config.materialized == 'vault_insert_by_rank' %}
+    {%- set source_cols_with_rank = source_cols + dbtvault.escape_column_names([config.get('rank_column')]) -%}
+{%- endif -%}
+
+{{ dbtvault.prepend_generated_by() }}
+
+{{ 'WITH ' -}}
+
+{%- if not (source_model is iterable and source_model is not string) -%}
+    {%- set source_model = [source_model] -%}
+{%- endif -%}
+
+{%- set ns = namespace(last_cte= "") -%}
+
+{%- for src in source_model -%}
+
+{%- set source_number = loop.index | string -%}
+
+row_rank_{{ source_number }} AS (
+  SELECT * FROM (
+    {%- if model.config.materialized == 'vault_insert_by_rank' %}
+    SELECT {{ dbtvault.prefix(source_cols_with_rank, 'rr') }},
+    {%- else %}
+    SELECT {{ dbtvault.prefix(source_cols, 'rr') }},
+    {%- endif %}
+           ROW_NUMBER() OVER(
+               PARTITION BY {{ dbtvault.prefix([src_pk], 'rr') }}
+               ORDER BY {{ dbtvault.prefix([src_ldts], 'rr') }}
+           ) AS row_number
+    FROM {{ ref(src) }} AS rr
+    {%- if source_model | length == 1 %}
+    WHERE {{ dbtvault.multikey(src_pk, prefix='rr', condition='IS NOT NULL') }}
+    AND {{ dbtvault.multikey(fk_cols, prefix='rr', condition='IS NOT NULL') }}
+    {%- endif %}
+  ) as l
+  WHERE row_number = 1
+    {%- set ns.last_cte = "row_rank_{}".format(source_number) %}
+),{{ "\n" if not loop.last }}
+{% endfor -%}
+{% if source_model | length > 1 %}
+stage_union AS (
+    {%- for src in source_model %}
+    SELECT * FROM row_rank_{{ loop.index | string }}
+    {%- if not loop.last %}
+    UNION ALL
+    {%- endif %}
+    {%- endfor %}
+    {%- set ns.last_cte = "stage_union" %}
+),
+{%- endif -%}
+{%- if model.config.materialized == 'vault_insert_by_period' %}
+stage_mat_filter AS (
+    SELECT *
+    FROM {{ ns.last_cte }}
+    WHERE __PERIOD_FILTER__
+    {%- set ns.last_cte = "stage_mat_filter" %}
+),
+{%- elif model.config.materialized == 'vault_insert_by_rank' %}
+stage_mat_filter AS (
+    SELECT *
+    FROM {{ ns.last_cte }}
+    WHERE __RANK_FILTER__
+    {%- set ns.last_cte = "stage_mat_filter" %}
+),
+{% endif %}
+{%- if source_model | length > 1 %}
+
+row_rank_union AS (
+  SELECT * FROM (
+    SELECT ru.*,
+           ROW_NUMBER() OVER(
+               PARTITION BY {{ dbtvault.prefix([src_pk], 'ru') }}
+               ORDER BY {{ dbtvault.prefix([src_ldts], 'ru') }}, {{ dbtvault.prefix([src_source], 'ru') }} ASC
+           ) AS row_rank_number
+    FROM {{ ns.last_cte }} AS ru
+    WHERE {{ dbtvault.multikey(src_pk, prefix='ru', condition='IS NOT NULL') }}
+    AND {{ dbtvault.multikey(fk_cols, prefix='ru', condition='IS NOT NULL') }}
+  ) AS a
+  WHERE row_rank_number = 1
+    {%- set ns.last_cte = "row_rank_union" %}
+),
+{% endif %}
+records_to_insert AS (
+    SELECT {{ dbtvault.prefix(source_cols, 'a', alias_target='target') }}
+    FROM {{ ns.last_cte }} AS a
+    {%- if dbtvault.is_any_incremental() %}
+    LEFT JOIN {{ this }} AS d
+    ON {{ dbtvault.multikey(src_pk, prefix=['a','d'], condition='=') }}
+    WHERE {{ dbtvault.multikey(src_pk, prefix='d', condition='IS NULL') }}
+    {%- endif %}
+)
+
+SELECT * FROM records_to_insert
+
+{%- endmacro -%}