Full-Stack Data Analytics and Data Warehousing Support

SQL · warehouse DDL with surrogate keys and a tested additivity rule

-- Conformed dimension with SCD Type 2 history (ANSI-style DDL; Snowflake shown)
CREATE TABLE IF NOT EXISTS dw.dim_customer (
    customer_key      BIGINT        NOT NULL,   -- surrogate key
    customer_id       VARCHAR(64)   NOT NULL,   -- natural / business key
    customer_name     VARCHAR(256),
    segment           VARCHAR(32),
    country_code      CHAR(2),
    scd2_valid_from   TIMESTAMP_NTZ NOT NULL,
    scd2_valid_to     TIMESTAMP_NTZ NOT NULL DEFAULT '9999-12-31 00:00:00'::TIMESTAMP_NTZ,
    scd2_is_current   BOOLEAN       NOT NULL DEFAULT TRUE,
    row_hash          VARCHAR(64)   NOT NULL,   -- change detection
    CONSTRAINT pk_dim_customer PRIMARY KEY (customer_key)   -- informational in Snowflake
);

-- Fact at ONE grain: the order line. Never mix grains in one fact table.
CREATE TABLE IF NOT EXISTS dw.fact_sales (
    sale_id          VARCHAR(64)    NOT NULL,
    date_key         INTEGER        NOT NULL,
    customer_key     BIGINT         NOT NULL,
    product_key      BIGINT         NOT NULL,
    store_key        BIGINT         NOT NULL,
    order_line_id    VARCHAR(64)    NOT NULL,   -- degenerate dimension
    quantity         NUMBER(18,3)   NOT NULL,
    gross_amount     NUMBER(18,4)   NOT NULL,
    discount_amount  NUMBER(18,4)   NOT NULL DEFAULT 0,
    net_amount       NUMBER(18,4)   NOT NULL,
    margin_amount    NUMBER(18,4),
    dw_loaded_at     TIMESTAMP_NTZ  NOT NULL,
    CONSTRAINT pk_fact_sales PRIMARY KEY (sale_id),
    CONSTRAINT fk_fact_sales_customer FOREIGN KEY (customer_key) REFERENCES dw.dim_customer (customer_key)
)
CLUSTER BY (date_key, store_key);

-- Snowflake does not enforce PK/FK/CHECK, so additivity is tested, not assumed (dbt test or scheduled check):
SELECT COUNT(*) AS broken_rows
FROM dw.fact_sales
WHERE net_amount <> gross_amount - discount_amount;

Cloud warehouses such as Snowflake and BigQuery treat primary, foreign and check constraints as informational, so data warehousing support turns every modelling rule into a test that runs on each load. The same logical contract is then materialised per engine. On ClickHouse the fact becomes a sorted, compressed MergeTree with a projection for the most-used dashboard filter and a roll-up the dashboard reads instead of raw rows.

SQL · ClickHouse physical design: partitioning, sort order, codecs and projections

CREATE TABLE analytics.fact_sales
(
    event_date        Date,
    event_time        DateTime64(3, 'UTC'),
    store_id          UInt32   CODEC(T64, ZSTD(3)),
    product_id        UInt32   CODEC(T64, ZSTD(3)),
    customer_id       UInt64   CODEC(T64, ZSTD(3)),
    channel           LowCardinality(String),
    quantity          Decimal(18,3),
    net_amount        Decimal(18,4) CODEC(ZSTD(3)),
    ingested_at       DateTime  DEFAULT now(),
    PROJECTION proj_store_daily
    (
        SELECT store_id, event_date, sum(net_amount), sum(quantity)
        GROUP BY store_id, event_date
    )
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(event_date)
ORDER BY (store_id, event_date, product_id)
TTL event_date + INTERVAL 36 MONTH TO VOLUME 'cold',
    event_date + INTERVAL 84 MONTH DELETE
SETTINGS index_granularity = 8192,
         min_bytes_for_wide_part = 10485760;

-- Incremental roll-up so dashboards never scan raw rows
CREATE MATERIALIZED VIEW analytics.mv_store_daily
ENGINE = SummingMergeTree
PARTITION BY toYYYYMM(event_date)
ORDER BY (store_id, event_date)
AS SELECT store_id, event_date, sum(net_amount) AS net_amount, sum(quantity) AS quantity
FROM analytics.fact_sales
GROUP BY store_id, event_date;

SQL · dbt incremental model with a late-arrival window and point-in-time SCD2 joins

-- models/marts/fact_sales.sql  (dbt-snowflake 1.8+; delete+insert keeps late rows idempotent)
{{ config(
    materialized         = 'incremental',
    incremental_strategy = 'delete+insert',
    unique_key           = 'sale_id',
    cluster_by           = ['date_key', 'store_key'],
    on_schema_change     = 'append_new_columns',
    tags                 = ['mart', 'revenue']
) }}

WITH orders AS (
    SELECT
        o.order_line_id,
        o.order_date,
        o.customer_id,
        o.store_code,
        o.product_key,
        o.quantity,
        o.gross_amount,
        o.discount_amount,
        o.gross_amount - o.discount_amount AS net_amount,
        o._ingested_at
    FROM {{ ref('stg_sales__order_lines') }} AS o
    {% if is_incremental() %}
    -- 3-day trailing window: late-arriving events self-heal on the next run
    WHERE o.order_date >= (SELECT DATEADD('day', -3, MAX(order_date)) FROM {{ this }})
    {% endif %}
)

SELECT
    {{ dbt_utils.generate_surrogate_key(['o.order_line_id']) }} AS sale_id,
    d.date_key,
    c.customer_key,
    o.product_key,
    s.store_key,
    o.order_line_id,
    o.order_date,
    o.quantity,
    o.gross_amount,
    o.discount_amount,
    o.net_amount,
    o.net_amount - (o.quantity * p.unit_cost)                  AS margin_amount,
    o._ingested_at                                             AS dw_loaded_at
FROM orders AS o
JOIN {{ ref('dim_date') }}     AS d ON d.calendar_date = o.order_date
JOIN {{ ref('dim_customer') }} AS c ON c.customer_id   = o.customer_id
                                   AND o.order_date >= c.scd2_valid_from
                                   AND o.order_date <  c.scd2_valid_to     -- point-in-time, not "current"
JOIN {{ ref('dim_store') }}    AS s ON s.store_code    = o.store_code
JOIN {{ ref('dim_product') }}  AS p ON p.product_key   = o.product_key AND p.scd2_is_current

For data warehousing support reviews, two details in the model matter more than they look. The three-day trailing window lets late events self-heal without a full refresh, and the customer join is point-in-time against scd2_valid_from and scd2_valid_to, so a customer's current segment never rewrites last year's revenue. Contracts and tests then block a bad deploy in CI, per the dbt model contracts documentation.

YAML · dbt contracts, tests and freshness SLAs that block a bad deploy

version: 2

sources:
  - name: raw
    database: analytics_raw
    freshness:
      warn_after:  {count: 30, period: minute}
      error_after: {count: 90, period: minute}
    loaded_at_field: _ingested_at
    tables:
      - name: sales_stream
        columns:
          - name: order_line_id
            tests: [not_null, unique]

models:
  - name: fact_sales
    description: "Revenue fact at order-line grain. Owner: analytics-platform@minervadb.com"
    config:
      contract: {enforced: true}
    columns:
      - name: sale_id
        data_type: varchar
        constraints: [{type: not_null}, {type: primary_key}]
        tests: [unique, not_null]
      - name: customer_key
        data_type: bigint
        tests:
          - relationships: {to: ref('dim_customer'), field: customer_key}
      - name: net_amount
        data_type: numeric(18,4)
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= -1000000"
          - dbt_expectations.expect_column_values_to_not_be_null
    tests:
      - dbt_utils.equal_rowcount:
          compare_model: ref('stg_sales__order_lines')
      - dbt_utils.recency:
          datepart: hour
          field: order_date
          interval: 24

The pattern alongside classifies a column once and lets policy follow the tag, rather than multiplying secure views. IS_ROLE_IN_SESSION respects role hierarchies, which CURRENT_ROLE() comparisons do not, and the row-policy argument is named differently from the column it protects to avoid silent shadowing. Equivalent controls exist on BigQuery (policy tags, row-level security), Redshift (dynamic data masking, RLS) and Databricks Unity Catalog.

In data warehousing support, evidence matters as much as enforcement. Access history answers "who read this PII column" with a query, not a spreadsheet exercise, and this work aligns with our database auditing, privacy and security practice.

SQL · tag-based classification, masking, row access policy and audit evidence

-- 1. Classify once, enforce everywhere (Snowflake Enterprise Edition features)
CREATE TAG IF NOT EXISTS governance.data_sensitivity
    ALLOWED_VALUES 'public', 'internal', 'confidential', 'pii', 'phi';

ALTER TABLE dw.dim_customer MODIFY COLUMN email_address
    SET TAG governance.data_sensitivity = 'pii';

-- 2. Tag-based masking: every STRING column tagged pii is masked by role
CREATE OR REPLACE MASKING POLICY governance.mask_pii_string AS (val STRING)
RETURNS STRING ->
    CASE
        WHEN IS_ROLE_IN_SESSION('DATA_PROTECTION_OFFICER') THEN val
        WHEN IS_ROLE_IN_SESSION('ANALYST')                  THEN REGEXP_REPLACE(val, '^[^@]+', '****')
        ELSE '***MASKED***'
    END;

ALTER TAG governance.data_sensitivity SET MASKING POLICY governance.mask_pii_string;

-- 3. Row access policy for residency (argument named to avoid shadowing the column)
CREATE OR REPLACE ROW ACCESS POLICY governance.region_rap AS (p_country CHAR(2))
RETURNS BOOLEAN ->
    EXISTS (
        SELECT 1
        FROM governance.role_region_map AS m
        WHERE IS_ROLE_IN_SESSION(m.role_name)
          AND m.country_code IN (p_country, 'ALL')
    );

ALTER TABLE dw.dim_customer ADD ROW ACCESS POLICY governance.region_rap ON (country_code);

-- 4. Evidence: who read the PII column in the last 30 days (ACCOUNT_USAGE lags by up to ~3 hours)
SELECT
    ah.user_name,
    ah.query_start_time,
    obj.value:"objectName"::STRING AS object_name
FROM snowflake.account_usage.access_history AS ah,
     LATERAL FLATTEN(input => ah.base_objects_accessed) AS obj,
     LATERAL FLATTEN(input => obj.value:"columns")      AS col
WHERE col.value:"columnName"::STRING = 'EMAIL_ADDRESS'
  AND ah.query_start_time >= DATEADD('day', -30, CURRENT_TIMESTAMP())
ORDER BY ah.query_start_time DESC;