#!/usr/bin/env python3 """Controlled SQLite import observations. Only fresh temporary databases are used.""" import argparse import datetime from decimal import Decimal import hashlib import json from pathlib import Path import platform import sqlite3 import tempfile LEXEMES = [ "9007199254740993", "9007199254740993.0", "9.007199254740993e15", "00123", "9223372036854775807", ] SCHEMAS = { "ordinary_numeric": ("NUMERIC", ""), "ordinary_any": ("ANY", ""), "strict_integer": ("INTEGER", " STRICT"), "strict_any": ("ANY", " STRICT"), "strict_text": ("TEXT", " STRICT"), } TRANSPORTS = ("text", "exact_integer", "binary64") def bind_value(lexeme, transport): if transport == "text": return lexeme if transport == "binary64": return float(lexeme) if transport == "exact_integer": value = Decimal(lexeme) if not value.is_finite() or value != value.to_integral_value(): raise ValueError("exact_integer requires a finite integral value") number = int(value) if not -(2 ** 63) <= number < 2 ** 63: raise ValueError("outside SQLite signed 64-bit integer range") return number raise ValueError("unknown transport") def encoded(value): # All measured values are strings in JSON, including large integers. if isinstance(value, str): return {"python_type": "str", "value": value, "exact_decimal": str(Decimal(value))} if type(value) is int: return {"python_type": "int", "value": str(value), "exact_decimal": str(value)} if type(value) is float: return {"python_type": "float", "value": repr(value), "exact_decimal": str(Decimal.from_float(value)), "hex": value.hex()} raise TypeError("unexpected SQLite value type") def error_record(error): return {"name": error.sqlite_errorname, "code": error.sqlite_errorcode, "message": str(error)} def read_matrix(connection, attempts): rows = [] for attempt in attempts: value = connection.execute( "SELECT v, typeof(v) FROM " + attempt["schema"] + " WHERE id=?", (attempt["id"],), ).fetchone() row = dict(attempt) row["stored"] = None if value is not None: row["stored"] = {**encoded(value[0]), "storage_class": value[1]} row["numeric_equals_source"] = ( Decimal(row["stored"]["exact_decimal"]) == Decimal(row["source_lexeme"]) ) row["lexeme_equals_source"] = type(value[0]) is str and value[0] == row["source_lexeme"] rows.append(row) return rows def checks(connection): return {name: [r[0] for r in connection.execute("PRAGMA " + name)] for name in ("quick_check", "integrity_check")} def readonly(path): return sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True) def observe(work_root=None): if sqlite3.sqlite_version_info < (3, 37, 0): raise RuntimeError("SQLite >= 3.37.0 is required for STRICT") with tempfile.TemporaryDirectory(prefix="hesela-numeric-", dir=work_root) as temp: root = Path(temp) source, backup = root / "source.sqlite", root / "backup.sqlite" attempts, collisions, rejections = [], [], [] con = sqlite3.connect(source, isolation_level=None) try: con.execute("PRAGMA journal_mode=DELETE") for name, (kind, suffix) in SCHEMAS.items(): con.execute(f"CREATE TABLE {name}(id INTEGER PRIMARY KEY, v {kind}){suffix}") for lexeme in LEXEMES: for transport in TRANSPORTS: bound = bind_value(lexeme, transport) for schema in SCHEMAS: row = {"id": len(attempts) + 1, "source_lexeme": lexeme, "transport": transport, "schema": schema, "bound": encoded(bound), "error": None} try: con.execute("INSERT INTO " + schema + " VALUES(?,?)", (row["id"], bound)) except sqlite3.IntegrityError as error: row["error"] = error_record(error) attempts.append(row) for name, kind in (("collision_integer", "INTEGER"), ("collision_any", "ANY")): con.execute(f"CREATE TABLE {name}(v {kind} UNIQUE) STRICT") events = [] for lexeme in ("9007199254740992.0", "9007199254740993.0"): event = {"source_lexeme": lexeme, "error": None} try: con.execute(f"INSERT INTO {name} VALUES(?)", (lexeme,)) except sqlite3.IntegrityError as error: event["error"] = error_record(error) events.append(event) collisions.append({"table": name, "events": events}) con.execute("CREATE TABLE rejection(v INTEGER) STRICT") for lexeme in ("1.5", "abc"): try: con.execute("INSERT INTO rejection VALUES(?)", (lexeme,)) except sqlite3.IntegrityError as error: rejections.append({"source_lexeme": lexeme, "error": error_record(error)}) else: rejections.append({"source_lexeme": lexeme, "error": None}) finally: con.close() before = hashlib.sha256(source.read_bytes()).hexdigest() con = readonly(source) try: rows = read_matrix(con, attempts) source_checks = checks(con) for collision in collisions: collision["reopened_rows"] = [ {**encoded(v), "storage_class": kind} for v, kind in con.execute("SELECT v,typeof(v) FROM " + collision["table"] + " ORDER BY rowid") ] progress = [] dest = sqlite3.connect(backup) try: con.backup(dest, pages=1, progress=lambda status, remaining, total: progress.append({"status": status, "remaining": remaining, "total": total})) finally: dest.close() finally: con.close() restored = readonly(backup) try: backup_rows = read_matrix(restored, attempts) backup_checks = checks(restored) backup_collisions = { item["table"]: [{**encoded(v), "storage_class": kind} for v, kind in restored.execute( "SELECT v,typeof(v) FROM " + item["table"] + " ORDER BY rowid")] for item in collisions } finally: restored.close() after = hashlib.sha256(source.read_bytes()).hexdigest() return { "schema_version": 1, "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "kind": "controlled software observation", "environment": {"python": platform.python_version(), "sqlite": sqlite3.sqlite_version, "platform": platform.platform(), "machine": platform.machine(), "sqlite_source_id": _source_id()}, "provenance": {"generator_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "fixture": "Five explicit decimal lexemes; no external data or personal information", "license": "CC0-1.0; applies to this experiment code and generated observations", "units": "Exact numeric values or source strings; no performance measurements", "large_number_encoding": "strings, with SQLite storage class and Python type recorded"}, "schemas": {key: f"CREATE TABLE {key}(id INTEGER PRIMARY KEY, v {kind}){suffix}" for key, (kind, suffix) in SCHEMAS.items()}, "rows": rows, "collision_controls": collisions, "rejection_controls": rejections, "source_checks": source_checks, "backup": {"progress": progress, "checks": backup_checks, "matrix_equal_after_reopen": rows == backup_rows, "collision_rows": backup_collisions, "sha256": hashlib.sha256(backup.read_bytes()).hexdigest()}, "source_sha256_before": before, "source_sha256_after": after, "limits": ["Designed cases on one recorded runtime, not a frequency estimate or new vulnerability.", "No custom adapters, converters, concurrent writer, crash, WAL or storage fault.", "Source spelling and exact numeric value are separate acceptance criteria.", "A backup preserves the imported database, not discarded original digits."], } def _source_id(): with sqlite3.connect(":memory:") as con: return con.execute("SELECT sqlite_source_id()").fetchone()[0] def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--work-root", type=Path) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() result = observe(args.work_root) args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n") print(f"{len(result['rows'])} cases; SQLite {result['environment']['sqlite']}; backup equal: " f"{result['backup']['matrix_equal_after_reopen']}") if __name__ == "__main__": main()