Full-Stack Data Analytics and Data Warehousing Support

• What full-stack analytics and warehousing support includes
• Reference architecture of a modern analytics platform
• Warehouses, lakehouses and query engines we support
• Dimensional modelling and physical warehouse design
• Ingestion, CDC and streaming pipelines
• Transformation, dbt and the semantic layer
• Query performance engineering and concurrency
• Observability, SLOs, data quality and FinOps
• Security, governance and compliance
• Data Analytics and Data Warehousing Support plans
• How a MinervaDB engagement works
• Why enterprises choose MinervaDB
• Frequently asked questions
• Talk to a data warehousing expert

Figure 1: MinervaDB reference architecture for full-stack Data Analytics and Data Warehousing Support.

Figure 2: conformed star schema with surrogate keys and Type 2 history, reviewed in every MinervaDB Data Analytics and Data Warehousing Support design audit.

-- Conformed dimension with SCD Type 2 history
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    NOT NULL,
    scd2_valid_to     TIMESTAMP    NOT NULL DEFAULT TIMESTAMP '9999-12-31 00:00:00',
    scd2_is_current   BOOLEAN      NOT NULL DEFAULT TRUE,
    row_hash          VARCHAR(64)  NOT NULL,          -- change detection
    dw_loaded_at      TIMESTAMP    NOT NULL DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT pk_dim_customer PRIMARY KEY (customer_key)
);

-- Fact table: grain is ONE ORDER LINE. Never mix grains in one fact.
CREATE TABLE IF NOT EXISTS dw.fact_sales (
    sale_id           BIGINT       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          NUMERIC(18,3) NOT NULL,
    gross_amount      NUMERIC(18,4) NOT NULL,
    discount_amount   NUMERIC(18,4) NOT NULL DEFAULT 0,
    net_amount        NUMERIC(18,4) NOT NULL,
    margin_amount     NUMERIC(18,4),
    dw_batch_id       BIGINT       NOT NULL,
    CONSTRAINT pk_fact_sales PRIMARY KEY (sale_id)
)
CLUSTER BY (date_key, store_key);                     -- Snowflake / BigQuery layout

-- Additivity guard: net_amount must always reconcile
ALTER TABLE dw.fact_sales
  ADD CONSTRAINT ck_fact_sales_net
  CHECK (net_amount = gross_amount - discount_amount);

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;

Figure 3: the ingestion path MinervaDB hardens during Data Analytics and Data Warehousing Support onboarding, with schema contracts, dead-letter handling and replayable history.

{
  "name": "pg-sales-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "pg-primary.internal",
    "database.dbname": "sales",
    "slot.name": "dbz_sales_slot",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.orders,public.order_lines,public.customers",
    "topic.prefix": "sales",
    "snapshot.mode": "initial",
    "incremental.snapshot.chunk.size": 20480,
    "heartbeat.interval.ms": 10000,
    "heartbeat.action.query": "UPDATE dbz.heartbeat SET ts = now()",
    "decimal.handling.mode": "precise",
    "time.precision.mode": "adaptive_time_microseconds",
    "tombstones.on.delete": "true",
    "producer.override.compression.type": "zstd",
    "producer.override.acks": "all",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.sales",
    "errors.deadletterqueue.context.headers.enable": "true",
    "transforms": "route,unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.add.fields": "op,source.lsn,source.ts_ms",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

SELECT slot_name,
       active,
       wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn))  AS retained_wal,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS unflushed,
       safe_wal_size
FROM   pg_replication_slots
WHERE  slot_type = 'logical'
ORDER  BY pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) DESC;

-- Alert thresholds we deploy by default:
--   WARNING  retained_wal > 10 GB or wal_status = 'extended'
--   CRITICAL retained_wal > 40 GB or wal_status IN ('unreserved','lost')
--   CRITICAL slot inactive for more than 5 minutes

from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.providers.common.sql.operators.sql import SQLCheckOperator

DEFAULT_ARGS = {
    "owner": "minervadb-analytics",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(hours=2),
}

@dag(
    dag_id="warehouse_sales_incremental",
    schedule="*/30 * * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,               # protects the warehouse from run pile-up
    default_args=DEFAULT_ARGS,
    tags=["warehouse", "incremental", "minervadb"],
)
def warehouse_sales_incremental():

    @task
    def resolve_watermark(data_interval_start=None) -> str:
        """Never trust wall-clock time: read the last committed watermark."""
        from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
        hook = SnowflakeHook(snowflake_conn_id="dw")
        low = hook.get_first(
            "SELECT COALESCE(MAX(dw_loaded_at), '1970-01-01') FROM dw.fact_sales"
        )[0]
        return low.isoformat()

    @task
    def merge_increment(watermark: str) -> int:
        """MERGE is idempotent, so a retried task cannot double-count revenue."""
        from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
        hook = SnowflakeHook(snowflake_conn_id="dw")
        rows = hook.run(
            """
            MERGE INTO dw.fact_sales AS t
            USING (
                SELECT s.sale_id, d.date_key, c.customer_key, p.product_key,
                       st.store_key, s.order_line_id, s.quantity,
                       s.gross_amount, s.discount_amount, s.net_amount,
                       s.updated_at AS dw_loaded_at
                FROM   raw.sales_stream      s
                JOIN   dw.dim_date           d  ON d.calendar_date = s.order_date
                JOIN   dw.dim_customer       c  ON c.customer_id = s.customer_id
                                                AND c.scd2_is_current
                JOIN   dw.dim_product        p  ON p.product_key = s.product_key
                JOIN   dw.dim_store          st ON st.store_code = s.store_code
                WHERE  s.updated_at > %(watermark)s
                  AND  s._op != 'd'
            ) AS s
            ON t.sale_id = s.sale_id
            WHEN MATCHED THEN UPDATE SET
                 t.quantity = s.quantity, t.net_amount = s.net_amount,
                 t.dw_loaded_at = s.dw_loaded_at
            WHEN NOT MATCHED THEN INSERT VALUES (
                 s.sale_id, s.date_key, s.customer_key, s.product_key,
                 s.store_key, s.order_line_id, s.quantity, s.gross_amount,
                 s.discount_amount, s.net_amount, NULL, HASH(s.sale_id))
            """,
            parameters={"watermark": watermark},
            handler=lambda cur: cur.rowcount,
        )
        return rows

    freshness_gate = SQLCheckOperator(
        task_id="freshness_sla_gate",
        conn_id="dw",
        sql="""
            SELECT TIMESTAMPDIFF('minute', MAX(dw_loaded_at), CURRENT_TIMESTAMP()) < 45
            FROM dw.fact_sales
        """,
    )

    reconciliation_gate = SQLCheckOperator(
        task_id="source_to_warehouse_reconciliation",
        conn_id="dw",
        sql="""
            WITH src AS (SELECT COUNT(*) c FROM raw.sales_stream WHERE _op != 'd'),
                 dwh AS (SELECT COUNT(*) c FROM dw.fact_sales)
            SELECT ABS(src.c - dwh.c) / NULLIF(src.c, 0) < 0.001 FROM src, dwh
        """,
    )

    merge_increment(resolve_watermark()) >> freshness_gate >> reconciliation_gate

warehouse_sales_incremental()

{{ config(
    materialized      = 'incremental',
    incremental_strategy = 'insert_overwrite',
    partition_by      = {'field': 'order_date', 'data_type': 'date', 'granularity': 'day'},
    cluster_by        = ['store_key', 'product_key'],
    on_schema_change  = 'append_new_columns',
    tags              = ['mart', 'revenue']
) }}

WITH bounds AS (
    /* Reprocess a 3-day trailing window so late events self-heal */
    SELECT DATEADD('day', -3, COALESCE(MAX(order_date), '1970-01-01')) AS lower_bound
    FROM   {{ this }}
    {% if not is_incremental() %} WHERE FALSE {% endif %}
),

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
    FROM {{ ref('stg_sales__order_lines') }} o
    {% if is_incremental() %}
    WHERE o.order_date >= (SELECT lower_bound FROM bounds)
    {% 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
FROM orders o
JOIN {{ ref('dim_date') }}     d ON d.calendar_date = o.order_date
JOIN {{ ref('dim_customer') }} c ON c.customer_id = o.customer_id AND c.scd2_is_current
JOIN {{ ref('dim_store') }}    s ON s.store_code  = o.store_code
JOIN {{ ref('dim_product') }}  p ON p.product_key = o.product_key AND p.scd2_is_current

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

SELECT
    query_hash,
    ANY_VALUE(LEFT(query_text, 120))                       AS sample_sql,
    COUNT(*)                                               AS executions,
    ROUND(SUM(total_elapsed_time) / 1000 / 60, 1)          AS total_minutes,
    ROUND(AVG(total_elapsed_time) / 1000, 2)               AS avg_seconds,
    ROUND(SUM(bytes_scanned) / POWER(1024, 4), 3)          AS tb_scanned,
    ROUND(AVG(percentage_scanned_from_cache), 1)           AS pct_from_cache,
    SUM(bytes_spilled_to_remote_storage)                   AS remote_spill,
    ROUND(SUM(credits_used_cloud_services), 3)             AS cloud_credits
FROM   snowflake.account_usage.query_history
WHERE  start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
  AND  execution_status = 'SUCCESS'
GROUP  BY query_hash
HAVING total_minutes > 5
ORDER  BY total_minutes DESC
LIMIT  25;

-- The same triage on ClickHouse:
SELECT normalized_query_hash,
       count()                                             AS executions,
       round(avg(query_duration_ms))                       AS avg_ms,
       formatReadableSize(sum(read_bytes))                 AS read_total,
       round(sum(read_rows) / 1e9, 2)                      AS billion_rows,
       formatReadableSize(max(memory_usage))               AS peak_memory
FROM   system.query_log
WHERE  type = 'QueryFinish' AND event_time > now() - INTERVAL 7 DAY
GROUP  BY normalized_query_hash
ORDER  BY sum(query_duration_ms) DESC
LIMIT  25;

-- One row per certified mart: freshness, volume anomaly and null drift
WITH observed AS (
    SELECT 'dw.fact_sales' AS object_name,
           MAX(dw_loaded_at)                                AS last_loaded_at,
           COUNT(*)                                         AS row_count,
           COUNT_IF(customer_key IS NULL) / NULLIF(COUNT(*),0) AS null_customer_rate
    FROM   dw.fact_sales
    WHERE  dw_loaded_at >= DATEADD('day', -1, CURRENT_TIMESTAMP())
),
baseline AS (
    SELECT object_name,
           AVG(row_count)                                   AS mean_rows,
           STDDEV_POP(row_count)                            AS sd_rows
    FROM   monitoring.mart_volume_history
    WHERE  observed_on >= DATEADD('day', -28, CURRENT_DATE())
    GROUP  BY object_name
)
SELECT o.object_name,
       TIMESTAMPDIFF('minute', o.last_loaded_at, CURRENT_TIMESTAMP()) AS staleness_minutes,
       o.row_count,
       ROUND((o.row_count - b.mean_rows) / NULLIF(b.sd_rows, 0), 2)   AS volume_z_score,
       ROUND(o.null_customer_rate * 100, 3)                           AS null_customer_pct,
       CASE
         WHEN TIMESTAMPDIFF('minute', o.last_loaded_at, CURRENT_TIMESTAMP()) > 45 THEN 'PAGE_ONCALL'
         WHEN ABS((o.row_count - b.mean_rows) / NULLIF(b.sd_rows, 0)) > 3        THEN 'PAGE_ONCALL'
         WHEN o.null_customer_rate > 0.001                                       THEN 'WARN'
         ELSE 'OK'
       END                                                            AS action
FROM   observed o
JOIN   baseline b USING (object_name);

-- 1. Classify once, enforce everywhere
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. Column-level dynamic masking driven by role, not by view sprawl
CREATE OR REPLACE MASKING POLICY governance.mask_email AS (val STRING)
  RETURNS STRING ->
    CASE
      WHEN CURRENT_ROLE() IN ('DATA_PROTECTION_OFFICER', 'ANALYTICS_ADMIN') THEN val
      WHEN CURRENT_ROLE() IN ('ANALYST_ROLE')
        THEN REGEXP_REPLACE(val, '^[^@]+', '****')
      ELSE '***MASKED***'
    END;

ALTER TAG governance.data_sensitivity
  SET MASKING POLICY governance.mask_email FOR STRING;

-- 3. Row-level security for regional data residency
CREATE OR REPLACE ROW ACCESS POLICY governance.region_rap AS (country_code CHAR(2))
  RETURNS BOOLEAN ->
    EXISTS (
      SELECT 1 FROM governance.role_region_map m
      WHERE m.role_name = CURRENT_ROLE()
        AND (m.country_code = country_code OR m.country_code = 'ALL')
    );

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

-- 4. Prove it: who touched PII in the last 30 days
SELECT user_name, role_name, query_start_time, LEFT(query_text, 100) AS statement
FROM   snowflake.account_usage.access_history a,
       LATERAL FLATTEN(input => a.base_objects_accessed) b
WHERE  b.value:"columns"[0]:"columnName"::STRING = 'EMAIL_ADDRESS'
  AND  query_start_time >= DATEADD('day', -30, CURRENT_TIMESTAMP())
ORDER  BY query_start_time DESC;

Figure 4: the MinervaDB onboarding path for Data Analytics and Data Warehousing Support engagements.

Get a written Data Analytics and Data Warehousing Support assessment in three business days

Tell us which engines you run, where the pain is and what your reporting deadlines look like. A MinervaDB principal engineer will review your architecture, quantify the risk and cost exposure, and show you exactly what full-stack Data Analytics and Data Warehousing Support would change. No obligation, no scripted sales call.

Book an appointment Contact the support desk