Merge pull request #54 from bcodell/bryce/bug-support-anonymous-id

Bug: Support `anonymous_customer_id`

.github/workflows/ci.yml

@@ -1,8 +1,6 @@
 name: CI
 on:
   workflow_call:
-    paths-ignore:
-      - 'README.md'
     secrets:
       gcp_keyfile:
         required: true
@@ -57,7 +55,7 @@ jobs:
           localstack start -d
           cd ./integration_tests
           sed -i 's/skip_stream: true/skip_stream: false/' dbt_project.yml
-          dbt build -x --target snowflake
+          dbt build -x --target snowflake --exclude dataset__select_all_aggregate_all
         env:
           DBT_PROFILES_DIR: . # Use integration_tests/profiles.yml
           DEBUG: 1
@@ -66,11 +64,10 @@ jobs:
         id: snowflake_ci_skip_stream
         if: github.repository == 'bcodell/dbt-aql'
         run: |
-          localstack stop
-          localstack start -d
           cd ./integration_tests
           sed -i 's/skip_stream: false/skip_stream: true/' dbt_project.yml
-          dbt build -x --target snowflake
+          dbt run -s activities+ -x --target snowflake --exclude dataset__select_all_aggregate_all
+          dbt test -s activities+ -x --target snowflake --exclude dataset__select_all_aggregate_all
         env:
           DBT_PROFILES_DIR: . # Use integration_tests/profiles.yml
           DEBUG: 1
@@ -95,7 +92,8 @@ jobs:
           sed -i 's/skip_stream: false/skip_stream: true/' dbt_project.yml
           echo $GCP_KEYFILE > $GCP_KEYFILE_PATH
           ls -l
-          dbt build -s activities+ -x --target bigquery --exclude config.materialized:seed
+          dbt run -s activities+ -x --target bigquery
+          dbt test -s activities+ -x --target bigquery
         env:
           GCP_KEYFILE_PATH: ./gcp_keyfile.json
           DBT_PROFILES_DIR: . # Use integration_tests/profiles.yml

.github/workflows/pr.yml

@@ -2,8 +2,6 @@ name: CI - Pull Request
 on: 
   pull_request:
     branches: [ main ]
-    paths-ignore:
-      - 'README.md'
 
 jobs:
   CI:

integration_tests/dbt_project.yml

@@ -34,11 +34,6 @@ vars:
         anonymous_customer_id_alias: anonymous_entity_uuid
         model_prefix: customer__
 
-quoting:
-  database: "{{ true if target.name == 'snowflake' else false }}"
-  schema: "{{ true if target.name == 'snowflake' else false }}"
-  identifier: "{{ true if target.name == 'snowflake' else false }}"
-
 
 seeds:
   dbt_aql_integration_tests:

integration_tests/macros/listagg_delimiter.sql

@@ -0,0 +1,3 @@
+{% macro default__listagg_delimiter() %}
+{%- do return(dbt.string_literal(",")) -%}
+{% endmacro %}

integration_tests/models/datasets/primary__all/dataset__select_all_aggregate_all.sql

@@ -3,10 +3,13 @@ using customer_stream
 select all signed_up (
     activity_id as activity_id,
     entity_uuid as customer_id,
+    anonymous_entity_uuid as anonymous_customer_id,
     ts as signed_up_at
 )
 aggregate all visited_page (
-    count(activity_id) as pages_visited_all
+    count(activity_id) as pages_visited_all,
+    listagg(referrer_url) as referrer_urls,
+    listagg_distinct(referrer_url) as distinct_referrer_urls
 )
 aggregate all bought_something(
     sum(total_items_purchased) as total_items_purchased_all,

integration_tests/seeds/datasets/primary__all/output__select_all_aggregate_all.csv

@@ -1,5 +1,5 @@
-activity_id,customer_id,signed_up_at,pages_visited_all,total_items_purchased_all,total_sales_all,total_purchases_all
-5a5e6cf0133a3faa73356ff291e26471,1,2022-01-02 22:10:11,3,11,600,3
-574b51c586b446456ac8da4b504b4d27,4,2022-01-05 22:10:11,3,8,680,3
-f77c4be94da1197fb6e8864f59b85dd2,7,2022-01-08 22:10:11,3,5,270,3
-22ef1cea8e5cb3dc154fd727b1776a6e,10,2022-01-11 22:10:11,3,22,1650,3
+activity_id,customer_id,anonymous_customer_id,signed_up_at,pages_visited_all,referrer_urls,distinct_referrer_urls,total_items_purchased_all,total_sales_all,total_purchases_all
+5a5e6cf0133a3faa73356ff291e26471,1,,2022-01-02 22:10:11,3,"bing.com,google.com,yahoo.com","bing.com,google.com,yahoo.com",11,600,3
+f77c4be94da1197fb6e8864f59b85dd2,7,,2022-01-08 22:10:11,3,"gmail.com,google.com,yahoo.com","gmail.com,google.com,yahoo.com",5,270,3
+22ef1cea8e5cb3dc154fd727b1776a6e,10,,2022-01-11 22:10:11,3,"bing.com,gmail.com,google.com","bing.com,gmail.com,google.com",22,1650,3
+574b51c586b446456ac8da4b504b4d27,4,,2022-01-05 22:10:11,3,"bing.com,gmail.com,yahoo.com","bing.com,gmail.com,yahoo.com",8,680,3

integration_tests/seeds/datasets/seed_schema.yml

@@ -14,6 +14,13 @@ seeds:
         last_before_total_items_purchased: integer
         last_before_bought_something_at: "{{ 'datetime' if target.name == 'bigquery' else 'timestamp' }}"
 
+  - name: output__select_all_aggregate_all
+    config:
+      column_types:
+        anonymous_customer_id: "{{ 'string' if target.name == 'bigquery' else 'text' }}"
+        last_before_total_items_purchased: integer
+        last_before_bought_something_at: "{{ 'datetime' if target.name == 'bigquery' else 'timestamp' }}"
+
   - name: output__select_all_append_nth_before
     config:
       column_types:

macros/activity_schema/activity/build_activity.sql

@@ -86,11 +86,19 @@ select
     {% endif %}
     , {{ dbt_aql.build_json(data_types) }} as {{columns.feature_json}}
     , row_number() over (
+        {% if columns.anonymous_customer_id is not defined or 'anonymous_customer_id' in null_columns %}
         partition by {{columns.customer}}
+        {% else %}
+        partition by coalesce({{columns.customer}}, {{columns.anonymous_customer_id}})
+        {% endif %}
         order by {{columns.ts}}, {{surrogate_key_statement}}
     ) as activity_occurrence
     , lead(cast({{columns.ts}} as {{dbt.type_timestamp()}})) over (
+        {% if columns.anonymous_customer_id is not defined or 'anonymous_customer_id' in null_columns %}
         partition by {{columns.customer}}
+        {% else %}
+        partition by coalesce({{columns.customer}}, {{columns.anonymous_customer_id}})
+        {% endif %}
         order by {{columns.ts}}, {{surrogate_key_statement}}
     ) as activity_repeated_at
 from {{cte}}

macros/activity_schema/dataset/aggregations/_helpers.sql

@@ -1,4 +1,8 @@
-{% macro _listagg_delimiter() %}
+{% macro listagg_delimiter() %}
+    {{ return(adapter.dispatch("listagg_delimiter", "dbt_aql")())}}
+{% endmacro %}
+
+{% macro default__listagg_delimiter() %}
 {%- do return(dbt.string_literal("\n")) -%}
 {% endmacro %}
 

macros/activity_schema/dataset/aggregations/listagg.sql

@@ -3,5 +3,17 @@
 {% endmacro %}
 
 {% macro default__aggfunc_listagg(column) %}
-listagg({{ column.column_sql }}, {{dbt_aql._listagg_delimiter()}})
+nullif(string_agg({{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}} order by {{ column.column_sql }}), '')
+{% endmacro %}
+
+{% macro duckdb__aggfunc_listagg(column) %}
+nullif(string_agg({{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}} order by {{ column.column_sql }}), '')
+{% endmacro %}
+
+{% macro snowflake__aggfunc_listagg(column) %}
+nullif(listagg({{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}}) within group (order by {{ column.column_sql }}), '')
+{% endmacro %}
+
+{% macro redshift__aggfunc_listagg(column) %}
+nullif(listagg({{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}}), '')
 {% endmacro %}

macros/activity_schema/dataset/aggregations/listagg_distinct.sql

@@ -3,5 +3,17 @@
 {% endmacro %}
 
 {% macro default__aggfunc_listagg_distinct(column) %}
-listagg(distinct {{ column.column_sql }}, {{dbt_aql._listagg_delimiter()}})
+nullif(string_agg(distinct {{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}} order by {{ column.column_sql }}), '')
+{% endmacro %}
+
+{% macro duckdb__aggfunc_listagg_distinct(column) %}
+nullif(string_agg(distinct {{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}} order by {{ column.column_sql }}), '')
+{% endmacro %}
+
+{% macro snowflake__aggfunc_listagg_distinct(column) %}
+nullif(listagg(distinct {{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}}) within group (order by {{ column.column_sql }}), '')
+{% endmacro %}
+
+{% macro redshift__aggfunc_listagg_distinct(column) %}
+nullif(listagg(distinct {{ column.column_sql }}, {{dbt_aql.listagg_delimiter()}}), '')
 {% endmacro %}

macros/activity_schema/dataset/dataset.sql

@@ -91,11 +91,19 @@ with
         {%- endif %}
         {%- endfor %}
         row_number() over (
+            {% if columns.anonymous_customer_id is not defined %}
             partition by {{primary}}.{{columns.customer}}
+            {% else %}
+            partition by coalesce({{primary}}.{{columns.customer}}, {{primary}}.{{columns.anonymous_customer_id}})
+            {% endif %}
             order by {{primary}}.{{columns.ts}}, {{primary}}.{{columns.activity_id}}
         ) as {{columns.activity_occurrence}},
         lead({{columns.ts}}) over (
+            {% if columns.anonymous_customer_id is not defined %}
             partition by {{primary}}.{{columns.customer}}
+            {% else %}
+            partition by coalesce({{primary}}.{{columns.customer}}, {{primary}}.{{columns.anonymous_customer_id}})
+            {% endif %}
             order by {{primary}}.{{columns.ts}}, {{primary}}.{{columns.activity_id}}
         ) as {{columns.activity_repeated_at}}
     {% if not skip_stream %}
@@ -121,6 +129,9 @@ with
         {{primary}}.{{columns.ts}} as {{req}}{{columns.ts}},
         {{primary}}.{{columns.activity_occurrence}} as {{req}}{{columns.activity_occurrence}},
         {{primary}}.{{columns.activity_repeated_at}} as {{req}}{{columns.activity_repeated_at}},
+        {% if columns.anonymous_customer_id is defined %}
+        {{primary}}.{{columns.anonymous_customer_id}} as {{req}}{{columns.anonymous_customer_id}},
+        {% endif %}
         {%- for column in primary_activity.columns %}
         {{ dbt_aql.select_column(stream, primary, column).column_sql }} as {{column.alias}}{% if not loop.last -%},{%- endif -%}
         {%- endfor %}
@@ -143,11 +154,19 @@ with
         {%- endif %}
         {%- endfor %}
         row_number() over (
+            {% if columns.anonymous_customer_id is not defined %}
             partition by {{joined}}.{{columns.customer}}
+            {% else %}
+            partition by coalesce({{joined}}.{{columns.customer}}, {{joined}}.{{columns.anonymous_customer_id}})
+            {% endif %}
             order by {{joined}}.{{columns.ts}}, {{joined}}.{{columns.activity_id}}
         ) as {{columns.activity_occurrence}},
         lead({{columns.ts}}) over (
+            {% if columns.anonymous_customer_id is not defined %}
             partition by {{joined}}.{{columns.customer}}
+            {% else %}
+            partition by coalesce({{joined}}.{{columns.customer}}, {{joined}}.{{columns.anonymous_customer_id}})
+            {% endif %}
             order by {{joined}}.{{columns.ts}}, {{joined}}.{{columns.activity_id}}
         ) as {{columns.activity_repeated_at}}
     {% if not skip_stream %}
@@ -183,7 +202,11 @@ with
         {%- if ja.relationship_clause is not none %}
         and {{ ja.relationship_clause }}
         {%- endif %}
+        {% if columns.anonymous_customer_id is defined %}
+        and coalesce({{primary}}.{{req}}{{columns.customer}}, {{primary}}.{{req}}{{columns.anonymous_customer_id}}) = coalesce({{joined}}.{{columns.customer}}, {{joined}}.{{columns.anonymous_customer_id}})
+        {% else %}
         and {{primary}}.{{req}}{{columns.customer}} = {{joined}}.{{columns.customer}}
+        {% endif %}
         and {{ ja.join_clause }}
         {%- if ja.extra_joins is not none %}
         {%- for ej in ja.extra_joins %}