Create a Physical Design from Workload Evidence Rather Than Copying Generic Partition Recommendations

from pathlib import Pathimport sqlite3, json, hashlib, csv, shutil, statistics, timefrom datetime import date, timedeltaROOT = Path('/mnt/data/atlasmart_ch17_lab')if ROOT.exists():    shutil.rmtree(ROOT)ROOT.mkdir()for d in ['db','layouts/date','layouts/month','layouts/hash_customer','reports']:    (ROOT/d).mkdir(parents=True, exist_ok=True)PYTHON_VERSION = __import__('sys').version.split()[0]SQLITE_VERSION = sqlite3.sqlite_versionTZ = 'UTC'# Chapter 16 accepted control state: 9 paid lines, 7 orders, 11 units, 740 GMV, 450 cost, 290 gross profit.seed = [    # order_id, line_no, customer_id, product_id, store_id, qty, revenue, cost    ('O0999',1,'C003','P300','SALES',1,80,50),    ('O1000',1,'C001','P200','WEB',1,75,45),    ('O1001',1,'C001','P100','WEB',2,100,60),    ('O1001',2,'C001','P200','WEB',1,25,15),    ('O1002',1,'C002','P300','MOBILE',1,195,125),    ('O1003',1,'C001','P400','WEB',1,100,60),    ('O1003',2,'C001','P200','WEB',2,55,30),    ('O1005',1,'C004','P100','MOBILE',1,50,30),    ('O1006',1,'C002','P100','WEB',1,60,35),]assert sum(x[5] for x in seed)==11assert sum(x[6] for x in seed)==740assert sum(x[7] for x in seed)==450assert len({x[0] for x in seed})==7# Deterministic benchmark expansion; does not replace continuity controls.rows=[]start=date(2026,7,24)replicas=40for day_i in range(60):    d=(start+timedelta(days=day_i)).isoformat()    for rep in range(replicas):        for idx,s in enumerate(seed):            order_id,line_no,cust,prod,store,qty,rev,cost=s            # Stable unique order id per benchmark replica/day while preserving dimensional distribution.            oid=f'{order_id}-D{day_i:02d}-R{rep:02d}'            rows.append((d,oid,line_no,cust,prod,store,qty,rev,cost,rev-cost))assert len(rows)==60*replicas*len(seed)logical_checksum=hashlib.sha256('\n'.join('|'.join(map(str,r)) for r in rows).encode()).hexdigest()schema='''PRAGMA foreign_keys=ON;CREATE TABLE dim_customer(customer_id TEXT PRIMARY KEY, customer_name TEXT NOT NULL);CREATE TABLE dim_product(product_id TEXT PRIMARY KEY, product_name TEXT NOT NULL);CREATE TABLE dim_store(store_id TEXT PRIMARY KEY, store_name TEXT NOT NULL);CREATE TABLE fact_sales(  order_date TEXT NOT NULL,  order_id TEXT NOT NULL,  line_no INTEGER NOT NULL,  customer_id TEXT NOT NULL REFERENCES dim_customer(customer_id),  product_id TEXT NOT NULL REFERENCES dim_product(product_id),  store_id TEXT NOT NULL REFERENCES dim_store(store_id),  quantity INTEGER NOT NULL CHECK(quantity > 0),  revenue_usd INTEGER NOT NULL CHECK(revenue_usd >= 0),  cost_usd INTEGER NOT NULL CHECK(cost_usd >= 0),  gross_profit_usd INTEGER NOT NULL,  PRIMARY KEY(order_id,line_no));'''def make_db(path, indexed=False):    cx=sqlite3.connect(path)    cx.executescript(schema)    cx.executemany('INSERT INTO dim_customer VALUES (?,?)',[(f'C{i:03d}',f'Customer {i:03d}') for i in range(1,6)])    cx.executemany('INSERT INTO dim_product VALUES (?,?)',[(f'P{i}',f'Product {i}') for i in (100,200,300,400)])    cx.executemany('INSERT INTO dim_store VALUES (?,?)',[('WEB','Web'),('MOBILE','Mobile'),('SALES','Sales')])    cx.executemany('INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?)',rows)    if indexed:        cx.execute('CREATE INDEX idx_fact_sales_date_product ON fact_sales(order_date, product_id)')        cx.execute('CREATE INDEX idx_fact_sales_customer_date ON fact_sales(customer_id, order_date)')    cx.commit(); cx.close()heap=ROOT/'db/atlasmart_heap.db'; indexed=ROOT/'db/atlasmart_indexed.db'make_db(heap,False); make_db(indexed,True)queries={ 'Q1_daily_gmv':("SELECT SUM(revenue_usd) FROM fact_sales WHERE order_date=?",('2026-09-21',)), 'Q2_date_product':("SELECT SUM(revenue_usd) FROM fact_sales WHERE order_date BETWEEN ? AND ? AND product_id=?",('2026-09-15','2026-09-21','P100')), 'Q3_customer_window':("SELECT SUM(revenue_usd) FROM fact_sales WHERE customer_id=? AND order_date BETWEEN ? AND ?",('C002','2026-09-01','2026-09-21')), 'Q4_full_gmv':("SELECT SUM(revenue_usd) FROM fact_sales",()),}def explain(db,sql,args):    cx=sqlite3.connect(db)    plan=[' | '.join(map(str,r)) for r in cx.execute('EXPLAIN QUERY PLAN '+sql,args)]    val=cx.execute(sql,args).fetchone()[0]    cx.close(); return plan,valplans={}for name,(sql,args) in queries.items():    hp,hv=explain(heap,sql,args); ip,iv=explain(indexed,sql,args)    assert hv==iv    plans[name]={'result':hv,'heap_plan':hp,'indexed_plan':ip}# Local timing is real but environment-specific; use medians across 15 executions after 3 warmups.def timing(db,sql,args):    cx=sqlite3.connect(db)    for _ in range(3): cx.execute(sql,args).fetchone()    vals=[]    for _ in range(15):        t=time.perf_counter_ns(); cx.execute(sql,args).fetchone(); vals.append((time.perf_counter_ns()-t)/1e6)    cx.close()    return {'median_ms':round(statistics.median(vals),4),'p95_ms':round(sorted(vals)[int(0.95*(len(vals)-1))],4),'runs':15}timings={n:{'heap':timing(heap,*q),'indexed':timing(indexed,*q)} for n,q in queries.items()}# Physical file-layout simulator: write canonical rows as compact JSONL.def dump_partition(path, part_rows):    with path.open('w',encoding='utf-8',newline='\n') as f:        for r in part_rows:            f.write(json.dumps(r,separators=(',',':'))+'\n')    return path.stat().st_sizeby_date={}by_month={}for r in rows:    by_date.setdefault(r[0],[]).append(r)    by_month.setdefault(r[0][:7],[]).append(r)date_manifest={d:dump_partition(ROOT/f'layouts/date/{d}.jsonl',rs) for d,rs in sorted(by_date.items())}month_manifest={m:dump_partition(ROOT/f'layouts/month/{m}.jsonl',rs) for m,rs in sorted(by_month.items())}# Stable hash buckets by customer_id, to simulate distribution/co-location reasoning.def bucket(key,n=8):    return int(hashlib.sha256(key.encode()).hexdigest()[:16],16)%nhash_rows={i:[] for i in range(8)}for r in rows: hash_rows[bucket(r[3])].append(r)hash_manifest={str(i):dump_partition(ROOT/f'layouts/hash_customer/bucket={i}.jsonl',rs) for i,rs in hash_rows.items()}# Dimension co-location using same hash rule.customers=[f'C{i:03d}' for i in range(1,6)]customer_buckets={c:bucket(c) for c in customers}row_bucket_counts={str(i):len(hash_rows[i]) for i in range(8)}nonempty=[v for v in row_bucket_counts.values() if v]skew_ratio=round(max(nonempty)/min(nonempty),3)# Scan accounting for Q1-Q4 on the file layouts. No database-engine pruning is claimed.def bytes_sum(man): return sum(man.values())all_date_bytes=bytes_sum(date_manifest); all_month_bytes=bytes_sum(month_manifest); all_hash_bytes=bytes_sum(hash_manifest)def dates_between(a,b):    return [d for d in date_manifest if a<=d<=b]scan={ 'Q1_daily_gmv':{   'date':{'files':1,'bytes':date_manifest['2026-09-21']},   'month':{'files':1,'bytes':month_manifest['2026-09']},   'hash_customer':{'files':8,'bytes':all_hash_bytes}, }, 'Q2_date_product':{   'date':{'files':len(dates_between('2026-09-15','2026-09-21')),'bytes':sum(date_manifest[d] for d in dates_between('2026-09-15','2026-09-21'))},   'month':{'files':1,'bytes':month_manifest['2026-09']},   'hash_customer':{'files':8,'bytes':all_hash_bytes}, }, 'Q3_customer_window':{   'date':{'files':len(dates_between('2026-09-01','2026-09-21')),'bytes':sum(date_manifest[d] for d in dates_between('2026-09-01','2026-09-21'))},   'month':{'files':1,'bytes':month_manifest['2026-09']},   'hash_customer':{'files':1,'bytes':hash_manifest[str(customer_buckets['C002'])]}, }, 'Q4_full_gmv':{   'date':{'files':len(date_manifest),'bytes':all_date_bytes},   'month':{'files':len(month_manifest),'bytes':all_month_bytes},   'hash_customer':{'files':8,'bytes':all_hash_bytes}, },}# Same query results from partition files, proving layout did not change logical semantics.def scan_json_files(paths,predicate):    gmv=0    for p in paths:        for line in p.read_text(encoding='utf-8').splitlines():            r=json.loads(line)            if predicate(r): gmv+=r[7]    return gmvq1_file=scan_json_files([ROOT/'layouts/date/2026-09-21.jsonl'],lambda r:r[0]=='2026-09-21')q2_file=scan_json_files([ROOT/f'layouts/date/{d}.jsonl' for d in dates_between('2026-09-15','2026-09-21')],lambda r:'2026-09-15'<=r[0]<='2026-09-21' and r[4]=='P100')q3_bucket=customer_buckets['C002']q3_file=scan_json_files([ROOT/f'layouts/hash_customer/bucket={q3_bucket}.jsonl'],lambda r:r[3]=='C002' and '2026-09-01'<=r[0]<='2026-09-21')q4_file=scan_json_files([ROOT/f'layouts/date/{d}.jsonl' for d in date_manifest],lambda r:True)assert q1_file==plans['Q1_daily_gmv']['result']assert q2_file==plans['Q2_date_product']['result']assert q3_file==plans['Q3_customer_window']['result']assert q4_file==plans['Q4_full_gmv']['result']# Constraints: valid insert succeeds; invalid quantity and orphan customer are rejected.cx=sqlite3.connect(indexed); cx.execute('PRAGMA foreign_keys=ON')constraint_results={}for name,sql,args in [ ('negative_quantity','INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?)',('2026-09-22','BAD-QTY',1,'C001','P100','WEB',0,10,5,5)), ('orphan_customer','INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?)',('2026-09-22','BAD-CUST',1,'C999','P100','WEB',1,10,5,5)),]:    try:        cx.execute(sql,args); cx.rollback(); constraint_results[name]='NOT_REJECTED'    except sqlite3.IntegrityError as e:        constraint_results[name]='REJECTED: '+str(e)cx.close()assert all(v.startswith('REJECTED') for v in constraint_results.values())# Tiny/high-cardinality partition calculation (do not create 21,600 files).tiny_partition_count=len(date_manifest)*4  # date x producthigh_card_partition_count=len(rows)  # one per row/order-line surrogate-like keyavg_date_file=round(all_date_bytes/len(date_manifest),1)avg_tiny_est=round(all_date_bytes/tiny_partition_count,1)# Decision record based on this specific corpus.decision={ 'logical_model_changed':False, 'benchmark_rows':len(rows), 'continuity_controls':{'paid_lines':9,'paid_orders':7,'units':11,'gmv_usd':740,'cost_usd':450,'gross_profit_usd':290}, 'recommended_local_layout':{   'fact_sales_processing_partition':'order_date (logical/file partition in this harness)',   'access_path':'SQLite composite indexes on (order_date, product_id) and (customer_id, order_date)',   'distribution':'not applicable in single-node SQLite; simulate customer hash only for skew/co-location reasoning' }, 'reasons':[   'Q1/Q2 date predicates prune the deterministic file corpus substantially.',   'Q3 benefits from customer-led access; hash-by-customer illustrates possible co-location but is poor for date-only pruning.',   'Q4 is a full scan by definition, so partitioning does not remove scan work for that query.',   'Date-by-product tiny partitions multiply file count without changing grain.',   'Distribution/clustering semantics are engine-specific and require re-benchmarking after migration.' ], 'rollback':'Keep logical schema/grain unchanged; remove/rebuild indexes or rematerialize file layout from canonical rows if evidence changes.'}report={ 'runtime':{'python':PYTHON_VERSION,'sqlite':SQLITE_VERSION,'timezone':TZ,'mode':'local/synthetic','cache_note':'SQLite page/OS cache not flushed; plans and byte accounting are primary evidence; timings are environment-specific.'}, 'logical_checksum':logical_checksum, 'rows':len(rows), 'plans':plans, 'timings_ms':timings, 'file_bytes':{'date_total':all_date_bytes,'month_total':all_month_bytes,'hash_total':all_hash_bytes,'date_files':len(date_manifest),'month_files':len(month_manifest),'hash_files':8,'avg_date_file':avg_date_file,'estimated_date_product_files':tiny_partition_count,'estimated_avg_date_product_bytes':avg_tiny_est,'high_cardinality_partition_count':high_card_partition_count}, 'scan':scan, 'hash_distribution':{'customer_buckets':customer_buckets,'row_bucket_counts':row_bucket_counts,'nonempty_skew_ratio':skew_ratio}, 'constraint_tests':constraint_results, 'query_results':{k:v['result'] for k,v in plans.items()}, 'layout_equivalence':{'Q1':q1_file,'Q2':q2_file,'Q3':q3_file,'Q4':q4_file}, 'decision':decision,}(ROOT/'reports/benchmark_report.json').write_text(json.dumps(report,indent=2),encoding='utf-8')(ROOT/'reports/physical_design_decision.json').write_text(json.dumps(decision,indent=2),encoding='utf-8')# Human-readable concise output used by lesson verification.print(f'runtime: Python {PYTHON_VERSION} / SQLite {SQLITE_VERSION} / {TZ}')print('continuity controls: (9, 7, 11, 740, 450, 290)')print('benchmark rows:',len(rows))print('query results:',{k:v['result'] for k,v in plans.items()})print('Q1 heap plan:',plans['Q1_daily_gmv']['heap_plan'])print('Q1 indexed plan:',plans['Q1_daily_gmv']['indexed_plan'])print('Q3 indexed plan:',plans['Q3_customer_window']['indexed_plan'])print('date files/bytes:',len(date_manifest),all_date_bytes)print('month files/bytes:',len(month_manifest),all_month_bytes)print('hash bucket counts:',row_bucket_counts,'skew_ratio:',skew_ratio)print('Q1 date scan:',scan['Q1_daily_gmv']['date'])print('Q1 hash scan:',scan['Q1_daily_gmv']['hash_customer'])print('Q3 hash scan:',scan['Q3_customer_window']['hash_customer'])print('tiny partition estimate:',tiny_partition_count,'files avg_bytes',avg_tiny_est)print('high-cardinality partition estimate:',high_card_partition_count,'files')print('constraints:',constraint_results)print('layout equivalence:',report['layout_equivalence'])print('decision logical_model_changed:',decision['logical_model_changed'])print('logical checksum:',logical_checksum)print('cleanup: rm -rf atlasmart_ch17_lab')

runtime: Python 3.13.5 / SQLite 3.46.1 / UTCcontinuity controls: (9, 7, 11, 740, 450, 290)benchmark rows: 21600query results: {'Q1_daily_gmv': 29600, 'Q2_date_product': 58800, 'Q3_customer_window': 214200, 'Q4_full_gmv': 1776000}Q1 heap plan: ['3 | 0 | 0 | SCAN fact_sales']Q1 indexed plan: ['4 | 0 | 0 | SEARCH fact_sales USING INDEX idx_fact_sales_date_product (order_date=?)']date files/bytes: 60 1411200month files/bytes: 3 1411200hash bucket counts: {'0': 4800, '1': 0, '2': 0, '3': 16800, '4': 0, '5': 0, '6': 0, '7': 0} skew_ratio: 3.5Q1 date scan: {'files': 1, 'bytes': 23520}Q3 hash scan: {'files': 1, 'bytes': 319200}tiny partition estimate: 240 files avg_bytes 5880.0high-cardinality partition estimate: 21600 fileslayout equivalence: {'Q1': 29600, 'Q2': 58800, 'Q3': 214200, 'Q4': 1776000}decision logical_model_changed: Falselogical checksum: c1ec45150cb017bed341ec1de5f728fbf8df533e77d7b7f1f73418c5f7cc0a4a

rm -rf atlasmart_ch17_lab# Windows PowerShell equivalent:# Remove-Item -Recurse -Force .tlasmart_ch17_lab