Plan: Scheduling — Incremental Refresh Strategy                                                                                                                                           
                                                                                                                                                                                         
 Context                                                                                                                                                                                 

 30-minute scheduling of the Shopify global schema tables. Full CREATE OR REPLACE TABLE on every run is fine for small tables (PRODUCT: 26K rows, ~55s) but will be too slow/expensive for large tables like ORDER or ORDER_LINE (potentially millions of rows). Need incremental updates: insert new rows, update changed rows.

 How Fivetran incremental detection works

 Fivetran stamps every row with _FIVETRAN_SYNCED (timestamp of last sync) and _FIVETRAN_DELETED (soft-delete flag). These are already in the PRODUCT table and will be in all entity tables. This is the standard filter key for incremental dbt models against Fivetran sources.

 Approach: dbt incremental materialization with Snowflake MERGE

 dbt's incremental + unique_key compiles to a Snowflake MERGE statement:
 - New rows → INSERT
 - Existing rows with same unique key → UPDATE (picks up field changes)
 - Deleted rows → handled via _FIVETRAN_DELETED flag (soft delete, not physical row removal)

 On first run (or dbt run --full-refresh): builds full table identically to materialized='table'.
 On subsequent runs: only processes rows where _FIVETRAN_SYNCED > max(_FIVETRAN_SYNCED) already in target.

 Required changes

 1. macros/generate_shopify_table_union.sql — add incremental filter support

 Add an optional second parameter incremental_filter (default empty string). When provided, append a WHERE clause to each source table's SELECT:

 from CRM_ECOMMERCE_DATA.{{ schema }}.{{ table_name }}
 {% if incremental_filter %}where {{ incremental_filter }}{% endif %}

 Called with filter when incremental:
 {{ generate_shopify_table_union('PRODUCT', incremental_filter) }}

 2. models/staging/stg_shopify_product.sql — switch to incremental

 Config change:
 {{ config(
     materialized='incremental',
     unique_key=['SOURCE_SCHEMA', 'ID'],
     incremental_strategy='merge',
     alias='BF_SHOPIFY_PRODUCT'
 ) }}

 Add SOURCE_SCHEMA as an explicit column (needed as part of the unique key — product IDs are only unique within a schema):
 p.SOURCE_DATABASE || '.' || p.SOURCE_SCHEMA as SOURCE_IDENTIFIER,
 split_part(p.SOURCE_DATABASE || '.' || p.SOURCE_SCHEMA, '.', 2) as SOURCE_SCHEMA_KEY
 Actually cleaner: expose p.SOURCE_SCHEMA directly as a named column.

 Incremental filter block (inserted before macro call):
 {% if is_incremental() %}
   {% set incremental_filter %}
     _FIVETRAN_SYNCED > (select max(_FIVETRAN_SYNCED) from {{ this }})
   {% endset %}
 {% else %}
   {% set incremental_filter = '' %}
 {% endif %}

 with raw_union as (
     {{ generate_shopify_table_union('PRODUCT', incremental_filter) }}
 ),

 3. dbt_project.yml — update staging config

 staging:
   +materialized: incremental
   +incremental_strategy: merge
   +transient: false

 (Individual models can still override with materialized='table' if needed for small/reference tables.)

 Snowflake Task scheduling (Option A — recommended)

 Once models are incremental, runs are fast (seconds for small delta). The Stored Procedure + Task approach still applies:

 1. SP_REFRESH_BF_SHOPIFY_STORES_MASTER — stores master stays materialized='table' (184-row lookup table, always fast to rebuild, needs to stay current)
 2. SP_REFRESH_BF_SHOPIFY_PRODUCT — compiled incremental merge SQL
 3. Tasks: parent (stores master, every 30 min) → child (product, after parent succeeds)
 4. Warehouse: EXPLORATION_WH
 5. Schedule: USING CRON '*/30 * * * *' UTC

 Full-refresh trigger

 When Fivetran adds a new store schema, run locally:
 dbt run --select stg_shopify_product --full-refresh
 Then update the stored procedures with newly compiled SQL.

 Critical files

 - macros/generate_shopify_table_union.sql — add incremental_filter parameter
 - models/staging/stg_shopify_product.sql — switch config, add filter block, expose SOURCE_SCHEMA column
 - dbt_project.yml — update staging materialization default

 Verification

 After first incremental run, wait 30 min and check:
 -- Confirm merge is working (row count should be stable, not doubling)
 select count(*) from DELPHI_EXPLORATION.EXPLORATION_SANDBOX.BF_SHOPIFY_PRODUCT;

 -- Confirm _FIVETRAN_SYNCED range looks right
 select min(_FIVETRAN_SYNCED), max(_FIVETRAN_SYNCED)
 from DELPHI_EXPLORATION.EXPLORATION_SANDBOX.BF_SHOPIFY_PRODUCT;