Tutorial — Convert BTEQ Script to dbt Model¶

.LOGON tdserver/etl_user,${ETL_PASSWORD}

.SET WIDTH 200
.SET ERRORLEVEL SEVERITY 8

DATABASE production;

/*----------------------------------------------------------
  DAILY REVENUE SUMMARY
  Runs daily at 06:00 via cron
  Owner: finance-data-team@company.com
  Dependencies: orders, customers, products, regions
----------------------------------------------------------*/

/* Step 1: Drop and recreate volatile staging table */
DROP TABLE tmp_daily_orders;
CREATE VOLATILE TABLE tmp_daily_orders AS (
    SELECT
        o.order_id,
        o.customer_id,
        o.order_date,
        o.amount,
        o.discount,
        o.amount - o.discount AS net_amount,
        c.customer_name,
        c.customer_segment,
        p.product_category,
        r.region_name,
        r.territory
    FROM orders o
    INNER JOIN customers c ON o.customer_id = c.customer_id
    INNER JOIN products p ON o.product_id = p.product_id
    INNER JOIN regions r ON o.region_id = r.region_id
    WHERE o.order_date = CURRENT_DATE - 1
      AND o.status = 'COMPLETED'
) WITH DATA
PRIMARY INDEX (order_id)
ON COMMIT PRESERVE ROWS;

.IF ERRORCODE <> 0 THEN .GOTO ERROR_EXIT

/* Step 2: Collect statistics on staging table */
COLLECT STATISTICS ON tmp_daily_orders COLUMN (order_date);
COLLECT STATISTICS ON tmp_daily_orders COLUMN (customer_segment);
COLLECT STATISTICS ON tmp_daily_orders COLUMN (region_name);

/* Step 3: Delete existing data for the date (idempotent reload) */
DELETE FROM daily_revenue_summary
WHERE report_date = CURRENT_DATE - 1;

.IF ERRORCODE <> 0 THEN .GOTO ERROR_EXIT

/* Step 4: Insert aggregated results */
INSERT INTO daily_revenue_summary
SELECT
    CURRENT_DATE - 1 AS report_date,
    t.region_name,
    t.territory,
    t.customer_segment,
    t.product_category,
    COUNT(*) AS order_count,
    COUNT(DISTINCT t.customer_id) AS unique_customers,
    SUM(t.amount) AS gross_revenue,
    SUM(t.discount) AS total_discount,
    SUM(t.net_amount) AS net_revenue,
    AVG(t.net_amount) AS avg_order_value,
    CURRENT_TIMESTAMP AS loaded_at
FROM tmp_daily_orders t
GROUP BY 2, 3, 4, 5;

.IF ERRORCODE <> 0 THEN .GOTO ERROR_EXIT

/* Step 5: Update control table */
UPDATE etl_control
SET last_run = CURRENT_TIMESTAMP,
    row_count = (SELECT COUNT(*) FROM daily_revenue_summary WHERE report_date = CURRENT_DATE - 1),
    status = 'SUCCESS'
WHERE job_name = 'daily_revenue_summary';

/* Step 6: Export report to file for downstream systems */
.EXPORT FILE=daily_revenue_${DATE}.csv
SELECT * FROM daily_revenue_summary
WHERE report_date = CURRENT_DATE - 1
ORDER BY region_name, territory, customer_segment;
.EXPORT RESET

.LOGOFF
.QUIT 0

.LABEL ERROR_EXIT
.EXPORT FILE=daily_revenue_error_${DATE}.log
SELECT 'ERROR: daily_revenue_summary failed at ' || CAST(CURRENT_TIMESTAMP AS VARCHAR(30));
.EXPORT RESET

UPDATE etl_control
SET status = 'FAILED', last_run = CURRENT_TIMESTAMP
WHERE job_name = 'daily_revenue_summary';

.LOGOFF
.QUIT 8

-- Replaces: CREATE VOLATILE TABLE tmp_daily_orders
-- Source: daily_revenue_summary.bteq (Step 1)

{{ config(materialized='ephemeral') }}

SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount,
    o.discount,
    o.amount - o.discount AS net_amount,
    c.customer_name,
    c.customer_segment,
    p.product_category,
    r.region_name,
    r.territory
FROM {{ source('teradata_raw', 'orders') }} o
INNER JOIN {{ source('teradata_raw', 'customers') }} c
    ON o.customer_id = c.customer_id
INNER JOIN {{ source('teradata_raw', 'products') }} p
    ON o.product_id = p.product_id
INNER JOIN {{ source('teradata_raw', 'regions') }} r
    ON o.region_id = r.region_id
WHERE o.status = 'COMPLETED'
{% if is_incremental() %}
    AND o.order_date >= DATE_SUB(CURRENT_DATE(), 3)
{% endif %}

-- Replaces: daily_revenue_summary.bteq (Steps 3-4)
-- Original: DELETE FROM daily_revenue_summary WHERE report_date = CURRENT_DATE - 1;
--           INSERT INTO daily_revenue_summary SELECT ...
-- Converted to: dbt incremental model with merge strategy

{{ config(
    materialized='incremental',
    unique_key=['report_date', 'region_name', 'territory', 'customer_segment', 'product_category'],
    incremental_strategy='merge',
    file_format='delta',
    partition_by=['report_date'],
    post_hook=[
        "OPTIMIZE {{ this }} ZORDER BY (region_name, customer_segment)"
    ]
) }}

WITH daily_orders AS (
    SELECT * FROM {{ ref('stg_daily_orders') }}
)

SELECT
    order_date AS report_date,
    region_name,
    territory,
    customer_segment,
    product_category,
    COUNT(*) AS order_count,
    COUNT(DISTINCT customer_id) AS unique_customers,
    SUM(amount) AS gross_revenue,
    SUM(discount) AS total_discount,
    SUM(net_amount) AS net_revenue,
    AVG(net_amount) AS avg_order_value,
    CURRENT_TIMESTAMP() AS loaded_at
FROM daily_orders
GROUP BY
    order_date,
    region_name,
    territory,
    customer_segment,
    product_category

-- Validate that daily revenue totals are within expected bounds
-- This replaces manual reconciliation against Teradata output

WITH daily_totals AS (
    SELECT
        report_date,
        SUM(gross_revenue) AS total_gross,
        SUM(net_revenue) AS total_net,
        SUM(order_count) AS total_orders
    FROM {{ ref('daily_revenue_summary') }}
    WHERE report_date >= DATE_SUB(CURRENT_DATE(), 7)
    GROUP BY report_date
)

SELECT *
FROM daily_totals
WHERE total_gross <= 0                    -- No revenue = problem
   OR total_net > total_gross             -- Net > gross = logic error
   OR total_orders < 100                  -- Suspiciously low order count
   OR total_gross > 100000000             -- Suspiciously high ($100M/day)

-- During parallel-run period: compare dbt output with Teradata output
-- Remove this test after Teradata decommission

{% if var('parallel_run', false) %}

WITH teradata_totals AS (
    SELECT report_date, SUM(net_revenue) AS td_net
    FROM {{ source('teradata_raw', 'daily_revenue_summary_td') }}
    WHERE report_date = DATE_SUB(CURRENT_DATE(), 1)
    GROUP BY report_date
),
azure_totals AS (
    SELECT report_date, SUM(net_revenue) AS az_net
    FROM {{ ref('daily_revenue_summary') }}
    WHERE report_date = DATE_SUB(CURRENT_DATE(), 1)
    GROUP BY report_date
)

SELECT
    t.report_date,
    t.td_net,
    a.az_net,
    ABS(t.td_net - a.az_net) AS diff
FROM teradata_totals t
JOIN azure_totals a ON t.report_date = a.report_date
WHERE ABS(t.td_net - a.az_net) > 0.01  -- Allow 1 cent rounding

{% else %}
-- Parallel run disabled; skip reconciliation
SELECT 1 WHERE FALSE
{% endif %}