Chapter 10 · Engine Features

The Data Quality Gate

Loading data is not the same as publishing it. The gate runs after the loads and the index maintenance, and before anything is marked as published. Every rule produces a count of failing rows, every result is stored, and a single failing error rule stops the run.

The rules

The audit table also has a warning severity: a rule with Severity = 'warning' is recorded and reported but never stops the run. Use it for things a person should look at, not for things that make the data wrong.

/* dq_gate - data quality gate. Every rule is checked and written to audit.DQResult.
   Severity error   -> the task fails, publish is skipped and the watermark is NOT advanced
   DQ05 reconciles the warehouse against the live source system. */
SET NOCOUNT ON;

DECLARE @BatchID int = (SELECT BatchID FROM audit.BatchRun WHERE DagRunID = N'{{ run_id }}');
DECLARE @rules TABLE (RuleID varchar(10), RuleName nvarchar(200), Severity varchar(10), FailedRows bigint);

-- DQ05 inputs: in-scope online orders in the source vs the fact table
DECLARE @src_orders bigint, @src_total money, @dw_orders bigint, @dw_total money;
SELECT @src_orders = COUNT_BIG(*), @src_total = SUM(h.SubTotal)
FROM AdventureWorks2022.Sales.SalesOrderHeader AS h
WHERE h.OnlineOrderFlag = 1
  AND EXISTS (SELECT 1 FROM Dimension.Customer AS c WHERE c.CustomerAlternateKey = h.CustomerID);
SELECT @dw_orders = COUNT_BIG(*), @dw_total = SUM(OrderTotal) FROM Fact.Orders;

INSERT INTO @rules VALUES
 ('DQ01', N'Fact.Orders rows with the unknown customer (-1)', 'error',
     (SELECT COUNT_BIG(*) FROM Fact.Orders WHERE CustomerKey = -1)),
 ('DQ02', N'Customers without exactly one current version', 'error',
     (SELECT COUNT_BIG(*) FROM (SELECT CustomerAlternateKey FROM Dimension.Customer WHERE CustomerKey <> -1
                               GROUP BY CustomerAlternateKey HAVING SUM(CAST(IsCurrent AS int)) <> 1) AS x)),
 ('DQ03', N'Overlapping SCD2 versions', 'error',
     (SELECT COUNT_BIG(*) FROM Dimension.Customer AS a JOIN Dimension.Customer AS b
        ON a.CustomerAlternateKey = b.CustomerAlternateKey AND a.CustomerKey < b.CustomerKey
       AND a.ValidFrom < COALESCE(b.ValidTo, '9999-12-31') AND b.ValidFrom < COALESCE(a.ValidTo, '9999-12-31')
     WHERE a.CustomerKey <> -1)),
 ('DQ04', N'Orders with a zero or negative total', 'error',
     (SELECT COUNT_BIG(*) FROM Fact.Orders WHERE OrderTotal <= 0)),
 ('DQ05', CONCAT(N'Reconciliation with source: ', @src_orders, N' orders / ', @src_total,
                 N' vs warehouse ', @dw_orders, N' / ', @dw_total), 'error',
     ABS(@src_orders - @dw_orders) + CASE WHEN @src_total = @dw_total THEN 0 ELSE 1 END);

INSERT INTO audit.DQResult (BatchID, RuleID, RuleName, Severity, FailedRows, Passed)
SELECT @BatchID, RuleID, RuleName, Severity, FailedRows, CASE WHEN FailedRows = 0 THEN 1 ELSE 0 END FROM @rules;

SELECT RuleID, Severity, FailedRows,
       CASE WHEN FailedRows = 0 THEN N'pass' WHEN Severity = 'error' THEN N'FAIL' ELSE N'warn' END AS Result,
       RuleName
FROM @rules ORDER BY RuleID;

DECLARE @failed nvarchar(2000) =
    (SELECT STRING_AGG(CONCAT(RuleID, N' ', RuleName, N' (', FailedRows, N')'), N'; ')
     FROM @rules WHERE Severity = 'error' AND FailedRows > 0);
DECLARE @note nvarchar(400) = COALESCE(N'FAILED: ' + LEFT(@failed, 380), N'all error rules passed');
EXEC audit.usp_LogStep @DagRunID = N'{{ run_id }}', @TaskID = N'{{ ti.task_id }}', @Note = @note;

IF @failed IS NOT NULL
    THROW 50003, @failed, 1;

The gate passing (run 01) and failing (run 04)

RuleID | Severity | FailedRows | Result | RuleName
DQ01   | error    | 0          | pass   | Fact.Orders rows with the unknown customer (-1)
DQ02   | error    | 0          | pass   | Customers without exactly one current version
DQ03   | error    | 0          | pass   | Overlapping SCD2 versions
DQ04   | error    | 0          | pass   | Orders with a zero or negative total
DQ05   | error    | 0          | pass   | Reconciliation with source: 9567 orders / 9389789.51 vs warehouse 9567 / 9389789.51

DQ04   | error    | 1          | FAIL   | Orders with a zero or negative total
DQ05   | error    | 0          | pass   | Reconciliation with source: 9571 orders / 9392049.53 vs warehouse 9571 / 9392049.53

airflow.exceptions.AirflowException: SQL batch failed: ('42000', '[42000] [Microsoft][ODBC Driver 18 for SQL Server]
[SQL Server]DQ04 Orders with a zero or negative total (1) (50003) (SQLMoreResults)')

DagRunID              | RuleID | Severity | FailedRows | Passed
sbdw_run_04_bad_order | DQ01   | error    | 0          | 1
sbdw_run_04_bad_order | DQ02   | error    | 0          | 1
sbdw_run_04_bad_order | DQ03   | error    | 0          | 1
sbdw_run_04_bad_order | DQ04   | error    | 1          | 0
sbdw_run_04_bad_order | DQ05   | error    | 0          | 1

Note that DQ05 still passed in run 04: the warehouse faithfully matched the source, and the source itself contained the bad order. Reconciliation proves the pipeline is correct; business rules like DQ04 catch bad data the source is happy to accept. You need both.

The (SQLMoreResults) in the error is a detail worth knowing: the rule results were returned first, and the THROW arrived as the next result. pyodbc only raises it when the operator calls nextset(). An operator that reads just the first result set would report success. Chapter 11 shows the loop that prevents this.