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"