Data Integration Guide

SELECT
    snapshot_id,
    snapshot_created_at,
    -- floor the pull range to a 36h lookback so steady-state scans stay bounded;
    -- older changes are caught by the backfill job (Job 2)
    GREATEST(
        snapshot_loaded_intervals__min_timestamp,
        CURRENT_TIMESTAMP - INTERVAL '36 hours'
    ) AS pull_from_block_timestamp,
    snapshot_loaded_intervals__max_timestamp,
    snapshot_loaded_intervals__loads,
    is_full_refresh
FROM delivery_metadata.snapshots.aggregated
WHERE chain = 'base'
  AND table_name = 'base.raw.logs'
  AND snapshot_created_at >= <last_pipeline_run_time> - INTERVAL '1 hour'

SELECT
    chain,
    table_name,
    hour,
    last_updated_at,
    last_full_refresh_timestamp
FROM delivery_metadata.intervals.changes
WHERE chain = 'base'
  AND table_name = 'base.raw.logs'
  AND last_updated_at >= CURRENT_TIMESTAMP - INTERVAL '1 day'
ORDER BY hour

SELECT verified_until
FROM delivery_metadata.intervals.verified_until
WHERE chain = 'ethereum'
  AND table_name = 'ethereum.raw.logs'

interval   1    2    3    4    5    6    7
verified   ✓    ✓    ✓    ✗    ✓    ✓    ✓
                      └─ verified_until = end of interval 3
                           │
                           └ interval 4 is the first gap; intervals 5–7
                             are verified but NOT contiguous with the
                             start, so verified_until does not advance.

A window that starts with a gap (interval 1 = ✗ or missing) yields no row:

interval   1    2    3    4
verified   ✗    ✓    ✓    ✓     →  no verified_until row

-- models/staging/stg_base_raw_logs.sql
{{ config(
    materialized='incremental',
    incremental_strategy='delete+insert',
    unique_key='block_hour'
) }}

-- block_hour is the hourly partition, NOT a row key: delete+insert on it
-- replaces every changed hour wholesale, so upstream deletions are applied too
SELECT
    *,
    DATE_TRUNC('hour', block_timestamp) AS block_hour
FROM {{ source('allium', 'base_raw_logs') }}

{% if is_incremental() %}
WHERE block_timestamp >= (
    SELECT MIN(snapshot_loaded_intervals__min_timestamp)
    FROM {{ source('allium_delivery_metadata', 'snapshots_aggregated') }}
    WHERE chain = 'base'
      AND table_name = 'base.raw.logs'
      AND snapshot_created_at >= (
          SELECT MAX(_loaded_at) FROM {{ this }}
      ) - INTERVAL '1 hour'
)
{% endif %}