Chapter 18 · Columnar Storage, Compression, Encoding, Vectorized Execution, and Analytical Scan Economics
Estimate Scan Bytes/CPU and Redesign Table Layout to Reduce Work Without Corrupting the Logical Model
Turn scan bytes, row-group elimination, compression size, CPU timing, cache policy, and correctness checks into a physical-design decision record that reduces work while leaving the dimensional model and control totals unchanged.
Learning outcomes
Explain the central warehouse mechanism in “Estimate Scan Bytes/CPU and Redesign Table Layout to Reduce Work Without Corrupting the Logical Model” and connect it to AtlasMart’s declared grain and governed metrics.
Keep logical correctness, history, and control totals unchanged while evaluating the lesson’s physical or operational choice.
Run or interpret the deterministic local evidence and distinguish what it proves from engine-, cache-, scale-, or cloud-dependent behavior.
Diagnose the controlled failure, repair it safely, and state the production checks required before adopting the pattern.
The Chapter 15–17 canonical current-state controls remain 9 paid lines, 7 orders, 11 units, 740 USD GMV, 450 USD cost, and 290 USD gross profit. Chapter 18 expands those nine seed lines deterministically to 216,000 benchmark rows only to make storage mechanisms observable. Benchmark rows are never reported as new production facts.
Runtime: mandatory path uses Python 3 standard library; generation evidence below used Python 3.13.5 and SQLite 3.46.1. Storage: local filesystem only. Source: synthetic AtlasMart sales facts. Time: dates are UTC business dates; no time-zone conversion is hidden in the benchmark. Grain: one current paid order line. Security: synthetic identifiers only; no secrets or PII. History: the benchmark is a deterministic physical scale expansion, not a new business history. Cache: OS cache is not flushed; timing probes use two warm-ups and seven measured runs. Cost: no paid service is required.
1. Optimization must start with a query contract
The capstone query asks for P100 GMV on 2027-01-15 over the benchmark expansion. The logical contract is fixed: one paid order line grain; same date and product predicates; same amount definition; same synthetic expansion; same current-history policy. The query result is 2,660,000 cents (26,600.00 USD) in every representation.
A physical redesign is accepted only if it reduces work while
preserving that contract, the 9/7/11/740/450/290 canonical
controls, and the benchmark semantic SHA-256
d430abb12f2cd1b4afc56c2d1c168c2860308a9b655fcfbd709bd87eb16ef460.
2. Read the scan-economics evidence before the timing
| Representation / query | Rows/groups examined | Compressed bytes opened | Generation median | Interpretation |
|---|---|---|---|---|
| row CSV, selective | 216,000 rows | 1,344,039 | 164.878 ms | Complete row stream parsed because format/harness has no pruning. |
| clustered column store, selective | 1/18 groups; 12,000 rows inside group | 425 | 1.231 ms | 17 groups eliminated by date min/max; only date/product/amount chunks opened. |
| shuffled column store, selective | 18/18 groups; 216,000 rows | 484,495 | 22.159 ms | Projection still helps, but date distribution defeats group skipping. |
| clustered column store, full SUM(amount) | 18/18 groups | 4,020 | not used for cross-engine claim | No row-group elimination; projection reads only amount chunks. |
Timing is secondary because parsers and representations differ. Scan bytes/group elimination more directly expose the intended mechanisms. The timings are retained only with environment and cache disclosure.
3. Full reproducible local lab
Save the following as ch18_lab.py and run it from
an empty working directory. It creates a disposable
atlasmart_ch18_lab/ directory containing the row
CSV, SQLite row table, clustered and shuffled didactic column
stores, manifests, and profile.json. It uses only
Python’s standard library.
from __future__ import annotationsimport csv, gzip, hashlib, json, os, random, shutil, sqlite3, statistics, struct, timefrom collections import Counterfrom datetime import date, timedeltafrom pathlib import PathROOT = Path('atlasmart_ch18_lab')ROWGROUP = 12000REPEATS = 24000MEASURED = 7WARMUPS = 2EPOCH = date(1970,1,1)SEED = [ # order_id,line_no,date,product,customer,channel,qty,amount,cost,status ('O1001',1,'2026-09-18','P100','C001','web',2,10000,6000,'paid'), ('O1001',2,'2026-09-18','P200','C001','web',1,2500,1500,'paid'), ('O1002',1,'2026-09-18','P300','C002','mobile',1,19000,12000,'paid'), ('O1003',1,'2026-09-19','P400','C001','web',1,10000,6500,'paid'), ('O1005',1,'2026-09-20','P100','C004','mobile',1,5000,3000,'paid'), ('O1005',2,'2026-09-20','P400','C004','mobile',1,10000,6500,'paid'), ('O1007',1,'2026-09-21','P300','C003','sales',1,7500,4500,'paid'), ('O1008',1,'2026-09-21','P200','C002','mobile',2,5000,2500,'paid'), ('O1009',1,'2026-09-22','P100','C005','web',1,5000,2500,'paid'),]assert (len(SEED), len({r[0] for r in SEED}), sum(r[6] for r in SEED), sum(r[7] for r in SEED), sum(r[8] for r in SEED), sum(r[7]-r[8] for r in SEED)) == (9,7,11,74000,45000,29000)PROD = {v:i for i,v in enumerate(sorted({r[3] for r in SEED}))}CUST = {v:i for i,v in enumerate(sorted({r[4] for r in SEED}))}CHAN = {v:i for i,v in enumerate(sorted({r[5] for r in SEED}))}STAT = {v:i for i,v in enumerate(sorted({r[9] for r in SEED}))}SCHEMA = { 'synthetic_id': ('q', 8), 'order_date': ('i', 4), 'order_num': ('i', 4), 'line_no': ('h', 2), 'product_code': ('B', 1), 'customer_code': ('B', 1), 'channel_code': ('B', 1), 'quantity': ('h', 2), 'amount_cents': ('i', 4), 'cost_cents': ('i', 4), 'profit_cents': ('i', 4), 'status_code': ('B', 1),}def day_int(s: str) -> int: y,m,d=map(int,s.split('-')) return (date(y,m,d)-EPOCH).daysdef make_rows(): rows=[] sid=1 for rep in range(REPEATS): shift=rep % 180 for base in SEED: oid,line,ds,prod,cust,ch,qty,amt,cost,status=base d=date.fromisoformat(ds)+timedelta(days=shift) order_num=int(oid[1:]) + rep*100 rows.append((sid,(d-EPOCH).days,order_num,line,PROD[prod],CUST[cust],CHAN[ch],qty,amt,cost,amt-cost,STAT[status])) sid+=1 rows.sort(key=lambda r:(r[1], r[0])) return rowsdef sha_rows(rows): h=hashlib.sha256() for r in rows: h.update(('|'.join(map(str,r))+'\n').encode()) return h.hexdigest()def write_csv_gz(rows, path): names=list(SCHEMA) with gzip.open(path,'wt',newline='',encoding='utf-8',compresslevel=6) as f: w=csv.writer(f); w.writerow(names); w.writerows(rows)def write_sqlite(rows, path): con=sqlite3.connect(path) con.execute('PRAGMA journal_mode=OFF') con.execute('PRAGMA synchronous=OFF') con.execute('CREATE TABLE fact_sales (synthetic_id INTEGER PRIMARY KEY, order_date INTEGER NOT NULL, order_num INTEGER NOT NULL, line_no INTEGER NOT NULL, product_code INTEGER NOT NULL, customer_code INTEGER NOT NULL, channel_code INTEGER NOT NULL, quantity INTEGER NOT NULL, amount_cents INTEGER NOT NULL, cost_cents INTEGER NOT NULL, profit_cents INTEGER NOT NULL, status_code INTEGER NOT NULL)') con.executemany('INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?,?,?)',rows) con.commit(); con.execute('ANALYZE'); con.commit(); con.close()def pack_col(fmt, values): return struct.pack('<'+fmt*len(values), *values)def unpack_col(fmt, b): size=struct.calcsize('<'+fmt) return struct.unpack('<'+fmt*(len(b)//size),b)def write_column_store(rows, root): root.mkdir(parents=True,exist_ok=True) dictionaries={'product':PROD,'customer':CUST,'channel':CHAN,'status':STAT} (root/'dictionaries.json').write_text(json.dumps(dictionaries,indent=2,sort_keys=True),encoding='utf-8') manifest={'row_count':len(rows),'row_group_size':ROWGROUP,'row_groups':[],'schema':SCHEMA} names=list(SCHEMA) for g,start in enumerate(range(0,len(rows),ROWGROUP)): block=rows[start:start+ROWGROUP] gd=root/f'rg{g:03d}'; gd.mkdir() stats={'row_group':g,'start_row':start,'row_count':len(block),'min_order_date':min(r[1] for r in block),'max_order_date':max(r[1] for r in block),'columns':{}} for ci,name in enumerate(names): fmt,width=SCHEMA[name] raw=pack_col(fmt,[r[ci] for r in block]) p=gd/f'{name}.bin.gz' with gzip.open(p,'wb',compresslevel=6) as f: f.write(raw) stats['columns'][name]={'raw_bytes':len(raw),'compressed_bytes':p.stat().st_size} manifest['row_groups'].append(stats) (root/'manifest.json').write_text(json.dumps(manifest,indent=2,sort_keys=True),encoding='utf-8') return manifestdef col_store_size(root): return sum(p.stat().st_size for p in root.rglob('*') if p.is_file())def read_col(gd,name): fmt,_=SCHEMA[name] with gzip.open(gd/f'{name}.bin.gz','rb') as f: b=f.read() return unpack_col(fmt,b), (gd/f'{name}.bin.gz').stat().st_sizedef column_query(root,target_day,product_code): m=json.loads((root/'manifest.json').read_text()) total=0; rows_seen=0; groups_seen=0; groups_skipped=0; bytes_read=0 for rg in m['row_groups']: if not (rg['min_order_date'] <= target_day <= rg['max_order_date']): groups_skipped+=1; continue groups_seen+=1 gd=root/f"rg{rg['row_group']:03d}" days,b=read_col(gd,'order_date'); bytes_read+=b prods,b=read_col(gd,'product_code'); bytes_read+=b amts,b=read_col(gd,'amount_cents'); bytes_read+=b for d,p,a in zip(days,prods,amts): rows_seen+=1 if d==target_day and p==product_code: total+=a return {'total_cents':total,'row_groups_read':groups_seen,'row_groups_skipped':groups_skipped,'rows_examined_in_read_groups':rows_seen,'bytes_read':bytes_read}def column_full_sum(root): m=json.loads((root/'manifest.json').read_text()) total=0; bytes_read=0 for rg in m['row_groups']: gd=root/f"rg{rg['row_group']:03d}" amts,b=read_col(gd,'amount_cents'); bytes_read+=b; total+=sum(amts) return total,bytes_readdef row_csv_query(path,target_day,product_code): total=0; count=0 with gzip.open(path,'rt',newline='',encoding='utf-8') as f: rd=csv.reader(f); next(rd) for row in rd: count+=1 if int(row[1])==target_day and int(row[4])==product_code: total+=int(row[8]) return {'total_cents':total,'rows_examined':count,'bytes_read':path.stat().st_size}def sqlite_query(db,target_day,product_code): con=sqlite3.connect(db) q='SELECT SUM(amount_cents) FROM fact_sales WHERE order_date=? AND product_code=?' plan=con.execute('EXPLAIN QUERY PLAN '+q,(target_day,product_code)).fetchall() total=con.execute(q,(target_day,product_code)).fetchone()[0] con.close(); return total,plandef time_call(fn): for _ in range(WARMUPS): fn() vals=[] for _ in range(MEASURED): t=time.perf_counter(); fn(); vals.append((time.perf_counter()-t)*1000) return {'median_ms':round(statistics.median(vals),3),'min_ms':round(min(vals),3),'max_ms':round(max(vals),3),'runs':MEASURED,'warmups':WARMUPS}def encoding_demo(rows): channels=[r[6] for r in rows] dates=[r[1] for r in rows] # Plain 32-bit code vs minimum 8-bit dictionary codes (didactic sizing). plain_channel=len(channels)*4 dict_channel=len(channels)*1 + sum(len(k.encode())+1 for k in CHAN) # RLE on sorted date codes: each run stored as 4-byte value + 4-byte run length. runs=1 for a,b in zip(dates,dates[1:]): if a!=b: runs+=1 rle_date=runs*8 plain_date=len(dates)*4 # Delta-like: first int32 + signed int16 deltas for this bounded sorted fixture. deltas=[b-a for a,b in zip(dates,dates[1:])] assert max(deltas)<=32767 and min(deltas)>=-32768 delta_date=4+len(deltas)*2 return {'channel_plain32_bytes':plain_channel,'channel_dictionary_bytes':dict_channel,'date_plain32_bytes':plain_date,'date_rle_bytes':rle_date,'date_runs':runs,'date_delta_like_bytes':delta_date}def batch_proxy(rows): amounts=[r[8] for r in rows] def row_loop(): s=0 for v in amounts: s+=v return s def builtin_batch(): return sum(amounts) assert row_loop()==builtin_batch() return {'row_loop':time_call(row_loop),'builtin_sum_proxy':time_call(builtin_batch),'note':'Python built-in sum is a C-level batching proxy; this does not prove SIMD.'}def main(): if ROOT.exists(): shutil.rmtree(ROOT) ROOT.mkdir() rows=make_rows() assert len(rows)==216000 baseline={'lines':9,'orders':7,'units':11,'gmv_usd':740,'cost_usd':450,'profit_usd':290} csv_gz=ROOT/'fact_sales_rows.csv.gz'; db=ROOT/'fact_sales.sqlite'; col=ROOT/'column_store_sorted'; shuffled_col=ROOT/'column_store_shuffled' write_csv_gz(rows,csv_gz); write_sqlite(rows,db); manifest=write_column_store(rows,col) shuffled=list(rows); random.Random(18018).shuffle(shuffled); shuffled_manifest=write_column_store(shuffled,shuffled_col) target_day=day_int('2027-01-15'); p100=PROD['P100'] row_res=row_csv_query(csv_gz,target_day,p100) col_res=column_query(col,target_day,p100) shuffled_res=column_query(shuffled_col,target_day,p100) sql_total,sql_plan=sqlite_query(db,target_day,p100) assert row_res['total_cents']==col_res['total_cents']==shuffled_res['total_cents']==sql_total full_col,full_col_bytes=column_full_sum(col) assert full_col==sum(r[8] for r in rows) enc=encoding_demo(rows) prof={ 'runtime':{'python':os.sys.version.split()[0],'sqlite':sqlite3.sqlite_version,'storage':'local filesystem','cache_policy':'OS cache not flushed; 2 warmups, 7 measured runs'}, 'canonical_controls':baseline, 'benchmark':{'rows':len(rows),'date_span_days':180,'row_group_size':ROWGROUP,'row_groups':len(manifest['row_groups']),'semantic_sha256':sha_rows(rows)}, 'sizes':{'row_csv_gz':csv_gz.stat().st_size,'sqlite_db':db.stat().st_size,'column_store_sorted_total':col_store_size(col),'column_store_shuffled_total':col_store_size(shuffled_col)}, 'selective_query':{'target_date':'2027-01-15','product':'P100','row_csv':row_res,'column_store_sorted':col_res,'column_store_shuffled':shuffled_res,'sqlite_total_cents':sql_total,'sqlite_plan':sql_plan}, 'full_sum':{'total_cents':full_col,'column_amount_bytes_read':full_col_bytes}, 'encoding_demo':enc, 'timing':{'row_csv_selective':time_call(lambda: row_csv_query(csv_gz,target_day,p100)),'column_sorted_selective':time_call(lambda: column_query(col,target_day,p100)),'column_shuffled_selective':time_call(lambda: column_query(shuffled_col,target_day,p100))}, 'batch_proxy':batch_proxy(rows), 'non_guarantees':['Custom column_store is a didactic row-group/column-chunk harness, not Parquet/ORC.','Cross-representation timings include different parsers and are not portable performance ratios.','Python built-in sum does not demonstrate hardware SIMD.','Min/max metadata enables skipping only when data distribution and predicates make groups excludable.'] } (ROOT/'profile.json').write_text(json.dumps(prof,indent=2,sort_keys=True),encoding='utf-8') print('canonical',baseline) print('benchmark rows',len(rows),'row groups',len(manifest['row_groups'])) print('sizes',prof['sizes']) print('selective row',row_res) print('selective column sorted',col_res) print('selective column shuffled',shuffled_res) print('sqlite plan',sql_plan) print('encoding',enc) print('timing',prof['timing']) print('batch_proxy',prof['batch_proxy']) print('sha256',prof['benchmark']['semantic_sha256']) print('cleanup: rm -rf atlasmart_ch18_lab')if __name__=='__main__': main()4. Expected acceptance evidence
canonical {'lines': 9, 'orders': 7, 'units': 11, 'gmv_usd': 740, 'cost_usd': 450, 'profit_usd': 290}benchmark rows 216000 row groups 18selective row {'total_cents': 2660000, 'rows_examined': 216000, 'bytes_read': 1344039}selective column sorted {'total_cents': 2660000, 'row_groups_read': 1, 'row_groups_skipped': 17, 'rows_examined_in_read_groups': 12000, 'bytes_read': 425}selective column shuffled {'total_cents': 2660000, 'row_groups_read': 18, 'row_groups_skipped': 0, 'rows_examined_in_read_groups': 216000, 'bytes_read': 484495}sqlite plan [(3, 0, 0, 'SCAN fact_sales')]encoding {'channel_plain32_bytes': 864000, 'channel_dictionary_bytes': 216017, 'date_plain32_bytes': 864000, 'date_rle_bytes': 1472, 'date_runs': 184, 'date_delta_like_bytes': 432002}sha256 d430abb12f2cd1b4afc56c2d1c168c2860308a9b655fcfbd709bd87eb16ef460cleanup: rm -rf atlasmart_ch18_labFile-size and timing fields can vary across Python/zlib/filesystem versions. The exact semantic totals, row/group counts, encoding arithmetic, and SHA-256 are deterministic for the provided script.
5. Controlled wrong redesigns and repairs
| Tempting change | Failure | Repair / acceptance test |
|---|---|---|
| Drop atomic columns because dashboard uses only GMV | Destroys drilldown/restatement/reconciliation capability. | Keep atomic governed fact; optimize physical projection or add governed aggregates later. |
| Shuffle data but expect min/max skipping | Every group spans target date; zero groups eliminated. | Cluster/sort only if workload evidence justifies rewrite/maintenance cost. |
| Compare a filtered column query to an unfiltered row query | Benchmark semantics differ. | Assert identical totals/filter predicates before measuring. |
| Publish local latency ratio as cloud savings | Different engine, cache, concurrency, storage, and billing model. | Re-run on target platform and map actual scanned bytes/compute to its pricing contract. |
| Mutate raw historical data to create better compression | Breaks replay/auditability. | Rebuild derived physical layouts from immutable/governed inputs; retain versioned manifests. |
6. Physical-design decision record
For AtlasMart’s benchmark workload, the evidence supports retaining the logical star schema while favoring column projection and date-local row groups for repeated date-selective analytics. It does not justify a universal row-group size, a specific production codec, or a claim that all queries should be date-clustered. Full-scan measures, product-heavy filters, update frequency, backfills, concurrency, and downstream format support must also be represented in the workload corpus.
Version physical layouts independently of model semantics. Before cutover, dual-read a representative query corpus and reconcile checksums/totals. Keep the previous layout until consumer compatibility and operational recovery are proven. Rebuilding the optimized representation from governed Chapter 14 raw/integration data is the preferred rollback path; do not edit historical facts to “fit” the storage design.
7. Correctness, freshness, security, and observability checklist
- Correctness: query results and canonical controls must reconcile across old/new layouts.
- Freshness/history: physical compaction/rewrite must not lose late corrections or SCD bindings.
- Idempotency/replay: same governed input + layout version yields the same semantic checksum.
- Data quality: malformed values remain governed/quarantined upstream; compression never hides them.
- Security/privacy: synthetic lab only; production metadata/statistics may expose ranges/cardinality and must follow access/encryption policy.
- Observability: record layout version, bytes, groups/partitions touched, rows, CPU/wall time, spills, cache state, result hash, and engine version.
- Performance/cost: measure representative concurrency and actual platform billing units.
- Compatibility: validate every reader before enabling format/encoding features.
8. Bridge to Chapter 19
Chapter 18 reduces the cost of reading the base fact. Chapter 19 asks a different question: when repeated scans remain too expensive, should the warehouse persist precomputed aggregate tables, materialized views, or cube-like structures? The same rule survives: acceleration is accepted only when freshness and reconciliation to the base fact are explicit.
Knowledge check
Check your understanding
- Why are scan bytes more diagnostic than the local latency ratio here?
- Why does the shuffled store still read fewer bytes than the row CSV?
- What proves the optimized layout did not corrupt semantics?
- Why is the full SUM(amount) unable to skip row groups?
- What should happen before a physical-layout cutover?
Review the answers
1. They directly expose projection/skipping mechanisms while cross-representation timing includes parser/runtime differences.
2. Projection reads only date/product/amount columns even though no groups can be eliminated.
3. Equal query totals, canonical controls, and deterministic semantic checksum.
4. Every logical row contributes to the aggregate; only column projection can reduce width.
5. Dual-read/reconcile representative queries, test compatibility/recovery, and retain rollback/rebuild capability.
Authoritative references
- Apache Parquet — ConceptsOfficial terminology for row groups, column chunks, pages, and the units at which I/O and encoding occur.
- Apache Parquet — File FormatOfficial layout showing column chunks organized inside row groups and metadata used to locate relevant chunks.
- Apache Parquet — EncodingsOfficial definitions of plain, dictionary, run-length/bit-packed, and delta encodings. Parquet is a reference, not a prerequisite for the mandatory lab.
- DuckDB — Execution FormatCurrent official example of a vectorized analytical engine. This course uses the page only as a non-prerequisite reference for the execution concept.
- SQLite — EXPLAIN QUERY PLANOfficial documentation for the local row-store plan evidence. The plan text is diagnostic output, not a stable application API.
- Python — gzipStandard-library compression used by the dependency-free local storage harness.
- Python — structStandard-library binary packing used for fixed-width didactic column chunks.
- Python — sqlite3Standard-library SQLite interface used for the local row-oriented table and query-plan evidence.
- Kimball Group — Dimensional Modeling TechniquesBackground for keeping dimensional grain and metric semantics stable while changing physical storage and access paths.