"""Reproduce ICU collation drift using two official PostgreSQL 18.6 images. Prerequisites: pull both official image digests in IMAGES below. Creates one uniquely named volume, no published ports, and no external network. Only resources created in this run are removed in finally. """ import json import subprocess import time import uuid from pathlib import Path ROOT = Path(__file__).resolve().parent IMAGES = { "old": "postgres@sha256:bbf83594798e144bff365a9c191c5d779f181fb3135902064ab8ff11918020fc", "new": "postgres@sha256:edb0c08dc33befd0262ab3b5cf1c4f3f8cfd8b09e88c05b45adee29d6761ddb6", } PREFIXES = ["0", "1", "2", "3", "9", "½", "¼", "①", "Ⅳ", "$", "€", "₿", "a", "z", "가", "한", "Ᲊ"] COPIES = 1000 SUFFIXES = [1, 500, 1000] TARGET = "한-0500" LABEL = "blog-collation-lab" RUN_ID = "blog-collation-" + uuid.uuid4().hex[:10] VOLUME = RUN_ID + "-data" containers = [] volume_created = False report = {"experiment_date": "2026-10-28", "article_date": "2026-10-29", "run_id": RUN_ID, "images": {}, "prefixes": PREFIXES, "copies": COPIES, "phases": {}} def command(args, text=None, check=True, timeout=90): done = subprocess.run(args, input=text, capture_output=True, text=True, timeout=timeout) record = {"exit_code": done.returncode, "stdout": done.stdout, "stderr": done.stderr} if check and done.returncode: raise RuntimeError(json.dumps({"args": args, **record}, ensure_ascii=False)) return record def sql(container, statement, check=True): return command(["docker", "exec", "-i", container, "psql", "-X", "-qAt", "-U", "postgres", "-v", "ON_ERROR_STOP=1"], statement, check) def query(container, statement): response = sql(container, statement) return {"value": json.loads(response["stdout"]), "stderr": response["stderr"], "exit_code": response["exit_code"]} def start(stage): name = RUN_ID + "-" + stage command(["docker", "run", "-d", "--name", name, "--label", LABEL, "--network", "none", "-e", "POSTGRES_HOST_AUTH_METHOD=trust", "-e", "POSTGRES_INITDB_ARGS=--encoding=UTF8 --locale=C", "-e", "PGDATA=/var/lib/postgresql/data", "-v", VOLUME + ":/var/lib/postgresql/data", IMAGES[stage]]) containers.append(name) for _ in range(60): ready = sql(name, "SELECT 1;", check=False) if ready["exit_code"] == 0: return name time.sleep(0.5) raise RuntimeError("Lab database did not become ready") def literal(value): return "'" + value.replace("'", "''") + "'" def count_query(table): rows = [] for prefix in PREFIXES: for suffix in SUFFIXES: value = literal(prefix + "-" + str(suffix).zfill(4)) rows.append(f"SELECT {value} AS target, (SELECT count(*) FROM {table} WHERE value={value}) AS hits") return "SELECT json_agg(t) FROM (" + " UNION ALL ".join(rows) + ") t;" def scan_settings(strategy): indexed = strategy == "index" return (f"SET enable_seqscan={'off' if indexed else 'on'}; " f"SET enable_indexscan={'on' if indexed else 'off'}; " "SET enable_indexonlyscan=off; SET enable_bitmapscan=off; ") def phase(container): result = {"metadata": query(container, """ SELECT json_build_object( 'server',version(), 'recorded',collversion, 'actual',pg_collation_actual_version(oid), 'provider',collprovider, 'locale',colllocale, 'index_filenode',pg_relation_filenode('sample_value_idx'::regclass), 'index_md5',md5(pg_read_binary_file(pg_relation_filepath('sample_value_idx'::regclass)))) FROM pg_collation WHERE collname='lab_unicode'; """)} result["data"] = query(container, scan_settings("seq") + """ SELECT json_build_object('rows',count(*), 'digest',md5(string_agg(id::text || ':' || value, '|' ORDER BY id))) FROM samples; """) result["fresh_order"] = query(container, scan_settings("seq") + """ SELECT json_agg(value ORDER BY value) FROM samples WHERE value LIKE '%-0500'; """) for table in ("samples", "control_c"): for strategy in ("seq", "index"): result[table + "_" + strategy] = query(container, scan_settings(strategy) + count_query(table)) result["representative_rows"] = {} for strategy in ("seq", "index"): statement = "SELECT coalesce(json_agg(t),'[]') FROM (SELECT id,value FROM samples WHERE value=" + literal(TARGET) + ") t;" result["representative_rows"][strategy] = query(container, scan_settings(strategy) + statement) seq = {row["target"]: row["hits"] for row in result["samples_seq"]["value"]} result["mismatches"] = [row for row in result["samples_index"]["value"] if row["hits"] != seq[row["target"]]] return result def plan_pair(container, target): results = {} for strategy in ("seq", "index"): statement = f"SELECT id,value FROM samples WHERE value={literal(target)};" results[strategy] = sql(container, scan_settings(strategy) + "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + statement) return results try: endpoint = command(["docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"])["stdout"].strip() assert endpoint.startswith(("unix://", "npipe://")), "This lab requires a local Docker socket" for stage, image in IMAGES.items(): report["images"][stage] = json.loads(command(["docker", "image", "inspect", image])["stdout"])[0] # Keep provenance, not unrelated image config/environment. report["images"][stage] = {k: report["images"][stage][k] for k in ("Id", "RepoDigests", "Architecture", "Os")} command(["docker", "volume", "create", "--label", LABEL, VOLUME]) volume_created = True old = start("old") report["old_packages"] = command(["docker", "exec", old, "dpkg-query", "-W", "libicu72", "libc6", "postgresql-18"]) values = ",".join("(" + literal(prefix) + ")" for prefix in PREFIXES) setup = f""" CREATE COLLATION lab_unicode (provider=icu, locale='und', deterministic=true); CREATE TABLE samples (id integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY, value text COLLATE lab_unicode NOT NULL); INSERT INTO samples(value) SELECT p.prefix || '-' || lpad(n::text,4,'0') FROM (VALUES {values}) p(prefix) CROSS JOIN generate_series(1,{COPIES}) n; CREATE INDEX sample_value_idx ON samples(value); CREATE TABLE control_c (id integer PRIMARY KEY, value text COLLATE "C" NOT NULL); INSERT INTO control_c SELECT * FROM samples; CREATE INDEX control_c_value_idx ON control_c(value); CREATE EXTENSION amcheck; VACUUM ANALYZE samples; VACUUM ANALYZE control_c; CHECKPOINT; """ (ROOT / "setup.sql").write_text(setup) report["setup"] = sql(old, setup) report["phases"]["before_upgrade"] = phase(old) assert not report["phases"]["before_upgrade"]["mismatches"] report["before_plan"] = plan_pair(old, TARGET) command(["docker", "stop", "-t", "30", old]) new = start("new") report["new_packages"] = command(["docker", "exec", new, "dpkg-query", "-W", "libicu76", "libc6", "postgresql-18"]) report["phases"]["after_upgrade"] = phase(new) mismatches = report["phases"]["after_upgrade"]["mismatches"] assert mismatches, "No query mismatch observed; investigate instead of writing a result claim" target = TARGET assert any(row["target"] == TARGET for row in mismatches) report["representative_target"] = target report["after_upgrade_plan"] = plan_pair(new, target) report["amcheck_before"] = sql(new, "SELECT bt_index_check('sample_value_idx',true);", check=False) report["refresh"] = sql(new, "ALTER COLLATION lab_unicode REFRESH VERSION;") report["phases"]["after_refresh"] = phase(new) report["after_refresh_plan"] = plan_pair(new, target) report["amcheck_after_refresh"] = sql(new, "SELECT bt_index_check('sample_value_idx',true);", check=False) report["reindex"] = sql(new, "REINDEX INDEX sample_value_idx; CHECKPOINT;") report["phases"]["after_reindex"] = phase(new) report["after_reindex_plan"] = plan_pair(new, target) report["amcheck_after_reindex"] = sql(new, "SELECT bt_index_check('sample_value_idx',true);") assert not report["phases"]["after_reindex"]["mismatches"] digests = {p["data"]["value"]["digest"] for p in report["phases"].values()} assert len(digests) == 1, "Rows changed during comparison" for current in report["phases"].values(): assert current["control_c_seq"]["value"] == current["control_c_index"]["value"] report["success"] = True finally: cleanup = [] for name in containers: cleanup.append(command(["docker", "rm", "-f", name], check=False)) if volume_created: cleanup.append(command(["docker", "volume", "rm", VOLUME], check=False)) report["cleanup"] = cleanup (ROOT / "observations.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") for name, result in report["phases"].items(): print(name, result["metadata"]["value"], "mismatches=", result["mismatches"]) print("amcheck:", report["amcheck_before"], report["amcheck_after_refresh"], report["amcheck_after_reindex"])