Embucket vs Snowflake parity harness for dbt-snowplow-web

README.md

@@ -164,3 +164,68 @@ AWS S3 Table Bucket
 - **dbt-embucket** adapter calls Lambda directly via AWS IAM — no public endpoints
 - **Embucket** is a Snowflake-compatible query engine built on Apache DataFusion + Apache Iceberg
 - Data is stored as Iceberg tables in your S3 Table Bucket
+
+## Comparing Embucket vs Snowflake
+
+Runs dbt-snowplow-web on both engines against the same S3 Tables Iceberg
+source (loaded in two 30-minute Athena batches) and diffs the three
+headline derived tables after each run to surface semantic drift.
+
+Prerequisites:
+- `~/.snowflake/connections.toml` has a `[connections.default]` entry with
+  credentials for an account where you have ACCOUNTADMIN (or a role that
+  can create catalog integrations and iceberg tables).
+- IAM role `snowflake-table-bucket-access` has `glue:Get*` on
+  `arn:aws:glue:us-east-2:<account>:catalog/s3tablescatalog/snowplow` and
+  `s3tables:*` on the snowplow bucket.
+- Lake Formation `DESCRIBE` + `SELECT` granted to that role on
+  `s3tablescatalog/snowplow.atomic.events_0416` (grant once after the
+  first `load_from_glue.py init`):
+  ```bash
+  aws lakeformation grant-permissions \
+    --principal DataLakePrincipalIdentifier=arn:aws:iam::767397688925:role/snowflake-table-bucket-access \
+    --resource '{"Table":{"CatalogId":"767397688925:s3tablescatalog/snowplow","DatabaseName":"atomic","Name":"events_0416"}}' \
+    --permissions DESCRIBE SELECT
+  ```
+- `profiles.yml` has a `snowflake` output under `embucket_demo.outputs`
+  (see `profiles.yml.example`).
+
+Run flow:
+
+```bash
+# 1. One-time Snowflake setup (catalog integration + iceberg table)
+uv run python scripts/snowflake_setup.py
+
+# 2. Reset the shared Athena-managed source table
+uv run python scripts/load_from_glue.py init
+
+# 3. Load batch 1
+uv run python scripts/load_from_glue.py insert \
+  --start '2026-04-22 15:00:00' --end '2026-04-22 15:30:00'
+
+# 4. Run dbt on both engines (Snowflake needs a metadata refresh first)
+uv run dbt run --profiles-dir . --target dev
+uv run python scripts/snowflake_refresh.py
+uv run dbt run --profiles-dir . --target snowflake
+
+# 5. Parity check
+uv run python scripts/parity.py   # exits 0 on zero diffs
+
+# 6. Load batch 2 (append, do not re-init)
+uv run python scripts/load_from_glue.py insert \
+  --start '2026-04-22 15:30:00' --end '2026-04-22 16:00:00'
+
+# 7. Second dbt run exercises the incremental path
+uv run dbt run --profiles-dir . --target dev
+uv run python scripts/snowflake_refresh.py
+uv run dbt run --profiles-dir . --target snowflake
+
+# 8. Parity check again
+uv run python scripts/parity.py
+```
+
+A non-zero exit from `parity.py` means the engines produced different
+output on the same input — that's the interesting signal the harness
+exists to surface. Rowcount-only mismatches hint at incremental-window or
+JOIN semantics divergence; hash mismatches with matching rowcounts hint at
+cast, NULL, or ordering divergence.

dbt_project.yml

@@ -10,18 +10,14 @@ dispatch:
     search_order: ['snowplow_utils', 'dbt']
 
 vars:
-  snowplow__start_date: '2026-04-13'
-  snowplow__events: "{{ source('atomic', 'events') }}"
+  snowplow__start_date: '2026-04-22'
   snowplow__atomic_schema: 'atomic'
-  # The fake event generator pins collector_tstamp to a fixed 2h52m window
-  # (generator-time never advances), so snowplow_web's default
-  # snowplow__session_timestamp=collector_tstamp cannot make incremental
-  # progress — every run re-scans the entire events table. load_tstamp *does*
-  # advance (it's the loader commit time), so override to it. atomic.events
-  # must also be partitioned by day(load_tstamp) for pruning to match.
-  snowplow__session_timestamp: load_tstamp
-  # Narrow the incremental lookback from 6h → 1h. With hour() partitioning
-  # on load_tstamp this should reduce the number of partitions Snowplow
-  # reads at the base layer.
-  snowplow__lookback_window_hours: 1
+  snowplow__events_table: "{{ 'events_0416_v' if target.type == 'snowflake' else 'events_0416' }}"
+  # Context flags — set to true once the corresponding context column is
+  # populated (see scripts/ctas_from_glue.sql). web_page_1 is always on.
+  snowplow__enable_iab: false
+  snowplow__enable_ua: false
+  snowplow__enable_yauaa: false
+  snowplow__enable_consent: false
+  snowplow__enable_cwv: false
 

patches/snowplow_utils/macros/base/base_create_snowplow_events_this_run.sql

@@ -0,0 +1,84 @@
+{#
+Copyright (c) 2021-present Snowplow Analytics Ltd. All rights reserved.
+This program is licensed to you under the Snowplow Community License Version 1.0,
+and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
+You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
+
+Patched: the final `qualify row_number() over (partition by event_id ...)` is
+removed. The original QUALIFY operates on the 137-column join output and
+OOMs DataFusion's RepartitionExec on Embucket. Event_id dedup is restored
+via a narrow-scratch pipeline downstream:
+
+  snowplow_web_base_events_raw_this_run      (wide, materialised, may have event_id duplicates)
+  snowplow_web_base_events_winners_this_run  (narrow: 2 columns, one row per event_id)
+  snowplow_web_base_events_this_run          (raw INNER JOIN winners ON (event_id, winner_hash))
+
+Each step materialises as a table, so DataFusion's memory pool resets
+between them; the QUALIFY that picks the winner runs on a 3-column
+projection instead of 137.
+#}
+
+
+{% macro base_create_snowplow_events_this_run(sessions_this_run_table='snowplow_base_sessions_this_run', session_identifiers=[{"schema" : "atomic", "field" : "domain_sessionid"}], session_sql=none, session_timestamp='load_tstamp', derived_tstamp_partitioned=true, days_late_allowed=3, max_session_days=3, app_ids=[], snowplow_events_database=none, snowplow_events_schema='atomic', snowplow_events_table='events', entities_or_sdes=none, custom_sql=none) %}
+    {{ return(adapter.dispatch('base_create_snowplow_events_this_run', 'snowplow_utils')(sessions_this_run_table, session_identifiers, session_sql, session_timestamp, derived_tstamp_partitioned, days_late_allowed, max_session_days, app_ids, snowplow_events_database, snowplow_events_schema, snowplow_events_table, entities_or_sdes, custom_sql)) }}
+{% endmacro %}
+
+{% macro default__base_create_snowplow_events_this_run(sessions_this_run_table, session_identifiers, session_sql, session_timestamp, derived_tstamp_partitioned, days_late_allowed, max_session_days, app_ids, snowplow_events_database, snowplow_events_schema, snowplow_events_table, entities_or_sdes, custom_sql) %}
+    {%- set lower_limit, upper_limit = snowplow_utils.return_limits_from_model(ref(sessions_this_run_table),
+                                                                            'start_tstamp',
+                                                                            'end_tstamp') %}
+    {% set sessions_this_run = ref(sessions_this_run_table) %}
+    {% set snowplow_events = api.Relation.create(database=snowplow_events_database, schema=snowplow_events_schema, identifier=snowplow_events_table) %}
+
+    {% set events_this_run_query %}
+        with identified_events AS (
+            select
+                {% if session_sql %}
+                    {{ session_sql }} as session_identifier,
+                {% else -%}
+                    COALESCE(
+                        {% for identifier in session_identifiers %}
+                            {%- if identifier['schema']|lower != 'atomic' -%}
+                                {{ snowplow_utils.get_field(identifier['schema'], identifier['field'], 'e', dbt.type_string(), 0, snowplow_events) }}
+                            {%- else -%}
+                                e.{{identifier['field']}}
+                            {%- endif -%}
+                            ,
+                        {%- endfor -%}
+                        NULL
+                    ) as session_identifier,
+                {%- endif %}
+                e.*
+                {% if custom_sql %}
+                    , {{ custom_sql }}
+                {% endif %}
+
+            from {{ snowplow_events }} e
+
+        )
+
+        select
+            a.*,
+            b.user_identifier -- take user_identifier from manifest. This ensures only 1 domain_userid per session.
+
+        from identified_events as a
+        inner join {{ sessions_this_run }} as b
+        on a.session_identifier = b.session_identifier
+
+        where a.{{ session_timestamp }} <= {{ snowplow_utils.timestamp_add('day', max_session_days, 'b.start_tstamp') }}
+        and a.dvce_sent_tstamp <= {{ snowplow_utils.timestamp_add('day', days_late_allowed, 'a.dvce_created_tstamp') }}
+        and a.{{ session_timestamp }} >= {{ lower_limit }}
+        and a.{{ session_timestamp }} <= {{ upper_limit }}
+        and a.{{ session_timestamp }} >= b.start_tstamp -- deal with late loading events
+
+        {% if derived_tstamp_partitioned and target.type == 'bigquery' | as_bool() %}
+            and a.derived_tstamp >= {{ snowplow_utils.timestamp_add('hour', -1, lower_limit) }}
+            and a.derived_tstamp <= {{ upper_limit }}
+        {% endif %}
+
+        and {{ snowplow_utils.app_id_filter(app_ids) }}
+    {% endset %}
+
+    {{ return(events_this_run_query) }}
+
+{% endmacro %}

patches/snowplow_web/models/base/scratch/snowflake/snowplow_web_base_events_raw_this_run.sql

@@ -0,0 +1,50 @@
+{#
+Copyright (c) 2020-present Snowplow Analytics Ltd. All rights reserved.
+This program is licensed to you under the Snowplow Community License Version 1.0,
+and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
+You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
+
+Patched new model: upstream's wrapper content (wide projection + derived
+page_view_id / domain_userid columns) but WITHOUT the event_id QUALIFY.
+May contain duplicate event_ids. Deduped downstream via:
+
+    snowplow_web_base_events_winners_this_run  (narrow: event_id + hash)
+    snowplow_web_base_events_this_run          (raw INNER JOIN winners)
+
+Each hop is materialised so DataFusion's memory pool resets between them.
+#}
+
+{{
+  config(
+    materialized='table',
+    tags=["this_run"],
+    sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt'))
+  )
+}}
+
+{% set base_events_query = snowplow_utils.base_create_snowplow_events_this_run(
+    sessions_this_run_table='snowplow_web_base_sessions_this_run',
+    session_identifiers=var('snowplow__session_identifiers', [{"schema" : "atomic", "field" : "domain_sessionid"}]),
+    session_sql=var('snowplow__session_sql', none),
+    session_timestamp=var('snowplow__session_timestamp', 'collector_tstamp'),
+    derived_tstamp_partitioned=var('snowplow__derived_tstamp_partitioned', true),
+    days_late_allowed=var('snowplow__days_late_allowed', 3),
+    max_session_days=var('snowplow__max_session_days', 3),
+    app_ids=var('snowplow__app_id', []),
+    snowplow_events_database=var('snowplow__database', target.database) if target.type not in ['databricks', 'spark'] else var('snowplow__databricks_catalog', 'hive_metastore') if target.type in ['databricks'] else var('snowplow__atomic_schema', 'atomic'),
+    snowplow_events_schema=var('snowplow__atomic_schema', 'atomic'),
+    snowplow_events_table=var('snowplow__events_table', 'events')) %}
+
+with base_query as (
+  {{ base_events_query }}
+)
+
+select
+  a.contexts_com_snowplowanalytics_snowplow_web_page_1[0]:id::varchar as page_view_id,
+  a.session_identifier as domain_sessionid,
+  a.domain_sessionid as original_domain_sessionid,
+  a.user_identifier as domain_userid,
+  a.domain_userid as original_domain_userid,
+  a.* exclude(contexts_com_snowplowanalytics_snowplow_web_page_1, domain_sessionid, domain_userid)
+
+from base_query a

patches/snowplow_web/models/base/scratch/snowflake/snowplow_web_base_events_this_run.sql

@@ -0,0 +1,25 @@
+{#
+Copyright (c) 2020-present Snowplow Analytics Ltd. All rights reserved.
+This program is licensed to you under the Snowplow Community License Version 1.0,
+and you may not use this file except in compliance with the Snowplow Community License Version 1.0.
+You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0
+
+Patched: replaces the upstream monolithic base_events_this_run with a
+3-model narrow-scratch dedup chain. This model is now just the final
+INNER JOIN of the wide raw table against the winners table, producing
+one wide row per event_id.
+#}
+
+{{
+  config(
+    materialized='table',
+    tags=["this_run"],
+    sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt'))
+  )
+}}
+
+select r.*
+from {{ ref('snowplow_web_base_events_raw_this_run') }} r
+inner join {{ ref('snowplow_web_base_events_winners_this_run') }} w
+  on r.event_id = w.event_id
+  and md5(concat_ws('|', r.event_id, cast(r.collector_tstamp as varchar), cast(r.dvce_created_tstamp as varchar), cast(r.load_tstamp as varchar))) = w.winner_hash

patches/snowplow_web/models/base/scratch/snowflake/snowplow_web_base_events_winners_this_run.sql

@@ -0,0 +1,31 @@
+{#
+Narrow-projection event_id dedup. Two columns buffered per partition
+(event_id + winner_hash); the QUALIFY is spillable.
+
+winner_hash is computed as HASH(event_id, collector_tstamp,
+dvce_created_tstamp, load_tstamp) so the raw table's inner join back on
+(event_id, winner_hash) picks exactly one source row per event_id, even
+when two events share the same (collector_tstamp, dvce_created_tstamp).
+#}
+
+{{
+  config(
+    materialized='table',
+    tags=["this_run"],
+    sql_header=snowplow_utils.set_query_tag(var('snowplow__query_tag', 'snowplow_dbt'))
+  )
+}}
+
+select event_id, winner_hash
+from (
+  select
+    event_id,
+    md5(concat_ws('|', event_id, cast(collector_tstamp as varchar), cast(dvce_created_tstamp as varchar), cast(load_tstamp as varchar))) as winner_hash,
+    collector_tstamp,
+    dvce_created_tstamp
+  from {{ ref('snowplow_web_base_events_raw_this_run') }}
+)
+qualify row_number() over (
+  partition by event_id
+  order by collector_tstamp, dvce_created_tstamp, winner_hash
+) = 1