feat(docs): update README and plan to reflect task implementations
- mark tasks 10 and 11 as implemented in the project overview - add provenance note regarding the initial prompt in README.md - clarify the relationship between the prompt and specifications in 00_PLAN.md - update documentation structure to include Initial_Prompt.md for historical context
This commit is contained in:
administrator
273
extrapolate.py
Normal file
273
extrapolate.py
Normal file
@@ -0,0 +1,273 @@
|
||||
"""Task 10 - Extrapolation to 600 GB / 1.2 TB / 6 TB.
|
||||
|
||||
Calibrates per-cell scaling laws on the measured 150 MB point (task 09
|
||||
medians) and projects wall times and hardware needs per format x scale.
|
||||
Model (spec 10, docs/rules/bench-methodology.md section 7):
|
||||
- per-format fixed overhead = the format's cheapest measured median
|
||||
(interpreter start, imports, connection); the remainder is the
|
||||
data-dependent calibration constant;
|
||||
- growth laws per access pattern: o1 (flat), log (index depth,
|
||||
1 + ln r / ln n0 with n0 = measured friction-cycle rows), logk (result
|
||||
set grows linearly with data), k_log_n (linear x index depth), scan
|
||||
(linear), scan_parallel (linear x base_cores / cores_at_scale; range
|
||||
partitions outnumber cores at every scale);
|
||||
- RDF join penalties are already embedded in the measured constants;
|
||||
- RDF q3/q5/q7 are scans because the implementation derives run-in /
|
||||
steady-state from raw cycles (no summary triples exist by design).
|
||||
Every number written here is a PROJECTION, never a measurement; rows are
|
||||
flagged IMPRACTICAL (> 1 h) and FAIL (> 24 h).
|
||||
Spec: docs/specs/10_extrapolation_model.md.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import logging
|
||||
import math
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import psycopg
|
||||
import yaml
|
||||
|
||||
from common.pg import get_dsn
|
||||
from common.pipeline import (
|
||||
check_dependencies,
|
||||
load_lab_config,
|
||||
parse_task_args,
|
||||
remove_stale_marker,
|
||||
write_marker,
|
||||
)
|
||||
from common.relational import expected_counts
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TASK_ID = "10"
|
||||
DEPENDS_ON = ["09"]
|
||||
|
||||
SCALES_GB = (600, 1200, 6000) # raw-CSV target scales (decimal GB), spec 10
|
||||
IMPRACTICAL_S = 3600.0 # projection > 1 h, spec 10
|
||||
FAIL_S = 24 * 3600.0 # projection > 24 h, spec 10
|
||||
DISK_HEADROOM = 1.3 # 30% free-space headroom, spec 10
|
||||
CACHE_INDEX_FRACTION = 0.10 # relational hot-set: 10% of index size, spec 10
|
||||
WORKING_SET_GB = 4.0 # relational session working set, spec 10 default
|
||||
RDF_HOT_FRACTION = 0.20 # dictionary+index working set, 15-25% midpoint, spec 10
|
||||
STREAM_RAM_GB = 16.0 # flat streaming baseline (csv/json servers)
|
||||
MIN_RAM_GB = 16.0 # smallest sensible server RAM
|
||||
NODE_RAM_CEILING_GB = 1024.0 # single-node RAM threshold, spec 10 RDF note
|
||||
BASE_CORES = 4 # cores behind the measured PG parallel scans (task 09 host)
|
||||
CORE_OPTIONS = (16, 32) # priced server tiers, hw_prices.yaml
|
||||
|
||||
FORMATS = ("csv", "json", "sqlite", "pg", "rdf")
|
||||
QUERY_NUMBERS = tuple(range(1, 8))
|
||||
|
||||
# working (queryable) footprint variant per format for storage scaling
|
||||
SIZE_VARIANT = {
|
||||
"csv": ("csv", "tree"), "json": ("json", "full"), "sqlite": ("sqlite", "db"),
|
||||
"pg": ("pg", "live"), "rdf": ("rdf", "store"),
|
||||
}
|
||||
|
||||
# access-pattern law per (format, query) - spec 10 table, adjusted to the
|
||||
# ACTUAL implementations (rdf q3/q5/q7 read raw cycles; see module docstring)
|
||||
LAWS = {
|
||||
("csv", 1): "o1", ("json", 1): "o1", ("sqlite", 1): "log", ("pg", 1): "log", ("rdf", 1): "log",
|
||||
("csv", 2): "scan", ("json", 2): "scan", ("sqlite", 2): "logk", ("pg", 2): "logk", ("rdf", 2): "logk",
|
||||
("csv", 3): "scan", ("json", 3): "scan", ("sqlite", 3): "k_log_n", ("pg", 3): "k_log_n", ("rdf", 3): "scan",
|
||||
("csv", 4): "scan", ("json", 4): "scan", ("sqlite", 4): "logk", ("pg", 4): "logk", ("rdf", 4): "logk",
|
||||
("csv", 5): "scan", ("json", 5): "scan", ("sqlite", 5): "scan", ("pg", 5): "scan_parallel", ("rdf", 5): "scan",
|
||||
("csv", 6): "scan", ("json", 6): "scan", ("sqlite", 6): "k_log_n", ("pg", 6): "k_log_n", ("rdf", 6): "k_log_n",
|
||||
("csv", 7): "scan", ("json", 7): "scan", ("sqlite", 7): "scan", ("pg", 7): "scan_parallel", ("rdf", 7): "scan",
|
||||
}
|
||||
|
||||
HW_PRICES_DEFAULT = """# Hardware cost parameters for extrapolation (task 10) - June 2026 defaults.
|
||||
# Operator-editable; re-run extrapolate.py after changes (code-config-yaml).
|
||||
ram_usd_per_gb: 14.0
|
||||
nvme_usd_per_tb: 250.0
|
||||
server_16_core_usd: 6000.0
|
||||
server_32_core_usd: 10000.0
|
||||
extra_node_usd: 6000.0
|
||||
"""
|
||||
HW_PRICE_KEYS = ("ram_usd_per_gb", "nvme_usd_per_tb", "server_16_core_usd", "server_32_core_usd", "extra_node_usd")
|
||||
|
||||
|
||||
def read_medians(path: Path) -> dict[tuple[str, int], float]:
|
||||
"""Median walls of the 35 cells; abort unless every cell is OK + checksummed
|
||||
(projections built on invalid measurements are forbidden, bench-methodology).
|
||||
"""
|
||||
medians: dict[tuple[str, int], float] = {}
|
||||
with open(path, encoding="ascii", newline="") as fh:
|
||||
for row in csv.DictReader(fh):
|
||||
if row["status"] != "OK" or row["checksum_ok"] != "true":
|
||||
raise ValueError(f"{path}: cell {row['format']}/{row['query']} is {row['status']}/checksum={row['checksum_ok']} - cannot calibrate on it")
|
||||
medians[(row["format"], int(row["query"][1:]))] = float(row["wall_s_median"])
|
||||
expected_cells = {(fmt, qn) for fmt in FORMATS for qn in QUERY_NUMBERS}
|
||||
if set(medians) != expected_cells:
|
||||
raise ValueError(f"{path}: {len(medians)} cells, expected the full 35-cell matrix")
|
||||
return medians
|
||||
|
||||
|
||||
def read_sizes(path: Path) -> dict[tuple[str, str], int]:
|
||||
sizes: dict[tuple[str, str], int] = {}
|
||||
with open(path, encoding="ascii", newline="") as fh:
|
||||
reader = csv.reader(fh)
|
||||
next(reader)
|
||||
for fmt, variant, nbytes in reader:
|
||||
sizes[(fmt, variant)] = int(nbytes)
|
||||
missing = set(SIZE_VARIANT.values()) - set(sizes)
|
||||
if missing:
|
||||
raise ValueError(f"{path}: missing rows {sorted(missing)}")
|
||||
return sizes
|
||||
|
||||
|
||||
def pg_index_share(dsn: str) -> tuple[float, int]:
|
||||
"""Index share of the live PG footprint (read-only catalog query)."""
|
||||
with psycopg.connect(dsn) as conn:
|
||||
index_bytes, total_bytes = conn.execute(
|
||||
"SELECT COALESCE(sum(pg_indexes_size(c.oid)), 0)::bigint,"
|
||||
" COALESCE(sum(pg_total_relation_size(c.oid)), 0)::bigint"
|
||||
" FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace"
|
||||
" WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'm')"
|
||||
).fetchone()
|
||||
if total_bytes == 0:
|
||||
raise ValueError("pg catalog reports zero relation bytes - is the database loaded?")
|
||||
return index_bytes / total_bytes, index_bytes
|
||||
|
||||
|
||||
def load_hw_prices(config_dir: Path) -> dict:
|
||||
path = config_dir / "hw_prices.yaml"
|
||||
if not path.exists():
|
||||
path.write_text(HW_PRICES_DEFAULT, encoding="ascii", newline="\n")
|
||||
logger.info("seeded editable defaults: %s", path)
|
||||
with open(path, encoding="ascii") as fh:
|
||||
prices = yaml.safe_load(fh)
|
||||
missing = [k for k in HW_PRICE_KEYS if k not in prices]
|
||||
if missing:
|
||||
raise ValueError(f"{path}: missing keys {missing}")
|
||||
return prices
|
||||
|
||||
|
||||
def growth(law: str, r: float, n0: int, cores: int) -> float:
|
||||
depth = 1.0 + math.log(r) / math.log(n0)
|
||||
if law == "o1":
|
||||
return 1.0
|
||||
if law == "log":
|
||||
return depth
|
||||
if law == "logk":
|
||||
return r
|
||||
if law == "k_log_n":
|
||||
return r * depth
|
||||
if law == "scan":
|
||||
return r
|
||||
if law == "scan_parallel":
|
||||
return r * BASE_CORES / cores
|
||||
logger.error("growth: unknown law=%r", law)
|
||||
raise ValueError(f"unknown scaling law: {law!r}")
|
||||
|
||||
|
||||
def flag_of(seconds: float) -> str:
|
||||
if seconds > FAIL_S:
|
||||
return "FAIL"
|
||||
if seconds > IMPRACTICAL_S:
|
||||
return "IMPRACTICAL"
|
||||
return "OK"
|
||||
|
||||
|
||||
def project(fmt: str, qn: int, medians: dict, overhead: dict, r: float, n0: int, cores: int) -> float:
|
||||
variable = max(0.0, medians[(fmt, qn)] - overhead[fmt])
|
||||
return overhead[fmt] + variable * growth(LAWS[(fmt, qn)], r, n0, cores)
|
||||
|
||||
|
||||
def pick_pg_cores(medians: dict, overhead: dict, r: float, n0: int) -> int:
|
||||
"""Smallest priced tier; escalate only when it fixes an IMPRACTICAL flag
|
||||
on a parallel-scan cell (the only law that benefits from cores)."""
|
||||
for cores in CORE_OPTIONS:
|
||||
worst = max(project("pg", qn, medians, overhead, r, n0, cores) for qn in QUERY_NUMBERS if LAWS[("pg", qn)] == "scan_parallel")
|
||||
if worst <= IMPRACTICAL_S:
|
||||
return cores
|
||||
return CORE_OPTIONS[-1]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_task_args("Task 10: extrapolate measurements to 600 GB / 1.2 TB / 6 TB")
|
||||
out_root: Path = args.out_root
|
||||
bench_dir = out_root / "bench"
|
||||
try:
|
||||
check_dependencies(out_root, DEPENDS_ON)
|
||||
remove_stale_marker(out_root, TASK_ID)
|
||||
medians = read_medians(bench_dir / "results_median.csv")
|
||||
sizes = read_sizes(bench_dir / "storage_sizes.csv")
|
||||
prices = load_hw_prices(out_root / "config")
|
||||
n0 = expected_counts(load_lab_config(out_root))["friction_cycles"]
|
||||
csv_bytes = sizes[SIZE_VARIANT["csv"]]
|
||||
coeff = {fmt: sizes[SIZE_VARIANT[fmt]] / csv_bytes for fmt in FORMATS}
|
||||
index_share, pg_index_bytes = pg_index_share(get_dsn())
|
||||
logger.info("storage coefficients vs csv: %s; pg index share %.3f",
|
||||
", ".join(f"{fmt}={coeff[fmt]:.2f}" for fmt in FORMATS), index_share)
|
||||
overhead = {fmt: min(medians[(fmt, qn)] for qn in QUERY_NUMBERS) for fmt in FORMATS}
|
||||
|
||||
flags = {"OK": 0, "IMPRACTICAL": 0, "FAIL": 0}
|
||||
with open(bench_dir / "extrapolation.csv", "w", encoding="ascii", newline="") as fh:
|
||||
writer = csv.writer(fh)
|
||||
writer.writerow(["format", "query", "scale_gb", "projected_wall_s", "flag"])
|
||||
for scale_gb in SCALES_GB:
|
||||
r = scale_gb * 1e9 / csv_bytes
|
||||
cores = {fmt: CORE_OPTIONS[0] for fmt in FORMATS}
|
||||
cores["pg"] = pick_pg_cores(medians, overhead, r, n0)
|
||||
for fmt in FORMATS:
|
||||
for qn in QUERY_NUMBERS:
|
||||
seconds = project(fmt, qn, medians, overhead, r, n0, cores[fmt])
|
||||
flag = flag_of(seconds)
|
||||
flags[flag] += 1
|
||||
writer.writerow([fmt, f"q{qn}", scale_gb, round(seconds, 3), flag])
|
||||
|
||||
with open(bench_dir / "hardware_sizing.csv", "w", encoding="ascii", newline="") as fh:
|
||||
writer = csv.writer(fh)
|
||||
writer.writerow(["format", "scale_gb", "disk_tb", "ram_gb", "cores", "nodes", "est_cost_usd"])
|
||||
for scale_gb in SCALES_GB:
|
||||
r = scale_gb * 1e9 / csv_bytes
|
||||
for fmt in FORMATS:
|
||||
fmt_bytes = coeff[fmt] * scale_gb * 1e9
|
||||
disk_tb = fmt_bytes * DISK_HEADROOM / 1e12
|
||||
if fmt in ("csv", "json"):
|
||||
ram_gb = STREAM_RAM_GB
|
||||
elif fmt in ("sqlite", "pg"):
|
||||
# same index share assumed for sqlite: identical rows,
|
||||
# keys, and indexes as the PG schema (variant B)
|
||||
ram_gb = max(MIN_RAM_GB, CACHE_INDEX_FRACTION * index_share * fmt_bytes / 1e9 + WORKING_SET_GB)
|
||||
else:
|
||||
ram_gb = max(MIN_RAM_GB, RDF_HOT_FRACTION * fmt_bytes / 1e9)
|
||||
cores = pick_pg_cores(medians, overhead, r, n0) if fmt == "pg" else CORE_OPTIONS[0]
|
||||
nodes = max(1, math.ceil(ram_gb / NODE_RAM_CEILING_GB))
|
||||
server_usd = prices["server_16_core_usd"] if cores == 16 else prices["server_32_core_usd"]
|
||||
cost = (server_usd + ram_gb * prices["ram_usd_per_gb"]
|
||||
+ disk_tb * prices["nvme_usd_per_tb"]
|
||||
+ (nodes - 1) * prices["extra_node_usd"])
|
||||
writer.writerow([fmt, scale_gb, round(disk_tb, 3), round(ram_gb, 1), cores, nodes, round(cost)])
|
||||
|
||||
entries = {
|
||||
"extrapolation_rows": len(SCALES_GB) * len(FORMATS) * len(QUERY_NUMBERS),
|
||||
"sizing_rows": len(SCALES_GB) * len(FORMATS),
|
||||
"flags_ok": flags["OK"],
|
||||
"flags_impractical": flags["IMPRACTICAL"],
|
||||
"flags_fail": flags["FAIL"],
|
||||
"pg_index_share": round(index_share, 3),
|
||||
"pg_index_mb": round(pg_index_bytes / 1048576, 1),
|
||||
"note_pandas": "pandas CSV variant not measured; its RAM is O(n) - infeasible above ~10 GB (spec 10)",
|
||||
}
|
||||
for fmt in FORMATS:
|
||||
entries[f"coeff_{fmt}"] = round(coeff[fmt], 3)
|
||||
marker_path = write_marker(out_root, TASK_ID, entries)
|
||||
except Exception:
|
||||
logger.critical("task %s failed", TASK_ID, exc_info=True)
|
||||
return 1
|
||||
|
||||
print(f"task 10 ok: {entries['extrapolation_rows']} projections "
|
||||
f"(ok={flags['OK']}, impractical={flags['IMPRACTICAL']}, fail={flags['FAIL']}), "
|
||||
f"{entries['sizing_rows']} sizing rows")
|
||||
print(f"outputs: {bench_dir / 'extrapolation.csv'}, {bench_dir / 'hardware_sizing.csv'}")
|
||||
print(f"marker: {marker_path}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user