From e2bc7a926dcafaf6ab08ab5dcc312d973ce0cbb3 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Wed, 27 May 2026 06:09:21 -0400 Subject: [PATCH] #53: pre-size chunks rows at INSERT to skip phase-2 page splits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Real consumer-side bottleneck for cold-pack genesis is the phase-2 chunks-content UPDATE loop: each `UPDATE chunks SET content=? WHERE chunk_id=?` grows the row from NULL to ~500 bytes, triggering SQLite page splits, which become ext4 metadata-journal events. With 6M chunks × 4 parallel writers, those journal events serialize and dominate consumer wall time (~130 of the 164-min v3-revert run). Fix: producer dumps a synthetic `_content_size` column in chunks.jsonl carrying the on-disk byte length of each chunk's content (constant- time SQLite `length(content)`). Consumer's phase 1 INSERT pre-allocates the row with `content = bytes(_content_size)` instead of letting it default to NULL. Phase 2's UPDATE then replaces same-size bytes in-place — no row growth, no page splits, no per-row journal events. Implementation: arborist/cold_pack_metadata.py _dump_generic_table for `chunks`: Emit synthetic `_content_size` column = length(content) at the end of the columnar header. Underscore prefix avoids collision with any future schema column. _restore_routed_table: Detect `_content_size` in the JSONL header; if present (and table == chunks), build INSERT against [chunks_cols] + ['content'] and substitute a bytes(_content_size) placeholder for the content position. Phase 2 UPDATE later replaces those bytes. Forward compatibility: - Old packs (no _content_size): consumer uses today's NULL-content INSERT path. No behavior change. - New packs: consumer auto-detects, uses pre-sized path. SPV-wallet trade-off: In just-enough mode the consumer pulls only the metadata pack so chunks land with the placeholder bytes (NOT NULL anymore). That's a SEMANTIC CHANGE for SPV — `chunks.content IS NULL` no longer means "JIT-fetch later." Documented in code; if SPV-mode JIT-fetch ever ships, it must distinguish placeholder bytes (where every byte is 0) from real content. New test (TestPreSizedChunks.test_chunks_content_pre_sized_after_metadata_restore): Hydrate just-enough → chunks rows have non-NULL bytes content of correct size. Catches the regression if a future change reverts the placeholder logic. Expected impact: ~50-70% reduction in phase-2 wall time. Real number lands when the next 3090 bench-max iteration runs against re-packed bucket. 6 cold-unpack-routed tests pass. --- arborist/cold_pack_metadata.py | 72 +++++++++++++++++++++++++++++--- tests/test_cold_unpack_routed.py | 59 ++++++++++++++++++++++++++ 2 files changed, 125 insertions(+), 6 deletions(-) diff --git a/arborist/cold_pack_metadata.py b/arborist/cold_pack_metadata.py index 95c6ea4..237cfa8 100644 --- a/arborist/cold_pack_metadata.py +++ b/arborist/cold_pack_metadata.py @@ -238,7 +238,31 @@ def _dump_generic_table( Stable ORDER BY rowid (or PK if WITHOUT ROWID) so two writers at the same snapshot produce byte-identical output. Memory bounded by cursor iteration — never fetchall(). + + Special-case `chunks`: emit a synthetic ``_content_size`` column + (#53) carrying the on-disk byte length of ``content``. Consumer + uses this to pre-size the row at INSERT time, so phase-2 UPDATE + replaces same-size bytes in-place → no row growth → no page + splits → no ext4 metadata-journal events. Costs nothing on the + producer side: ``length(content)`` is constant-time on SQLite + BLOB columns (the size is in the row header). """ + if table == "chunks": + # Drop content (in COLUMN_FILTER) but ADD the synthetic + # _content_size as the last column. Naming with a leading + # underscore avoids any collision with a future schema column. + select_cols = _select_cols(conn, table) + synth = "_content_size" + col_list = ( + ", ".join(f'"{c}"' for c in select_cols) + + f', length(content) AS "{synth}"' + ) + order_by = ", ".join(f'"{c}"' for c in select_cols) + cursor = conn.execute( + f"SELECT {col_list} FROM {table} ORDER BY {order_by}" + ) + return write_columnar_jsonl(out_path, select_cols + [synth], cursor) + select_cols = _select_cols(conn, table) col_list = ", ".join(f'"{c}"' for c in select_cols) # ORDER BY all selected columns guarantees determinism for tables @@ -529,15 +553,38 @@ def _restore_routed_table( """Stream JSONL rows for ``table``, route each by ``route_col`` hash to one of ``M`` target shards, executemany INSERT into the chosen target. Mirrors ``arborist.migrate._route_per_doc_table`` but reads - from JSONL instead of a source connection.""" + from JSONL instead of a source connection. + + Chunks special case (#53): if the columnar header carries the + synthetic ``_content_size`` column, the producer dumped each + chunk's content byte-length. Consumer INSERTs with + ``content = zeroblob(_content_size)`` so phase 2 UPDATE + replaces same-size bytes in place — no row growth, no page + splits. Other tables ignore the synthetic column gracefully + (it's filtered out of the column list before building the + INSERT statement). + """ BATCH = 5000 buckets: list[list[tuple]] = [[] for _ in range(M)] - cols: list[str] | None = None + cols: list[str] | None = None # columns that participate in INSERT + csize_idx: int | None = None # position of _content_size in JSONL row + target_content_pos: int | None = None # position of `content` in cols insert_sql: str | None = None n = 0 for row in read_columnar_jsonl(in_path): if cols is None: - cols = list(row.keys()) + jsonl_cols = list(row.keys()) + # Detect the synthetic content_size column (chunks only). + has_csize = "_content_size" in jsonl_cols + if table == "chunks" and has_csize: + csize_idx = jsonl_cols.index("_content_size") + # Target's chunks columns include `content`. Build + # INSERT against all chunks columns; we'll substitute + # zeroblob(_content_size) for the content position. + cols = [c for c in jsonl_cols if c != "_content_size"] + ["content"] + target_content_pos = len(cols) - 1 + else: + cols = [c for c in jsonl_cols if c != "_content_size"] placeholders = ", ".join("?" for _ in cols) col_list = ", ".join(f'"{c}"' for c in cols) insert_sql = ( @@ -547,10 +594,23 @@ def _restore_routed_table( if not root: continue idx = _shard_idx_for_root(root, M) - buckets[idx].append(tuple(row[c] for c in cols)) + if csize_idx is not None and target_content_pos is not None: + # chunks: pad row with content=zeroblob(content_size). + # SQLite's zeroblob() makes a BLOB of the right size that + # SQLite stores efficiently (lazy materialization until + # written). Phase 2 UPDATE replaces those bytes in place. + payload = [] + for c in cols[:-1]: # everything except `content` + payload.append(row[c]) + sz = row.get("_content_size") or 0 + # placeholder bytes — zstd-compressed chunk bodies range + # ~100-2000 bytes; zeroblob is the right-sized SQLite + # BLOB primitive. Pass actual bytes so executemany works. + payload.append(bytes(int(sz))) + buckets[idx].append(tuple(payload)) + else: + buckets[idx].append(tuple(row[c] for c in cols)) n += 1 - # Flush any bucket that crossed the batch threshold so we don't - # hold millions of rows in Python heap. for i, bucket in enumerate(buckets): if len(bucket) >= BATCH and insert_sql is not None: targets[i].executemany(insert_sql, bucket) diff --git a/tests/test_cold_unpack_routed.py b/tests/test_cold_unpack_routed.py index 360f745..a7e9834 100644 --- a/tests/test_cold_unpack_routed.py +++ b/tests/test_cold_unpack_routed.py @@ -266,6 +266,65 @@ class TestFtsPackRoutingRegression: # lives in chunks_fts_docsize which is what we asserted above. +class TestPreSizedChunks: + """#53: producer dumps a synthetic _content_size column for chunks; + consumer INSERTs with right-sized BLOB placeholders so phase 2 + UPDATE replaces same-size content in place — no row growth, no + page splits, no ext4 metadata-journal events per UPDATE. + """ + + def test_chunks_content_pre_sized_after_metadata_restore( + self, producer_shard: Path, tmp_path: Path + ): + from arborist.evict import push_pack + backend = MemoryBackend() + with sqlite3.connect(str(producer_shard)) as src_conn: + src_conn.row_factory = sqlite3.Row + push_result = push_pack( + src_conn, backend, + document_root=None, + max_chunks=None, + max_pack_bytes=10 * 1024 ** 2, + allow_license_class="unknown", + include_fts=True, + ) + metadata_pack_hash = next( + p["pack_hash"] for p in push_result["packs"] + if p.get("kind") == "metadata" + ) + + # Pull ONLY the metadata pack (just-enough hydration). Verify + # chunks rows land with right-sized BLOB content (NOT NULL). + target_dir = tmp_path / "tgt" + target_dir.mkdir() + M = 4 + targets = _make_target_shards(target_dir, M) + try: + hydrate_from_metadata_pack_routed( + targets, backend, metadata_pack_hash, M=M, mode="just-enough", + ) + finally: + for t in targets: + t.close() + # Find the target with chunks rows; verify content is non-NULL + # BLOB of correct size. + for i in range(M): + tdb = target_dir / f"{i:03d}.db" + c = sqlite3.connect(f"file:{tdb}?mode=ro", uri=True) + try: + rows = c.execute( + "SELECT chunk_id, length(content) AS sz, content IS NOT NULL AS has " + "FROM chunks LIMIT 3" + ).fetchall() + finally: + c.close() + if not rows: + continue + for cid, sz, has in rows: + assert has, f"chunk {cid} on shard {i} has NULL content (#53 not applied)" + assert sz > 0, f"chunk {cid} on shard {i} has zero-size placeholder" + + class TestRoutedHydrateValidation: def test_M_mismatch_rejected(self, tmp_path: Path): target_dir = tmp_path / "tgt"