model updates

.buildkite/pipeline.yml

@@ -58,7 +58,7 @@ steps:
     commands: |
       bash .buildkite/scripts/run_models.sh redshift
 
-  - label: ":bricks: Run Tests - Databricks"
+  - label: ":databricks: Run Tests - Databricks"
     key: "run_dbt_databricks"
     plugins:
       - docker#v3.13.0:

dbt_project.yml

@@ -42,6 +42,7 @@ vars:
     shopify_product_image: "{{ ref('stg_shopify__product_image') }}"
     shopify_product_tag: "{{ ref('stg_shopify__product_tag') }}"
     shopify_shop: "{{ ref('stg_shopify__shop') }}"
+    shopify_tax_line: "{{ ref('stg_shopify__tax_line') }}"
     shopify_tender_transaction: "{{ ref('stg_shopify__tender_transaction') }}"
     shopify_abandoned_checkout_discount_code: "{{ ref('stg_shopify__abandoned_checkout_discount_code') }}"
     shopify_order_discount_code: "{{ ref('stg_shopify__order_discount_code') }}"

integration_tests/ci/sample.profiles.yml

@@ -16,13 +16,13 @@ integration_tests:
       pass: "{{ env_var('CI_REDSHIFT_DBT_PASS') }}"
       dbname: "{{ env_var('CI_REDSHIFT_DBT_DBNAME') }}"
       port: 5439
-      schema: shopify_integration_tests_1
+      schema: shopify_integration_tests_3
       threads: 8
     bigquery:
       type: bigquery
       method: service-account-json
       project: 'dbt-package-testing'
-      schema: shopify_integration_tests_1
+      schema: shopify_integration_tests_3
       threads: 8
       keyfile_json: "{{ env_var('GCLOUD_SERVICE_KEY') | as_native }}"
     snowflake:
@@ -33,7 +33,7 @@ integration_tests:
       role: "{{ env_var('CI_SNOWFLAKE_DBT_ROLE') }}"
       database: "{{ env_var('CI_SNOWFLAKE_DBT_DATABASE') }}"
       warehouse: "{{ env_var('CI_SNOWFLAKE_DBT_WAREHOUSE') }}"
-      schema: shopify_integration_tests_1
+      schema: shopify_integration_tests_3
       threads: 8
     postgres:
       type: postgres
@@ -42,13 +42,13 @@ integration_tests:
       pass: "{{ env_var('CI_POSTGRES_DBT_PASS') }}"
       dbname: "{{ env_var('CI_POSTGRES_DBT_DBNAME') }}"
       port: 5432
-      schema: shopify_integration_tests_1
+      schema: shopify_integration_tests_3
       threads: 8
     databricks:
       catalog: null
       host: "{{ env_var('CI_DATABRICKS_DBT_HOST') }}"
       http_path: "{{ env_var('CI_DATABRICKS_DBT_HTTP_PATH') }}"
-      schema: shopify_integration_tests_1
+      schema: shopify_integration_tests_3
       threads: 2
       token: "{{ env_var('CI_DATABRICKS_DBT_TOKEN') }}"
       type: databricks

integration_tests/dbt_project.yml

@@ -5,12 +5,12 @@ config-version: 2
 
 
 vars:
-  shopify_schema: shopify_integration_tests_1
+  shopify_schema: shopify_integration_tests_3
   shopify_source:
     shopify_customer_identifier:          "shopify_customer_data"
     shopify_order_line_refund_identifier: "shopify_order_line_refund_data"
     shopify_order_line_identifier:        "shopify_order_line_data"
-    order_source:                         "{{ ref('shopify_order_data') }}"
+    shopify_order_identifier:             "shopify_order_data"
     shopify_product_identifier:           "shopify_product_data"
     shopify_order_adjustment_identifier:  "shopify_order_adjustment_data"
     shopify_product_variant_identifier:   "shopify_product_variant_data"
@@ -36,6 +36,7 @@ vars:
     shopify_product_image_identifier: "shopify_product_image_data"
     shopify_product_tag_identifier: "shopify_product_tag_data"
     shopify_shop_identifier: "shopify_shop_data"
+    shopify_tax_line_identifier: "shopify_tax_line_data"
     shopify_tender_transaction_identifier: "shopify_tender_transaction_data"
     shopify_abandoned_checkout_discount_code_identifier: "shopify_abandoned_checkout_discount_code_data"
     shopify_order_discount_code_identifier: "shopify_order_discount_code_data"
@@ -55,6 +56,7 @@ seeds:
         _fivetran_synced: timestamp
         id: "{%- if target.type == 'bigquery' -%} INT64 {%- else -%} bigint {%- endif -%}"
         default_address_id: "{%- if target.type == 'bigquery' -%} INT64 {%- else -%} bigint {%- endif -%}"
+        email: "{{ 'string' if target.type in ('bigquery', 'spark', 'databricks') else 'varchar' }}"
     shopify_order_data:
       +column_types:
         created_at: timestamp
@@ -166,4 +168,8 @@ seeds:
         markup: "{{ 'string' if target.type in ('bigquery', 'spark', 'databricks') else 'varchar' }}"
         price: float
         original_shop_markup: "{{ 'string' if target.type in ('bigquery', 'spark', 'databricks') else 'varchar' }}"
-        original_shop_price: "{{ 'string' if target.type in ('bigquery', 'spark', 'databricks') else 'varchar' }}"
+        original_shop_price: "{{ 'string' if target.type in ('bigquery', 'spark', 'databricks') else 'varchar' }}"
+    shopify_tax_line_data:
+      +column_types:
+        price: float
+        rate: float

integration_tests/seeds/shopify_tax_line_data.csv

@@ -0,0 +1,6 @@
+index,order_line_id,_fivetran_synced,price,rate,title,price_set
+1,29227,2022-11-19 05:30:34.023000,0.0,0.0,VAT,"{""shop_money"":{""amount"":""0.00"",""currency_code"":""USD""},""presentment_money"":{""amount"":""0.00"",""currency_code"":""USD""}}"
+1,1839083,2022-11-19 07:14:05.023000,0.0,0.0,VAT,"{""shop_money"":{""amount"":""0.00"",""currency_code"":""USD""},""presentment_money"":{""amount"":""0.00"",""currency_code"":""USD""}}"
+1,11995,2022-11-19 05:30:34.023000,0.0,0.0,VAT,"{""shop_money"":{""amount"":""0.00"",""currency_code"":""USD""},""presentment_money"":{""amount"":""0.00"",""currency_code"":""USD""}}"
+1,10751,2022-11-19 07:14:05.024000,0.0,0.0,VAT,"{""shop_money"":{""amount"":""0.00"",""currency_code"":""USD""},""presentment_money"":{""amount"":""0.00"",""currency_code"":""USD""}}"
+1,194763,2022-11-19 05:30:34.023000,0.0,0.0,VAT,"{""shop_money"":{""amount"":""0.00"",""currency_code"":""USD""},""presentment_money"":{""amount"":""0.00"",""currency_code"":""USD""}}"

models/intermediate/int_shopify__customer_email_rollup.sql

@@ -7,29 +7,36 @@ with customers as (
     from {{ var('shopify_customer') }}
     where email is not null -- nonsensical to include any null emails here
 
+), customer_tags as (
+
+    select 
+        *
+    from {{ var('shopify_customer_tag' )}}
+
 ), rollup_customers as (
 
     select
         -- fields to group by
-        lower(email) as email,
-        source_relation,
+        lower(customers.email) as email,
+        customers.source_relation,
 
         -- fields to string agg together
-        {{ fivetran_utils.string_agg("cast(customer_id as " ~ dbt.type_string() ~ ")", "', '") }} as customer_ids,
-        {{ fivetran_utils.string_agg("distinct cast(phone as " ~ dbt.type_string() ~ ")", "', '") }} as phone_numbers,
+        {{ fivetran_utils.string_agg("cast(customers.customer_id as " ~ dbt.type_string() ~ ")", "', '") }} as customer_ids,
+        {{ fivetran_utils.string_agg("distinct cast(customers.phone as " ~ dbt.type_string() ~ ")", "', '") }} as phone_numbers,
+        {{ fivetran_utils.string_agg("distinct cast(customer_tags.value as " ~ dbt.type_string() ~ ")", "', '") }} as customer_tags,
 
         -- fields to take aggregates of
-        min(created_timestamp) as first_account_created_at,
-        max(created_timestamp) as last_account_created_at,
-        max(updated_timestamp) as last_updated_at,
-        max(marketing_consent_updated_at) as marketing_consent_updated_at,
-        max(_fivetran_synced) as last_fivetran_synced,
-        sum(orders_count) as orders_count,
-        sum(total_spent) as total_spent,
+        min(customers.created_timestamp) as first_account_created_at,
+        max(customers.created_timestamp) as last_account_created_at,
+        max(customers.updated_timestamp) as last_updated_at,
+        max(customers.marketing_consent_updated_at) as marketing_consent_updated_at,
+        max(customers._fivetran_synced) as last_fivetran_synced,
+        sum(customers.orders_count) as orders_count,
+        sum(customers.total_spent) as total_spent,
 
         -- take true if ever given for boolean fields
-        {{ fivetran_utils.max_bool("case when customer_index = 1 then is_tax_exempt else null end") }} as is_tax_exempt, -- since this changes every year
-        {{ fivetran_utils.max_bool("is_verified_email") }} as is_verified_email
+        {{ fivetran_utils.max_bool("case when customers.customer_index = 1 then customers.is_tax_exempt else null end") }} as is_tax_exempt, -- since this changes every year
+        {{ fivetran_utils.max_bool("customers.is_verified_email") }} as is_verified_email
 
         -- for all other fields, just take the latest value
         {% set cols = adapter.get_columns_in_relation(ref('stg_shopify__customer')) %}
@@ -38,15 +45,18 @@ with customers as (
                                 'is_tax_exempt', 'is_verified_email'] %}
         {% for col in cols %}
             {% if col.column|lower not in except_cols %}
-            , max(case when customer_index = 1 then {{ col.column }} else null end) as {{ col.column }}
+            , max(case when customers.customer_index = 1 then customers.{{ col.column }} else null end) as {{ col.column }}
             {% endif %}
         {% endfor %}
 
     from customers 
+    left join customer_tags
+        on customers.customer_id = customer_tags.customer_id
+        and customers.source_relation = customer_tags.source_relation
 
     group by 1,2
 
 )
 
 select *
-from rollup_customers
+from rollup_customers

models/intermediate/int_shopify__discounts__abandoned_checkouts.sql

@@ -24,14 +24,26 @@ abandoned_checkout_shipping_line as (
     from {{ var('shopify_abandoned_checkout_shipping_line') }}
 ),
 
+roll_up_shipping_line as (
+
+    select 
+        checkout_id,
+        source_relation,
+        sum(price) as price
+
+    from abandoned_checkout_shipping_line
+    group by 1,2
+),
+
 abandoned_checkouts_aggregated as (
 
     select 
         abandoned_checkout_discount_code.code,
         abandoned_checkout_discount_code.type,
         abandoned_checkout_discount_code.source_relation,
         sum(abandoned_checkout_discount_code.amount) as total_abandoned_checkout_discount_amount,
-        sum(coalesce(abandoned_checkout_shipping_line.price, 0)) as total_abandoned_checkout_shipping_price,
+        sum(coalesce(abandoned_checkout.total_line_items_price, 0)) as total_abandoned_checkout_line_items_price,
+        sum(coalesce(roll_up_shipping_line.price, 0)) as total_abandoned_checkout_shipping_price,
         count(distinct customer_id) as count_abandoned_checkout_customers,
         count(distinct email) as count_abandoned_checkout_customer_emails,
         count(distinct abandoned_checkout.checkout_id) as count_abandoned_checkouts
@@ -40,9 +52,9 @@ abandoned_checkouts_aggregated as (
     left join abandoned_checkout
         on abandoned_checkout_discount_code.checkout_id = abandoned_checkout.checkout_id
         and abandoned_checkout_discount_code.source_relation = abandoned_checkout.source_relation
-    left join abandoned_checkout_shipping_line
-        on abandoned_checkout_shipping_line.checkout_id = abandoned_checkout_discount_code.checkout_id 
-        and abandoned_checkout_shipping_line.source_relation = abandoned_checkout_discount_code.source_relation
+    left join roll_up_shipping_line
+        on roll_up_shipping_line.checkout_id = abandoned_checkout_discount_code.checkout_id 
+        and roll_up_shipping_line.source_relation = abandoned_checkout_discount_code.source_relation
 
     group by 1,2,3
 )

models/intermediate/int_shopify__discounts__order_aggregates.sql

@@ -18,9 +18,9 @@ orders_aggregated as (
         order_discount_code.source_relation,
         avg(order_discount_code.amount) as avg_order_discount_amount,
         sum(order_discount_code.amount) as total_order_discount_amount,
-        sum(orders.total_line_items_price) as total_order_line_items_price,
-        sum(orders.shipping_cost) as total_order_shipping_cost,
-        sum(orders.refund_subtotal + orders.refund_total_tax) as total_order_refund_amount,
+        max(orders.total_line_items_price) as total_order_line_items_price, -- summing would multiply the total by the # of discount codes applied to an order
+        max(orders.shipping_cost) as total_order_shipping_cost, -- summing would multiply the total by the # of discount codes applied to an order
+        max(orders.refund_subtotal + orders.refund_total_tax) as total_order_refund_amount, -- summing would multiply the total by the # of discount codes applied to an order
         count(distinct customer_id) as count_customers,
         count(distinct email) as count_customer_emails,
         count(distinct order_discount_code.order_id) as count_orders

models/intermediate/int_shopify__emails__order_aggregates.sql

@@ -4,11 +4,30 @@ with orders as (
     from {{ var('shopify_order') }}
     where email is not null
 
+), order_aggregates as (
+
+    select *
+    from {{ ref('shopify__orders__order_line_aggregates') }}
+
 ), transactions as (
 
     select *
-    from {{ ref('shopify__transactions' )}}
+    from {{ ref('shopify__transactions')}}
+
     where lower(status) = 'success'
+    and lower(kind) not in ('authorization', 'void')
+    and lower(gateway) != 'gift_card' -- redeeming a giftcard does not introduce new revenue
+
+), transaction_aggregates as (
+    -- this is necessary as customers can pay via multiple payment gateways
+    select 
+        order_id,
+        source_relation,
+        kind,
+        sum(currency_exchange_calculated_amount) as currency_exchange_calculated_amount
+
+    from transactions
+    {{ dbt_utils.group_by(n=3) }}
 
 ), aggregated as (
 
@@ -17,14 +36,34 @@ with orders as (
         orders.source_relation,
         min(orders.created_timestamp) as first_order_timestamp,
         max(orders.created_timestamp) as most_recent_order_timestamp,
-        avg(case when lower(transactions.kind) in ('sale','capture') then transactions.currency_exchange_calculated_amount end) as average_order_value,
-        sum(case when lower(transactions.kind) in ('sale','capture') then transactions.currency_exchange_calculated_amount end) as lifetime_total_spent,
-        sum(case when lower(transactions.kind) in ('refund') then transactions.currency_exchange_calculated_amount end) as lifetime_total_refunded,
-        count(distinct orders.order_id) as lifetime_count_orders
+        avg(transaction_aggregates.currency_exchange_calculated_amount) as avg_order_value,
+        sum(transaction_aggregates.currency_exchange_calculated_amount) as lifetime_total_spent,
+        sum(refunds.currency_exchange_calculated_amount) as lifetime_total_refunded,
+        count(distinct orders.order_id) as lifetime_count_orders,
+        avg(order_aggregates.order_total_quantity) as avg_quantity_per_order,
+        sum(order_aggregates.order_total_tax) as lifetime_total_tax,
+        avg(order_aggregates.order_total_tax) as avg_tax_per_order,
+        sum(order_aggregates.order_total_discount) as lifetime_total_discount,
+        avg(order_aggregates.order_total_discount) as avg_discount_per_order,
+        sum(order_aggregates.order_total_shipping) as lifetime_total_shipping,
+        avg(order_aggregates.order_total_shipping) as avg_shipping_per_order,
+        sum(order_aggregates.order_total_shipping_with_discounts) as lifetime_total_shipping_with_discounts,
+        avg(order_aggregates.order_total_shipping_with_discounts) as avg_shipping_with_discounts_per_order,
+        sum(order_aggregates.order_total_shipping_tax) as lifetime_total_shipping_tax,
+        avg(order_aggregates.order_total_shipping_tax) as avg_shipping_tax_per_order
     from orders
-    left join transactions
-        on orders.order_id = transactions.order_id 
-        and orders.source_relation = transactions.source_relation
+    left join transaction_aggregates 
+        on orders.order_id = transaction_aggregates.order_id
+        and orders.source_relation = transaction_aggregates.source_relation
+        and transaction_aggregates.kind in ('sale','capture')
+    left join transaction_aggregates as refunds
+        on orders.order_id = refunds.order_id
+        and orders.source_relation = refunds.source_relation
+        and refunds.kind = 'refund'
+    left join order_aggregates
+        on orders.order_id = order_aggregates.order_id
+        and orders.source_relation = order_aggregates.source_relation
+
     group by 1,2
 
 )

models/intermediate/int_shopify__order__shipping_aggregates.sql

@@ -0,0 +1,40 @@
+with order_shipping_line as (
+
+    select
+        order_id,
+        source_relation,
+        order_shipping_line_id,
+        sum(price) as shipping_price,
+        sum(discounted_price) as discounted_shipping_price
+
+    from {{ var('shopify_order_shipping_line') }}
+    group by 1,2,3
+
+), order_shipping_tax_line as (
+
+    select
+        order_shipping_line_id,
+        source_relation,
+        sum(price) as shipping_tax
+
+    from {{ var('shopify_order_shipping_tax_line') }}
+    group by 1,2 
+
+), aggregated as (
+
+    select 
+        order_shipping_line.order_id,
+        order_shipping_line.source_relation,
+        sum(order_shipping_line.shipping_price) as shipping_price,
+        sum(order_shipping_line.discounted_shipping_price) as discounted_shipping_price,
+        sum(order_shipping_tax_line.shipping_tax) as shipping_tax
+
+    from order_shipping_line
+    left join order_shipping_tax_line
+        on order_shipping_line.order_shipping_line_id = order_shipping_tax_line.order_shipping_line_id
+        and order_shipping_line.source_relation = order_shipping_tax_line.source_relation
+    group by 1,2
+)
+
+select * 
+from aggregated

models/intermediate/int_shopify__product__order_line_aggregates.sql

@@ -0,0 +1,40 @@
+with order_lines as (
+
+    select *
+    from {{ ref('shopify__order_lines') }}
+
+), orders as (
+
+    select *
+    from {{ ref('shopify__orders')}}
+
+), product_aggregated as (
+    select 
+        order_lines.product_id,
+        order_lines.source_relation,
+
+        -- moved over from shopify__products
+        sum(order_lines.quantity) as quantity_sold,
+        sum(order_lines.pre_tax_price) as subtotal_sold,
+        sum(order_lines.quantity_net_refunds) as quantity_sold_net_refunds,
+        sum(order_lines.subtotal_net_refunds) as subtotal_sold_net_refunds,
+        min(orders.created_timestamp) as first_order_timestamp,
+        max(orders.created_timestamp) as most_recent_order_timestamp,
+
+        -- new columns
+        sum(order_lines.total_discount) as product_total_discount,
+        sum(order_lines.order_line_tax) as product_total_tax,
+        avg(order_lines.quantity) as avg_quantity_per_order_line,
+        avg(order_lines.total_discount) as product_avg_discount_per_order_line,
+        avg(order_lines.order_line_tax) as product_avg_tax_per_order_line
+
+    from order_lines
+    left join orders
+        on order_lines.order_id = orders.order_id
+        and order_lines.source_relation = orders.source_relation
+    group by 1,2
+
+)
+
+select *
+from product_aggregated