Big Data AcademyQuery · move · process · govern

Chapter 19 · Materialized Views, Aggregate Tables, Cubes, and Precomputation

Design an Acceleration Layer with Clear Freshness and Correctness Contracts

Assemble an acceleration layer contract that binds grain, metric semantics, freshness SLOs, lineage, invalidation, security, observability, rollback, and reconciliation into one production acceptance workflow.

Intermediate → Advanced180–225 minutesAcceleration-layer acceptance lab216,000 benchmark rows · 731 aggregate rowsLast reviewed: September 2026
Continuity and explicit migration

Chapter 19 begins from the Chapter 18 accepted current-state controls: 9 paid lines, 7 orders, 11 units, 740 USD GMV, 450 USD cost, and 290 USD gross profit. The atomic grain remains one current paid order line. Precomputation never replaces or rewrites that atomic fact. Later in the lab, source sequence 207 deliberately introduces one new paid line (O1010/1, 80 USD GMV, 45 USD cost) so the base truth becomes 10 lines, 8 orders, 12 units, 820 USD GMV, 495 USD cost, 325 USD gross profit. That change is an explicit Chapter 19 fixture extension used to demonstrate stale acceleration and refresh.

Lab contract

Mandatory runtime: Python standard library plus SQLite; generation evidence used Python 3.13.5 and SQLite 3.46.1. Environment: local filesystem/in-process database; no managed service or paid feature. Storage: SQLite tables; because SQLite has ordinary views but no native materialized-view command, agg_daily_product is an explicitly managed materialized equivalent table. Time: AtlasMart business dates are UTC dates and refresh timestamps are UTC ISO-8601 strings. Keys: atomic sales key is (order_id,line_no); aggregate key is (order_date,product_id). History: this chapter does not change SCD rules; it accelerates the current paid-sales fact. Security: synthetic identities only. Benchmark cache: two warm-ups and seven measured executions; OS cache is not flushed. Non-guarantee: local timings do not predict cloud latency or billing.

Learning outcomes

01

Define a production acceleration contract covering grain, semantics, freshness, lineage, ownership, and rollback.

02

Run the complete deterministic local lab and verify stale/fresh transitions.

03

Reconcile accelerated outputs to atomic facts and retain proof through hashes/watermarks.

04

Bridge acceleration to Chapter 20 semantic-layer governance without moving business logic into uncontrolled aggregates.

1. An acceleration layer is a governed data product, not a hidden cache

AtlasMart is ready to publish agg_daily_product only if consumers can answer five questions without reverse-engineering code: what grain and metric semantics does it represent; what base state does it include; how fresh may it be; what queries/security scopes may use it; and who repairs or rolls it back? The acceleration contract below makes those answers machine-readable enough to test and human-readable enough to operate.

acceleration catalog entry 
{  "object": "agg_daily_product",  "base": "fact_sales",  "grain": "one paid-sales aggregate row per order_date + product_id",  "measures": ["line_count", "units", "gmv_cents", "cost_cents", "profit_cents"],  "non_additive_warning": "order_count cannot be summed across product/date groups for global distinct orders",  "refresh": {    "mode": "incremental-partition-recompute",    "watermark": 207,    "freshness_slo_seconds": 900  },  "semantic_contract_hash": "9a7b963a31b049bbe64a822b6a2dd9549f43b2732222cfac29cd600c0bf9a747",  "owner": "Analytics Engineering",  "lineage": ["fact_sales -> agg_daily_product -> certified sales dashboard"]}

2. Eligibility rules for routing

Check Required evidence If it fails
Semantic compatibility Metric/grain contract hash matches requested version Route to compatible object/base; rebuild accelerator
Dimension/filter coverage Requested dimensions exist at valid grain Use lower-grain fact
Freshness Acceleration watermark/timestamp satisfies consumer SLO Refresh, wait, or route to base
Correctness Latest reconciliation passed Quarantine accelerator from routing
Security Derived path enforces equivalent purpose/row/column policy Deny or use secured alternative
Operational health No failed/partial publication; owner/runbook available Fallback and incident response

3. Complete local lab

Save the script below as ch19_lab.py and run python ch19_lab.py. It creates /mnt/data/atlasmart_ch19_lab in the generation environment; when you run it locally, change OUT to a relative writable path such as Path('atlasmart_ch19_lab') if desired. The logic itself uses only Python standard library and SQLite.

ch19_lab.py 
from __future__ import annotationsimport hashlib, json, sqlite3, statistics, timefrom datetime import date, timedeltafrom pathlib import PathOUT = Path('/mnt/data/atlasmart_ch19_lab')OUT.mkdir(parents=True, exist_ok=True)DB = OUT/'atlasmart_ch19.sqlite'if DB.exists(): DB.unlink()CANONICAL = [    ('O1001',1,'2026-09-18','P100','C001','web',2,10000,6000,'paid',121),    ('O1001',2,'2026-09-18','P200','C001','web',1,2500,1500,'paid',122),    ('O1002',1,'2026-09-18','P300','C002','mobile',1,19000,12000,'paid',206),    ('O1003',1,'2026-09-19','P400','C001','web',1,10000,6500,'paid',123),    ('O1005',1,'2026-09-20','P100','C004','mobile',1,5000,3000,'paid',125),    ('O1005',2,'2026-09-20','P400','C004','mobile',1,10000,6500,'paid',126),    ('O1007',1,'2026-09-21','P300','C003','sales',1,7500,4500,'paid',201),    ('O1008',1,'2026-09-21','P200','C002','mobile',2,5000,2500,'paid',202),    ('O1009',1,'2026-09-22','P100','C005','web',1,5000,2500,'paid',203),]# This exact set preserves the accepted Chapter 15-18 controls.EXPECTED_CANONICAL = {'lines':9,'orders':7,'units':11,'gmv_usd':740,'cost_usd':450,'profit_usd':290}SCHEMA = '''PRAGMA foreign_keys=ON;CREATE TABLE fact_sales(  order_id TEXT NOT NULL,  line_no INTEGER NOT NULL,  order_date TEXT NOT NULL,  product_id TEXT NOT NULL,  customer_id TEXT NOT NULL,  channel TEXT NOT NULL,  quantity INTEGER NOT NULL,  amount_cents INTEGER NOT NULL,  cost_cents INTEGER NOT NULL,  status TEXT NOT NULL,  source_seq INTEGER NOT NULL,  PRIMARY KEY(order_id,line_no));CREATE TABLE agg_daily_product(  order_date TEXT NOT NULL,  product_id TEXT NOT NULL,  line_count INTEGER NOT NULL,  order_count INTEGER NOT NULL,  units INTEGER NOT NULL,  gmv_cents INTEGER NOT NULL,  cost_cents INTEGER NOT NULL,  profit_cents INTEGER NOT NULL,  refreshed_through_seq INTEGER NOT NULL,  refreshed_at TEXT NOT NULL,  semantic_contract_hash TEXT NOT NULL,  PRIMARY KEY(order_date,product_id));CREATE TABLE acceleration_state(  object_name TEXT PRIMARY KEY,  grain TEXT NOT NULL,  refresh_mode TEXT NOT NULL,  refreshed_through_seq INTEGER NOT NULL,  refreshed_at TEXT NOT NULL,  freshness_slo_seconds INTEGER NOT NULL,  semantic_contract_hash TEXT NOT NULL,  status TEXT NOT NULL);CREATE TABLE change_log(  event_id TEXT PRIMARY KEY,  source_seq INTEGER NOT NULL UNIQUE,  op TEXT NOT NULL CHECK(op IN ('I','U','D')),  order_id TEXT NOT NULL,  line_no INTEGER NOT NULL,  old_order_date TEXT,  old_product_id TEXT,  new_order_date TEXT,  new_product_id TEXT,  applied_at TEXT NOT NULL);'''METRIC_CONTRACT = {    'grain':'one current paid order line',    'filters':['status = paid'],    'measures':{        'lines':'count(*)',        'orders':'count(distinct order_id)',        'units':'sum(quantity)',        'gmv_cents':'sum(amount_cents)',        'cost_cents':'sum(cost_cents)',        'profit_cents':'sum(amount_cents-cost_cents)'    },    'aggregate_grain':['order_date','product_id'],    'currency':'USD cents',}CONTRACT_HASH = hashlib.sha256(json.dumps(METRIC_CONTRACT,sort_keys=True,separators=(',',':')).encode()).hexdigest()def connect():    c=sqlite3.connect(DB)    c.executescript(SCHEMA)    c.executemany('INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?,?)',CANONICAL)    c.commit(); return cdef controls(c):    row=c.execute('''SELECT count(*),count(distinct order_id),sum(quantity),sum(amount_cents),sum(cost_cents),sum(amount_cents-cost_cents)                     FROM fact_sales WHERE status='paid' ''').fetchone()    return {'lines':row[0],'orders':row[1],'units':row[2],'gmv_usd':row[3]//100,'cost_usd':row[4]//100,'profit_usd':row[5]//100}def agg_controls(c):    row=c.execute('''SELECT sum(line_count),sum(order_count),sum(units),sum(gmv_cents),sum(cost_cents),sum(profit_cents)                     FROM agg_daily_product''').fetchone()    return {'lines':row[0] or 0,'orders_additive_warning':row[1] or 0,'units':row[2] or 0,'gmv_usd':(row[3] or 0)//100,'cost_usd':(row[4] or 0)//100,'profit_usd':(row[5] or 0)//100}def full_refresh(c, seq:int, at:str):    c.execute('DELETE FROM agg_daily_product')    c.execute('''INSERT INTO agg_daily_product      SELECT order_date,product_id,count(*),count(distinct order_id),sum(quantity),sum(amount_cents),sum(cost_cents),sum(amount_cents-cost_cents),?,?,?      FROM fact_sales WHERE status='paid' GROUP BY order_date,product_id''',(seq,at,CONTRACT_HASH))    c.execute('''INSERT INTO acceleration_state VALUES(?,?,?,?,?,?,?,?)                 ON CONFLICT(object_name) DO UPDATE SET grain=excluded.grain,refresh_mode=excluded.refresh_mode,                 refreshed_through_seq=excluded.refreshed_through_seq,refreshed_at=excluded.refreshed_at,                 freshness_slo_seconds=excluded.freshness_slo_seconds,semantic_contract_hash=excluded.semantic_contract_hash,status=excluded.status''',              ('agg_daily_product','one paid sales aggregate row per order_date + product_id','full',seq,at,900,CONTRACT_HASH,'fresh'))    c.commit()def mark_stale(c):    c.execute("UPDATE acceleration_state SET status='stale' WHERE object_name='agg_daily_product'"); c.commit()def apply_insert(c):    c.execute('''INSERT INTO fact_sales VALUES ('O1010',1,'2026-09-22','P100','C003','sales',1,8000,4500,'paid',207)''')    c.execute('''INSERT INTO change_log VALUES ('EV207',207,'I','O1010',1,NULL,NULL,'2026-09-22','P100','2026-09-21T09:07:00Z')''')    mark_stale(c)def dirty_keys(c, from_seq, through_seq):    keys=set()    for row in c.execute('''SELECT old_order_date,old_product_id,new_order_date,new_product_id FROM change_log WHERE source_seq>? AND source_seq<=? ORDER BY source_seq''',(from_seq,through_seq)):        if row[0] and row[1]: keys.add((row[0],row[1]))        if row[2] and row[3]: keys.add((row[2],row[3]))    return sorted(keys)def incremental_refresh(c, through_seq:int, at:str):    state=c.execute("SELECT refreshed_through_seq FROM acceleration_state WHERE object_name='agg_daily_product'").fetchone()[0]    keys=dirty_keys(c,state,through_seq)    for d,p in keys:        c.execute('DELETE FROM agg_daily_product WHERE order_date=? AND product_id=?',(d,p))        c.execute('''INSERT INTO agg_daily_product          SELECT order_date,product_id,count(*),count(distinct order_id),sum(quantity),sum(amount_cents),sum(cost_cents),sum(amount_cents-cost_cents),?,?,?          FROM fact_sales WHERE status='paid' AND order_date=? AND product_id=? GROUP BY order_date,product_id''',(through_seq,at,CONTRACT_HASH,d,p))    c.execute('''UPDATE agg_daily_product SET refreshed_through_seq=?, refreshed_at=?, semantic_contract_hash=?''',(through_seq,at,CONTRACT_HASH))    c.execute('''UPDATE acceleration_state SET refresh_mode='incremental-partition-recompute',refreshed_through_seq=?,refreshed_at=?,status='fresh',semantic_contract_hash=? WHERE object_name='agg_daily_product' ''',(through_seq,at,CONTRACT_HASH))    c.commit(); return keysdef benchmark_rows():    # Reuse the Chapter 18 idea: deterministic physical expansion, not new production history.    start=date(2026,9,18); rows=[]    for rep in range(24000):        shift=rep%180        for r in CANONICAL:            d=(start+timedelta(days=shift+(date.fromisoformat(r[2])-start).days)).isoformat()            rows.append((f'{r[0]}-B{rep:05d}',r[1],d,r[3],r[4],r[5],r[6],r[7],r[8],r[9]))    return rowsdef run_benchmark():    bench=sqlite3.connect(':memory:')    bench.execute('''CREATE TABLE f(order_id TEXT,line_no INT,order_date TEXT,product_id TEXT,customer_id TEXT,channel TEXT,quantity INT,amount_cents INT,cost_cents INT,status TEXT)''')    rows=benchmark_rows(); bench.executemany('INSERT INTO f VALUES (?,?,?,?,?,?,?,?,?,?)',rows)    bench.execute('''CREATE TABLE a AS SELECT order_date,product_id,count(*) line_count,sum(quantity) units,sum(amount_cents) gmv_cents,sum(cost_cents) cost_cents FROM f WHERE status='paid' GROUP BY order_date,product_id''')    qbase="SELECT product_id,sum(amount_cents) FROM f WHERE status='paid' AND order_date BETWEEN '2027-01-01' AND '2027-01-31' GROUP BY product_id ORDER BY product_id"    qagg="SELECT product_id,sum(gmv_cents) FROM a WHERE order_date BETWEEN '2027-01-01' AND '2027-01-31' GROUP BY product_id ORDER BY product_id"    rb=bench.execute(qbase).fetchall(); ra=bench.execute(qagg).fetchall(); assert rb==ra    def measure(sql):        for _ in range(2): bench.execute(sql).fetchall()        vals=[]        for _ in range(7):            t=time.perf_counter(); bench.execute(sql).fetchall(); vals.append((time.perf_counter()-t)*1000)        return {'min_ms':round(min(vals),3),'median_ms':round(statistics.median(vals),3),'max_ms':round(max(vals),3)}    result={'fact_rows':bench.execute('select count(*) from f').fetchone()[0],            'aggregate_rows':bench.execute('select count(*) from a').fetchone()[0],            'result':rb,'base_timing':measure(qbase),'aggregate_timing':measure(qagg),            'base_plan':bench.execute('EXPLAIN QUERY PLAN '+qbase).fetchall(),            'aggregate_plan':bench.execute('EXPLAIN QUERY PLAN '+qagg).fetchall()}    bench.close(); return resultdef cube_evidence(c):    leaf=c.execute("SELECT count(*) FROM (SELECT order_date,product_id,channel FROM fact_sales WHERE status='paid' GROUP BY order_date,product_id,channel)").fetchone()[0]    dates=c.execute("SELECT count(distinct order_date) FROM fact_sales WHERE status='paid'").fetchone()[0]    products=c.execute("SELECT count(distinct product_id) FROM fact_sales WHERE status='paid'").fetchone()[0]    channels=c.execute("SELECT count(distinct channel) FROM fact_sales WHERE status='paid'").fetchone()[0]    dense=dates*products*channels    return {'dimensions':3,'grouping_sets_for_full_cube':2**3,'observed_leaf_combinations':leaf,'dense_leaf_space':dense,'sparsity_pct':round((1-leaf/dense)*100,2)}def checksum(c, table):    if table=='fact_sales':        rows=c.execute("SELECT order_id,line_no,order_date,product_id,quantity,amount_cents,cost_cents,status FROM fact_sales ORDER BY order_id,line_no").fetchall()    else:        rows=c.execute("SELECT order_date,product_id,line_count,units,gmv_cents,cost_cents,profit_cents FROM agg_daily_product ORDER BY order_date,product_id").fetchall()    return hashlib.sha256(json.dumps(rows,separators=(',',':')).encode()).hexdigest()def main():    c=connect()    canonical=controls(c); assert canonical==EXPECTED_CANONICAL, canonical    full_refresh(c,206,'2026-09-21T09:00:00Z')    fresh_agg=agg_controls(c); assert fresh_agg['gmv_usd']==740 and fresh_agg['cost_usd']==450 and fresh_agg['profit_usd']==290    base_hash_before=checksum(c,'fact_sales'); agg_hash_before=checksum(c,'agg_daily_product')    apply_insert(c)    stale_base=controls(c); stale_agg=agg_controls(c)    stale_state=c.execute("SELECT object_name,refresh_mode,refreshed_through_seq,refreshed_at,freshness_slo_seconds,status FROM acceleration_state").fetchone()    assert stale_base=={'lines':10,'orders':8,'units':12,'gmv_usd':820,'cost_usd':495,'profit_usd':325}    assert stale_agg['gmv_usd']==740    keys=incremental_refresh(c,207,'2026-09-21T09:10:00Z'); assert keys==[('2026-09-22','P100')]    repaired=agg_controls(c); assert repaired['gmv_usd']==820 and repaired['cost_usd']==495 and repaired['profit_usd']==325    first_refresh_hash=checksum(c,'agg_daily_product')    keys_rerun=incremental_refresh(c,207,'2026-09-21T09:11:00Z'); assert keys_rerun==[]    rerun_hash=checksum(c,'agg_daily_product'); assert rerun_hash==first_refresh_hash    state=c.execute('SELECT object_name,refresh_mode,refreshed_through_seq,refreshed_at,freshness_slo_seconds,status FROM acceleration_state').fetchone()    cube=cube_evidence(c)    bench=run_benchmark()    report={      'runtime':{'python':__import__('sys').version.split()[0],'sqlite':sqlite3.sqlite_version},      'semantic_contract_hash':CONTRACT_HASH,      'canonical_before_refresh':canonical,      'fresh_aggregate_before_change':fresh_agg,      'base_after_unrefreshed_insert':stale_base,      'stale_aggregate_after_insert':stale_agg,      'staleness_gap_usd':stale_base['gmv_usd']-stale_agg['gmv_usd'],      'stale_state_before_repair':stale_state,      'dirty_keys_recomputed':keys,      'rerun_dirty_keys':keys_rerun,      'idempotent_rerun_same_hash':rerun_hash==first_refresh_hash,      'aggregate_after_incremental_refresh':repaired,      'state':state,      'hashes':{'base_before_change':base_hash_before,'aggregate_before_change':agg_hash_before,'base_after_change':checksum(c,'fact_sales'),'aggregate_after_refresh':checksum(c,'agg_daily_product')},      'cube':cube,      'benchmark':bench,    }    (OUT/'report.json').write_text(json.dumps(report,indent=2),encoding='utf-8')    print(json.dumps(report,indent=2))    c.close()if __name__=='__main__': main()

4. Expected acceptance evidence

key expected output 
canonical before Chapter 19 extension: 9 lines / 7 orders / 11 units / $740 GMV / $450 cost / $290 profitfresh aggregate before change:            9 lines / 11 units / $740 GMV / $450 cost / $290 profitbase after unrefreshed seq-207 insert:     10 lines / 8 orders / 12 units / $820 GMV / $495 cost / $325 profitstale aggregate after insert:              9 lines / 11 units / $740 GMV / $450 cost / $290 profitstaleness gap:                              $80 GMVstate before repair:                        watermark 206, status=staledirty aggregate key:                        (2026-09-22, P100)aggregate after incremental refresh:        10 lines / 12 units / $820 GMV / $495 cost / $325 profitrefresh rerun:                              0 dirty keys; aggregate hash unchangedsemantic contract hash:                     9a7b963a31b049bbe64a822b6a2dd9549f43b2732222cfac29cd600c0bf9a747

The benchmark section also reports 216,000 synthetic scale rows versus 731 aggregate rows and a seven-run timing distribution. Timing numbers may vary by machine/cache/SQLite version, so the test asserts result equality, not a fixed speed ratio. The semantic-contract hash and deterministic table hashes provide replay evidence for the specific fixture.

5. Freshness failure injection and repair

The acceptance test deliberately creates a period in which the base says 820 USD and the aggregate says 740 USD. That stale window is not hidden. acceleration_state.status='stale' and watermark 206 make it observable. The repair recomputes exactly one dirty key and publishes watermark 207. An immediate rerun discovers no dirty keys and preserves the same aggregate hash.

What this proves

For this deterministic single-process fixture, the refresh function is replay-safe, detects the injected dependency change, restores additive reconciliation, and does not advance the published watermark before refresh commit. It does not prove distributed exactly-once processing, race-free concurrent readers/writers, native cloud materialized-view behavior, or target-engine cost.

6. Security, privacy, and governance

Precomputation can accidentally bypass security if the aggregate is stored outside the policy boundary of its base. A daily/product aggregate may seem non-personal, but small groups, sensitive products, tenant boundaries, or restricted regions can still expose protected information. Production design must document whether row-level security is applied before aggregation, after aggregation, or both, and must test that the accelerated route cannot return data the atomic route would deny.

Technical masking/access controls do not determine legal purpose limitation, retention, or jurisdictional obligations. The lab uses synthetic identities only.

7. Observability, cost, migration, and rollback

  • Observability: query route, base/aggregate watermark, refresh age, refresh duration, dirty groups, row counts, result checksum, reconciliation status, failures, and fallback count.
  • Cost: compare avoided query work with refresh compute, storage, invalidation/backfill, and operational support. Recalculate on the actual target billing model.
  • Migration: build the accelerator alongside existing queries, dual-run a representative corpus, and require semantic/result/freshness/security acceptance before routing traffic.
  • Rollback: disable aggregate navigation and route to atomic facts; keep the previous certified accelerator generation until the replacement passes.
  • Versioning: a metric-contract change produces a new semantic version/hash and invalidates incompatible accelerators rather than silently updating formulas in place.

8. Chapter acceptance checklist

  1. Atomic facts remain authoritative and accessible.
  2. Every accelerator has a declared business grain.
  3. Only compatible measures are advertised as additive.
  4. Freshness is observable and tied to a consumer SLO.
  5. Refresh/invalidation handles inserts, updates, deletes, late data, and retries as applicable.
  6. Base-versus-accelerator controls reconcile at the same watermark.
  7. Security equivalence is tested, not assumed.
  8. Benchmark claims disclose engine, cache, dataset, and query semantics.
  9. Lineage and ownership identify who responds to stale/corrupt acceleration.
  10. Fallback/rollback to governed base data is operationally available.

9. Bridge to Chapter 20

Chapter 19 has deliberately kept metric semantics outside the accelerator itself: the aggregate implements a versioned contract but does not own the business definition. Chapter 20 formalizes that separation through semantic layers, governed measures and metrics, filters/time windows, relationships, security, versioning, and BI contracts. Acceleration should become an implementation choice beneath those semantics—not a second source of truth.

Knowledge check

Check your understanding

  1. What should happen if the metric contract hash changes?
  2. Why can a fresh accelerator still be ineligible?
  3. What makes rollback unusually safe in this chapter’s design?
  4. Which acceptance evidence is stable across machines, and which is not?
  5. What belongs in Chapter 20 rather than inside an aggregate table?
Review the answers

1. Mark incompatible accelerators ineligible/stale for that semantic version and rebuild/revalidate them.

2. It may lack requested dimensions, exact metric semantics, or equivalent security scope.

3. Atomic governed facts remain available, so routing can fall back without reconstructing truth from aggregates.

4. Fixture totals, watermarks, contract/hash relationships, dirty-key counts, and result equality are stable; timing distributions are environment-dependent.

5. Central business definitions of measures/metrics, filters, time semantics, relationships, ownership, versioning, and consumer contracts.

Authoritative references

Keep knowledge open

Help the academy stay free and grow.

If these tutorials save you time, a small donation supports new lessons, technical review, diagrams, examples, and long-term maintenance.

ETHEthereum / ERC-20 only
0x716c4Ab160C4B66F31a28AE2448BfF68fc3a2ef0

Send only Ethereum or ERC-20 compatible assets to this address.