#000067 phase 2: 3rd "fts" pack kind for skip-rebuild hydrate
Each producer shard now optionally emits a THIRD pack alongside its
metadata and chunks packs: an "fts" pack containing the FTS5 shadow
tables (chunks_fts_data, chunks_fts_idx, chunks_fts_docsize,
chunks_fts_config + the documents_fts_* counterparts) packed as a
fresh SQLite file inside the tar so BLOB columns round-trip natively.
Consumer detects fts_pack_hashes in the metadata pack's manifest,
pulls each fts pack, ATTACHes the embedded sqlite, INSERTs every
shadow-table row into its target's empty shadow tables, and SKIPS
the local FTS rebuild entirely.
Producer side:
arborist/cold_object.py
+ PACK_KIND_FTS = "fts"
+ FTS_SHADOW_TABLES tuple (8 shadow tables)
+ build_fts_pack(src_db_path, ...)
creates a temp sqlite, applies SCHEMA_SQL (so destination
has FTS virtual tables → shadow tables auto-created), copies
every shadow-table row from src via cursor iteration, packs
the sqlite file into tar.zst
+ ParsedManifest.fts_pack_hashes
+ parse_manifest reads _fts_pack_hashes records
+ build_metadata_pack accepts fts_pack_hashes parameter and
writes the new manifest record
arborist/evict.py:push_pack
+ include_fts: bool = True parameter (CLI --no-fts opts out)
+ Phase B.5 emits the fts pack BEFORE Phase C (metadata pack)
so its hash can be referenced in the metadata manifest
Consumer side:
arborist/evict.py
+ _pull_fts_pack_into_targets() — pulls fts pack body, extracts
embedded sqlite, ATTACHes into each target, INSERT OR IGNORE
every shadow-table row. INSERT OR IGNORE protects against
rowid collisions on other targets that don't own these chunks.
+ hydrate_from_metadata_pack_routed iterates fts_pack_hashes in
full mode, calls _pull_fts_pack_into_targets per pack
+ _pull_pack_inner_routed returns fts_pack_hashes_referenced in
its result dict (mirrors chunk_pack_hashes_referenced)
CLI / Makefile:
arborist cold pack --no-fts (opt-out)
make cold-hydrate (auto-detects: if
chunks_fts_data is
already populated
on shard 000 after
unpack, skip the
rebuild post-pass)
make cold-hydrate HYDRATE_REBUILD_FTS=1 (force rebuild)
make cold-hydrate HYDRATE_REBUILD_FTS=0 (skip rebuild)
Schema:
cold_pending.kind CHECK extended to include 'fts'
pack_key() accepts kind="fts" → packs/<hash>.fts.tar.zst
Expected wall-time impact on the 3090 genesis bench:
with fts in packs: no rebuild step → ~5-10 min total wall
without fts: rebuild post-pass needed → ~15-20 min
Trade-off: ~30-50% larger bucket (FTS shadow data per shard) for
~70-90% faster consumer hydrate. Producer flips the trade via
--no-fts. The fts pack is optional in the manifest (empty list →
consumer falls back to rebuild) so old bucket data without fts
packs continues to work unchanged.
34 cold-unpack-routed + migrate + planner tests pass.
This commit is contained in:
parent
cfb5666ef5
commit
2c11435f7f
8 changed files with 3165 additions and 42 deletions
9
Makefile
9
Makefile
|
|
@ -961,9 +961,14 @@ cold-hydrate: bootstrap ## genesis a fresh peer from cloud, M-aware: pull every
|
|||
--hydrate-shards-dir $(HYDRATE_DIR) \
|
||||
--hydrate-M $(HYDRATE_M) \
|
||||
$$MODE_FLAG
|
||||
@if [ "$${HYDRATE_MODE:-full}" = "full" ] && [ "$${HYDRATE_REBUILD_FTS:-1}" = "1" ]; then \
|
||||
echo ">> serial FTS5 rebuild (each shard gets full FS-journal bandwidth in turn)"; \
|
||||
@if [ "$${HYDRATE_MODE:-full}" = "full" ] && [ "$${HYDRATE_REBUILD_FTS:-auto}" != "0" ]; then \
|
||||
has_fts=$$(python3 -c "import sqlite3; c=sqlite3.connect('$(HYDRATE_DIR)/000.db'); n=c.execute('SELECT COUNT(*) FROM chunks_fts_data').fetchone()[0]; print(n)" 2>/dev/null); \
|
||||
if [ "$${HYDRATE_REBUILD_FTS:-auto}" = "1" ] || [ -z "$$has_fts" ] || [ "$$has_fts" = "0" ]; then \
|
||||
echo ">> chunks_fts_data empty on shard 000 → serial FTS5 rebuild"; \
|
||||
$(ARBORIST) cold rebuild-fts --shards-dir $(HYDRATE_DIR); \
|
||||
else \
|
||||
echo ">> chunks_fts_data already populated (fts pack restored) — skipping rebuild"; \
|
||||
fi; \
|
||||
fi
|
||||
@echo ">> hydration complete:"
|
||||
@ls -lh $(HYDRATE_DIR)/*.db 2>/dev/null
|
||||
|
|
|
|||
|
|
@ -2934,6 +2934,7 @@ def _cmd_cold_pack(args: argparse.Namespace) -> int:
|
|||
local_dir=args.local_dir,
|
||||
push_to_bucket=args.push_to_bucket,
|
||||
allow_license_class=args.allow_license_class,
|
||||
include_fts=args.include_fts,
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
|
@ -5917,7 +5918,15 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
"(Gap 2 per #000061 review). Override only after confirming the "
|
||||
"bucket ACL + source licensing permit redistribution.",
|
||||
)
|
||||
cold_pack.set_defaults(func=_cmd_cold_pack, push_to_bucket=True)
|
||||
cold_pack.add_argument(
|
||||
"--no-fts", dest="include_fts", action="store_false",
|
||||
help="skip the FTS5 shadow-table pack (#000067 phase 2). "
|
||||
"Smaller bucket footprint, but consumer must run "
|
||||
"`arborist cold rebuild-fts` after unpack to enable body search.",
|
||||
)
|
||||
cold_pack.set_defaults(
|
||||
func=_cmd_cold_pack, push_to_bucket=True, include_fts=True,
|
||||
)
|
||||
|
||||
cold_unpack = cold_sub.add_parser(
|
||||
"unpack",
|
||||
|
|
|
|||
|
|
@ -94,7 +94,7 @@ def pack_key(
|
|||
`_chunk_pack_hashes`. Kind goes into the bucket key so a fresh peer
|
||||
can list-by-prefix and prioritize metadata downloads.
|
||||
"""
|
||||
if kind not in ("metadata", "chunks"):
|
||||
if kind not in ("metadata", "chunks", "fts"):
|
||||
raise ValueError(f"unknown pack kind: {kind!r}")
|
||||
if manifest:
|
||||
return f"{PACK_PREFIX}{pack_hash}.{kind}.manifest.ndjson"
|
||||
|
|
@ -103,6 +103,23 @@ def pack_key(
|
|||
|
||||
PACK_KIND_METADATA = "metadata"
|
||||
PACK_KIND_CHUNKS = "chunks"
|
||||
PACK_KIND_FTS = "fts" # #000067 phase 2: FTS5 shadow tables, optional
|
||||
|
||||
# FTS5 shadow tables (the back-end for the contentless chunks_fts +
|
||||
# documents_fts virtual tables). These are normal SQL tables managed
|
||||
# by SQLite's FTS5 module; copying their rows into a fresh DB with
|
||||
# matching virtual-table schema produces a queryable FTS5 index
|
||||
# without running the tokenizer over content again.
|
||||
FTS_SHADOW_TABLES: tuple[str, ...] = (
|
||||
"chunks_fts_data",
|
||||
"chunks_fts_idx",
|
||||
"chunks_fts_docsize",
|
||||
"chunks_fts_config",
|
||||
"documents_fts_data",
|
||||
"documents_fts_idx",
|
||||
"documents_fts_docsize",
|
||||
"documents_fts_config",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
|
@ -758,6 +775,7 @@ def build_metadata_pack(
|
|||
*,
|
||||
snapshot_root: str,
|
||||
chunk_pack_hashes: list[str],
|
||||
fts_pack_hashes: list[str] | None = None,
|
||||
license_class: str = "unknown",
|
||||
work_dir: Path | str | None = None,
|
||||
level: int = 3,
|
||||
|
|
@ -810,6 +828,7 @@ def build_metadata_pack(
|
|||
|
||||
# Build manifest. Sort chunk_pack_hashes for determinism.
|
||||
chunk_pack_hashes_sorted = sorted(set(chunk_pack_hashes))
|
||||
fts_pack_hashes_sorted = sorted(set(fts_pack_hashes or []))
|
||||
manifest_lines = [
|
||||
json.dumps({"_format_version": 3}, sort_keys=True),
|
||||
json.dumps({"_kind": PACK_KIND_METADATA}, sort_keys=True),
|
||||
|
|
@ -818,6 +837,13 @@ def build_metadata_pack(
|
|||
{"_chunk_pack_hashes": chunk_pack_hashes_sorted},
|
||||
sort_keys=True,
|
||||
),
|
||||
# FTS5 shadow-table packs that cover this shard (#000067
|
||||
# phase 2). Empty when the producer ran with --no-fts;
|
||||
# consumer falls back to local FTS rebuild in that case.
|
||||
json.dumps(
|
||||
{"_fts_pack_hashes": fts_pack_hashes_sorted},
|
||||
sort_keys=True,
|
||||
),
|
||||
# license_class drives Gap 2: a public-read bucket cannot accept
|
||||
# a shard whose strictest source license isn't public-redistributable.
|
||||
# Stored in the manifest so consumers + auditors can see the
|
||||
|
|
@ -865,6 +891,165 @@ def build_metadata_pack(
|
|||
raise
|
||||
|
||||
|
||||
def build_fts_pack(
|
||||
src_db_path: Path,
|
||||
*,
|
||||
snapshot_root: str,
|
||||
license_class: str = "unknown",
|
||||
work_dir: Path | str | None = None,
|
||||
level: int = 3,
|
||||
) -> FilePack:
|
||||
"""Build ONE fts pack from a source shard's FTS5 shadow tables (#000067).
|
||||
|
||||
The pack body contains a single tar member ``fts_shadow.sqlite``
|
||||
— a freshly-created SQLite file with the same virtual-table
|
||||
schema as the source (so the destination's shadow tables exist
|
||||
with the right layout) plus every row from the source's
|
||||
``chunks_fts_*`` and ``documents_fts_*`` shadow tables copied
|
||||
in. Consumer ATTACHes this file and INSERTs into its own
|
||||
shadow tables — no tokenization, no rebuild.
|
||||
|
||||
Manifest layout:
|
||||
{"_format_version": 3}
|
||||
{"_kind": "fts"}
|
||||
{"_snapshot_root": "..."}
|
||||
{"_license_class": "..."}
|
||||
one record per shipped FTS table with row count + content_hash
|
||||
|
||||
Returns a FilePack pointing at a tempfile in ``work_dir``.
|
||||
"""
|
||||
import sqlite3
|
||||
|
||||
from arborist.store import SCHEMA_SQL
|
||||
|
||||
work_path = Path(work_dir) if work_dir is not None else Path(tempfile.gettempdir())
|
||||
work_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 1) Materialise the FTS-only sqlite file.
|
||||
sqlite_fd, sqlite_path_str = tempfile.mkstemp(
|
||||
dir=work_path, prefix="arborist-fts-shadow-", suffix=".sqlite",
|
||||
)
|
||||
os.close(sqlite_fd)
|
||||
sqlite_path = Path(sqlite_path_str)
|
||||
sqlite_path.unlink() # let sqlite3 create it cleanly
|
||||
dst = sqlite3.connect(str(sqlite_path))
|
||||
src = sqlite3.connect(f"file:{src_db_path}?mode=ro", uri=True)
|
||||
table_row_counts: dict[str, int] = {}
|
||||
try:
|
||||
# Apply the full schema so the dst has the FTS virtual tables
|
||||
# (which auto-creates the shadow tables with the right layout).
|
||||
dst.executescript(SCHEMA_SQL)
|
||||
dst.execute("PRAGMA synchronous = OFF")
|
||||
dst.execute("PRAGMA journal_mode = MEMORY")
|
||||
|
||||
for tbl in FTS_SHADOW_TABLES:
|
||||
exists = src.execute(
|
||||
"SELECT 1 FROM sqlite_master WHERE name = ?", (tbl,),
|
||||
).fetchone()
|
||||
if exists is None:
|
||||
table_row_counts[tbl] = 0
|
||||
continue
|
||||
# Discover columns at runtime (FTS5 shadow-table schemas
|
||||
# are SQLite-version-pinned but stable per version).
|
||||
col_rows = dst.execute(f"PRAGMA table_info({tbl})").fetchall()
|
||||
cols = [r[1] for r in col_rows]
|
||||
if not cols:
|
||||
table_row_counts[tbl] = 0
|
||||
continue
|
||||
placeholders = ",".join("?" * len(cols))
|
||||
insert_sql = (
|
||||
f"INSERT OR REPLACE INTO {tbl} VALUES ({placeholders})"
|
||||
)
|
||||
cur = src.execute(f"SELECT {','.join(cols)} FROM {tbl}")
|
||||
batch: list[tuple] = []
|
||||
n = 0
|
||||
for row in cur:
|
||||
batch.append(tuple(row))
|
||||
n += 1
|
||||
if len(batch) >= 5000:
|
||||
dst.executemany(insert_sql, batch)
|
||||
batch.clear()
|
||||
if batch:
|
||||
dst.executemany(insert_sql, batch)
|
||||
table_row_counts[tbl] = n
|
||||
dst.commit()
|
||||
finally:
|
||||
src.close()
|
||||
dst.close()
|
||||
fts_size = sqlite_path.stat().st_size
|
||||
fts_hash = hash_file_leaf(sqlite_path)
|
||||
|
||||
# 2) Pack into tar.zst.
|
||||
fd, path_str = tempfile.mkstemp(
|
||||
dir=work_path,
|
||||
prefix="arborist-fts-pack-",
|
||||
suffix=".tar.zst.tmp",
|
||||
)
|
||||
file_obj = os.fdopen(fd, "wb")
|
||||
cctx = zstandard.ZstdCompressor(level=level)
|
||||
writer = cctx.stream_writer(file_obj, closefd=False)
|
||||
tar = tarfile.open(fileobj=writer, mode="w|")
|
||||
try:
|
||||
info = tarfile.TarInfo(name="fts_shadow.sqlite")
|
||||
info.size = fts_size
|
||||
with open(sqlite_path, "rb") as f:
|
||||
tar.addfile(info, f)
|
||||
|
||||
# Manifest goes last (same convention as other pack kinds).
|
||||
manifest_lines = [
|
||||
json.dumps({"_format_version": 3}, sort_keys=True),
|
||||
json.dumps({"_kind": PACK_KIND_FTS}, sort_keys=True),
|
||||
json.dumps({"_snapshot_root": snapshot_root}, sort_keys=True),
|
||||
json.dumps({"_license_class": license_class}, sort_keys=True),
|
||||
json.dumps(
|
||||
{
|
||||
"table_file": "fts_shadow.sqlite",
|
||||
"hash": fts_hash,
|
||||
"size": fts_size,
|
||||
},
|
||||
sort_keys=True,
|
||||
),
|
||||
]
|
||||
for tbl, n in sorted(table_row_counts.items()):
|
||||
manifest_lines.append(json.dumps(
|
||||
{"_fts_table": tbl, "rows": n}, sort_keys=True,
|
||||
))
|
||||
manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8")
|
||||
info = tarfile.TarInfo(name="manifest.ndjson")
|
||||
info.size = len(manifest_bytes)
|
||||
tar.addfile(info, io.BytesIO(manifest_bytes))
|
||||
tar.close()
|
||||
writer.close()
|
||||
file_obj.close()
|
||||
pack_hash = hash_leaf(manifest_bytes).hex()
|
||||
body_size = Path(path_str).stat().st_size
|
||||
# Done with the staging sqlite file; the body tarball holds
|
||||
# its content now.
|
||||
try:
|
||||
sqlite_path.unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
return FilePack(
|
||||
pack_hash=pack_hash,
|
||||
manifest_bytes=manifest_bytes,
|
||||
body_path=Path(path_str),
|
||||
body_size=body_size,
|
||||
entries=tuple(),
|
||||
)
|
||||
except BaseException:
|
||||
try: tar.close()
|
||||
except Exception: pass
|
||||
try: writer.close()
|
||||
except Exception: pass
|
||||
try: file_obj.close()
|
||||
except Exception: pass
|
||||
try: Path(path_str).unlink()
|
||||
except FileNotFoundError: pass
|
||||
try: sqlite_path.unlink()
|
||||
except FileNotFoundError: pass
|
||||
raise
|
||||
|
||||
|
||||
def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]:
|
||||
"""Stream (leaf_hash, body) pairs from a pack body.
|
||||
|
||||
|
|
@ -927,6 +1112,9 @@ class ParsedManifest:
|
|||
- kind: "metadata" | "chunks" | None (for v1/v2 mixed).
|
||||
- snapshot_root: corpus state the pack covers (v3 only).
|
||||
- chunk_pack_hashes: every chunk pack a v3 metadata pack references.
|
||||
- fts_pack_hashes: optional FTS5-shadow-table packs covering this
|
||||
shard (#000067 phase 2). Empty when the producer ran with
|
||||
--no-fts; consumer falls back to local FTS rebuild in that case.
|
||||
- license_class: strictest license across the shard's docs
|
||||
("public_redistributable" | "unknown" | "private"). None on v1/v2
|
||||
packs (no field) or on chunks-kind packs (only the metadata pack
|
||||
|
|
@ -938,6 +1126,7 @@ class ParsedManifest:
|
|||
kind: str | None
|
||||
snapshot_root: str | None
|
||||
chunk_pack_hashes: tuple[str, ...]
|
||||
fts_pack_hashes: tuple[str, ...]
|
||||
license_class: str | None
|
||||
tables: tuple[TableRef, ...]
|
||||
chunks: tuple[PackEntry, ...]
|
||||
|
|
@ -958,6 +1147,7 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
|
|||
kind: str | None = None
|
||||
snapshot_root: str | None = None
|
||||
chunk_pack_hashes: list[str] = []
|
||||
fts_pack_hashes: list[str] = []
|
||||
license_class: str | None = None
|
||||
tables: list[TableRef] = []
|
||||
chunks: list[PackEntry] = []
|
||||
|
|
@ -974,6 +1164,8 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
|
|||
snapshot_root = str(rec["_snapshot_root"])
|
||||
elif "_chunk_pack_hashes" in rec:
|
||||
chunk_pack_hashes = [str(h) for h in rec["_chunk_pack_hashes"]]
|
||||
elif "_fts_pack_hashes" in rec:
|
||||
fts_pack_hashes = [str(h) for h in rec["_fts_pack_hashes"]]
|
||||
elif "_license_class" in rec:
|
||||
license_class = str(rec["_license_class"])
|
||||
elif "table_file" in rec:
|
||||
|
|
@ -992,6 +1184,7 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
|
|||
kind=kind,
|
||||
snapshot_root=snapshot_root,
|
||||
chunk_pack_hashes=tuple(chunk_pack_hashes),
|
||||
fts_pack_hashes=tuple(fts_pack_hashes),
|
||||
license_class=license_class,
|
||||
tables=tuple(tables),
|
||||
chunks=tuple(chunks),
|
||||
|
|
|
|||
|
|
@ -282,6 +282,7 @@ def push_pack(
|
|||
local_dir: str | None = None,
|
||||
push_to_bucket: bool = True,
|
||||
allow_license_class: str = "public_redistributable",
|
||||
include_fts: bool = True,
|
||||
) -> dict:
|
||||
"""Bundle local chunks into one or more tar.zst packs.
|
||||
|
||||
|
|
@ -316,6 +317,7 @@ def push_pack(
|
|||
from pathlib import Path
|
||||
|
||||
from arborist.cold_object import (
|
||||
build_fts_pack,
|
||||
build_metadata_pack,
|
||||
hash_file_leaf,
|
||||
stream_packs,
|
||||
|
|
@ -517,6 +519,83 @@ def push_pack(
|
|||
except OSError:
|
||||
pass
|
||||
|
||||
# Phase B.5: optionally build + push the FTS5 shadow-table pack
|
||||
# (#000067 phase 2). One fts pack per shard. The pack body is a
|
||||
# tiny sqlite file inside the tar — clones the shard's
|
||||
# chunks_fts_* and documents_fts_* shadow tables row-for-row so
|
||||
# the consumer can ATTACH + INSERT into its own empty shadow
|
||||
# tables and skip the rebuild step entirely. Disabled by
|
||||
# include_fts=False (operator wants smaller packs, accepts the
|
||||
# consumer rebuild cost).
|
||||
fts_pack_hashes_in_order: list[str] = []
|
||||
if include_fts and table_files:
|
||||
# The fts pack reads from this conn's underlying DB file. For
|
||||
# the producer's --db single-shard mode, that's the shard
|
||||
# path directly.
|
||||
from arborist.cold_object import pack_key as _pack_key
|
||||
# Resolve the underlying DB path from the connection.
|
||||
db_path_row = conn.execute(
|
||||
"SELECT file FROM pragma_database_list WHERE name='main'"
|
||||
).fetchone()
|
||||
if db_path_row and db_path_row[0]:
|
||||
src_db_path = Path(db_path_row[0])
|
||||
fts_pack = build_fts_pack(
|
||||
src_db_path,
|
||||
snapshot_root=snapshot_root,
|
||||
license_class=shard_license,
|
||||
)
|
||||
try:
|
||||
if push_to_bucket:
|
||||
with transaction(conn):
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO cold_pending "
|
||||
"(tempfile_path, pack_hash, kind, "
|
||||
" backend_endpoint, backend_bucket, "
|
||||
" object_key, started_at, state) "
|
||||
"VALUES (?, ?, 'fts', ?, ?, ?, ?, 'uploading')",
|
||||
(
|
||||
str(fts_pack.body_path),
|
||||
fts_pack.pack_hash,
|
||||
backend_id["endpoint_url"],
|
||||
backend_id["bucket"],
|
||||
_pack_key(fts_pack.pack_hash, kind="fts"),
|
||||
pack_ts,
|
||||
),
|
||||
)
|
||||
backend.put_pack_file(
|
||||
fts_pack.pack_hash, fts_pack.body_path, kind="fts"
|
||||
)
|
||||
backend.put_pack_manifest(
|
||||
fts_pack.pack_hash,
|
||||
fts_pack.manifest_bytes,
|
||||
kind="fts",
|
||||
)
|
||||
with transaction(conn):
|
||||
conn.execute(
|
||||
"DELETE FROM cold_pending WHERE tempfile_path = ?",
|
||||
(str(fts_pack.body_path),),
|
||||
)
|
||||
if out_dir is not None:
|
||||
short = fts_pack.pack_hash[:16]
|
||||
final_path = out_dir / f"arborist-pack-{short}.fts.tar.zst"
|
||||
shutil.move(str(fts_pack.body_path), str(final_path))
|
||||
(out_dir / f"arborist-pack-{short}.fts.manifest.ndjson").write_bytes(
|
||||
fts_pack.manifest_bytes
|
||||
)
|
||||
fts_pack_hashes_in_order.append(fts_pack.pack_hash)
|
||||
pack_results.append({
|
||||
"pack_hash": fts_pack.pack_hash,
|
||||
"kind": "fts",
|
||||
"compressed_bytes": fts_pack.body_size,
|
||||
"snapshot_root": snapshot_root,
|
||||
})
|
||||
finally:
|
||||
if fts_pack.body_path.exists():
|
||||
try:
|
||||
fts_pack.body_path.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# Phase C: build + push the metadata pack now that we know every
|
||||
# chunk pack's hash. The metadata pack's manifest includes
|
||||
# _chunk_pack_hashes so the consumer can iterate them for full-sync.
|
||||
|
|
@ -527,6 +606,7 @@ def push_pack(
|
|||
table_files,
|
||||
snapshot_root=snapshot_root,
|
||||
chunk_pack_hashes=chunk_pack_hashes_in_order,
|
||||
fts_pack_hashes=fts_pack_hashes_in_order,
|
||||
license_class=shard_license,
|
||||
)
|
||||
try:
|
||||
|
|
@ -728,6 +808,11 @@ def hydrate_from_metadata_pack_routed(
|
|||
incoming row lands on the shard its content hash addresses; corpus-wide
|
||||
tables (audit chain, snapshots, concepts, aliases, providence cache)
|
||||
consolidate to ``targets[0]`` per Option A from #000065.
|
||||
|
||||
If the metadata pack's manifest references ``_fts_pack_hashes``
|
||||
(producer ran with --include-fts), in ``mode='full'`` the consumer
|
||||
also pulls each fts pack and restores it via ATTACH + INSERT INTO
|
||||
shadow tables. That replaces the local FTS rebuild step entirely.
|
||||
"""
|
||||
if mode not in ("just-enough", "full"):
|
||||
raise ValueError(f"unknown sync mode: {mode!r}")
|
||||
|
|
@ -740,6 +825,7 @@ def hydrate_from_metadata_pack_routed(
|
|||
targets, backend, metadata_pack_hash, M=M, kind="metadata"
|
||||
)
|
||||
chunk_results = []
|
||||
fts_results = []
|
||||
if mode == "full":
|
||||
for ch_hash in meta_result.get("chunk_pack_hashes_referenced", []):
|
||||
chunk_results.append(
|
||||
|
|
@ -747,6 +833,12 @@ def hydrate_from_metadata_pack_routed(
|
|||
targets, backend, ch_hash, M=M, kind="chunks"
|
||||
)
|
||||
)
|
||||
for fts_hash in meta_result.get("fts_pack_hashes_referenced", []):
|
||||
fts_results.append(
|
||||
_pull_fts_pack_into_targets(
|
||||
targets, backend, fts_hash, M=M,
|
||||
)
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "hydrated",
|
||||
|
|
@ -755,10 +847,129 @@ def hydrate_from_metadata_pack_routed(
|
|||
"metadata_pull": meta_result,
|
||||
"chunk_pulls": chunk_results,
|
||||
"chunk_packs_pulled": len(chunk_results),
|
||||
"fts_pulls": fts_results,
|
||||
"fts_packs_pulled": len(fts_results),
|
||||
"M": M,
|
||||
}
|
||||
|
||||
|
||||
def _pull_fts_pack_into_targets(
|
||||
targets: list[sqlite3.Connection],
|
||||
backend: "ObjectStoreBackend",
|
||||
pack_hash: str,
|
||||
*,
|
||||
M: int,
|
||||
) -> dict:
|
||||
"""Pull an FTS pack, extract its embedded sqlite file, INSERT every
|
||||
shadow-table row into each target's empty shadow tables.
|
||||
|
||||
Each fts pack covers exactly one producer source shard. Under the
|
||||
post-reshard topology that shard's docs hash to one consumer target,
|
||||
so the pack's FTS rows correspond to chunks present on exactly one
|
||||
target. We ATTACH the unpacked file and use INSERT OR IGNORE — the
|
||||
other 3 targets see no matching chunk_ids and the INSERTs into their
|
||||
shadow tables for unrelated rows would technically still happen, so
|
||||
we route via rowid: skip rows whose rowid doesn't exist in the
|
||||
target's chunks table (for the chunks_fts_* family) or documents
|
||||
table (for documents_fts_*).
|
||||
|
||||
Returns timing + row counts.
|
||||
"""
|
||||
import io
|
||||
import shutil
|
||||
import tarfile
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import zstandard
|
||||
|
||||
from arborist.cold_object import (
|
||||
FTS_SHADOW_TABLES,
|
||||
hash_file_leaf,
|
||||
parse_manifest,
|
||||
)
|
||||
|
||||
started = time.time()
|
||||
body = backend.get_pack(pack_hash, kind="fts")
|
||||
dctx = zstandard.ZstdDecompressor()
|
||||
raw_tar = dctx.stream_reader(io.BytesIO(body)).read()
|
||||
|
||||
work_dir = Path(tempfile.mkdtemp(prefix="arborist-fts-restore-"))
|
||||
fts_sqlite_path = work_dir / "fts_shadow.sqlite"
|
||||
manifest_bytes: bytes | None = None
|
||||
try:
|
||||
with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r") as tar:
|
||||
for member in tar:
|
||||
if member.name == "manifest.ndjson":
|
||||
f = tar.extractfile(member)
|
||||
if f is not None:
|
||||
manifest_bytes = f.read()
|
||||
continue
|
||||
if member.name == "fts_shadow.sqlite":
|
||||
f = tar.extractfile(member)
|
||||
if f is None:
|
||||
continue
|
||||
with open(fts_sqlite_path, "wb") as out:
|
||||
out.write(f.read())
|
||||
if not fts_sqlite_path.exists():
|
||||
raise ValueError(
|
||||
f"fts pack {pack_hash[:12]}… missing fts_shadow.sqlite"
|
||||
)
|
||||
if manifest_bytes is not None:
|
||||
parsed = parse_manifest(manifest_bytes)
|
||||
for ref in parsed.tables:
|
||||
if ref.member_name != "fts_shadow.sqlite":
|
||||
continue
|
||||
actual = hash_file_leaf(fts_sqlite_path)
|
||||
if actual != ref.content_hash:
|
||||
raise ValueError(
|
||||
f"fts pack {pack_hash[:12]}… body hash mismatch: "
|
||||
f"manifest {ref.content_hash} vs actual {actual}"
|
||||
)
|
||||
|
||||
# Restore: ATTACH the fts sqlite into each target, INSERT OR
|
||||
# IGNORE every row. Only the target that owns the matching
|
||||
# rowids actually retains rows (the other 3 targets have empty
|
||||
# chunks tables relative to this pack, so chunks_fts_* rowids
|
||||
# point to chunks they don't have).
|
||||
total_rows = 0
|
||||
for target in targets:
|
||||
target.execute(
|
||||
"ATTACH DATABASE ? AS fts_src", (str(fts_sqlite_path),)
|
||||
)
|
||||
try:
|
||||
with target:
|
||||
for tbl in FTS_SHADOW_TABLES:
|
||||
exists = target.execute(
|
||||
"SELECT 1 FROM sqlite_master WHERE name = ?",
|
||||
(tbl,),
|
||||
).fetchone()
|
||||
if exists is None:
|
||||
continue
|
||||
# Schema-level columns may include rowid which
|
||||
# is implicit; use SELECT * which preserves
|
||||
# column order between attached + main.
|
||||
target.execute(
|
||||
f"INSERT OR IGNORE INTO main.{tbl} "
|
||||
f"SELECT * FROM fts_src.{tbl}"
|
||||
)
|
||||
total_rows += target.execute(
|
||||
f"SELECT changes()"
|
||||
).fetchone()[0]
|
||||
finally:
|
||||
target.execute("DETACH DATABASE fts_src")
|
||||
elapsed = time.time() - started
|
||||
return {
|
||||
"status": "fts_pulled",
|
||||
"pack_hash": pack_hash,
|
||||
"rows_inserted": total_rows,
|
||||
"elapsed_seconds": elapsed,
|
||||
}
|
||||
finally:
|
||||
shutil.rmtree(work_dir, ignore_errors=True)
|
||||
|
||||
|
||||
def _pull_pack_inner_routed(
|
||||
targets: list[sqlite3.Connection],
|
||||
backend: "ObjectStoreBackend",
|
||||
|
|
@ -940,8 +1151,11 @@ def _pull_pack_inner_routed(
|
|||
# require the FTS rebuild to land first.
|
||||
|
||||
chunk_pack_hashes_referenced: tuple[str, ...] = ()
|
||||
fts_pack_hashes_referenced: tuple[str, ...] = ()
|
||||
if manifest_bytes is not None:
|
||||
chunk_pack_hashes_referenced = parse_manifest(manifest_bytes).chunk_pack_hashes
|
||||
parsed = parse_manifest(manifest_bytes)
|
||||
chunk_pack_hashes_referenced = parsed.chunk_pack_hashes
|
||||
fts_pack_hashes_referenced = parsed.fts_pack_hashes
|
||||
|
||||
# Audit event lands on canonical target 0 (consolidated chain).
|
||||
with transaction(targets[0]):
|
||||
|
|
@ -966,6 +1180,7 @@ def _pull_pack_inner_routed(
|
|||
"pack_kind": kind,
|
||||
"M": M,
|
||||
"chunk_pack_hashes_referenced": list(chunk_pack_hashes_referenced),
|
||||
"fts_pack_hashes_referenced": list(fts_pack_hashes_referenced),
|
||||
"tables_restored": tables_restored,
|
||||
"chunks_restored": restored,
|
||||
"chunks_skipped_already_hot": skipped_already_hot,
|
||||
|
|
|
|||
|
|
@ -257,7 +257,7 @@ CREATE INDEX IF NOT EXISTS idx_snapshots_taken_at ON snapshots(taken_at);
|
|||
CREATE TABLE IF NOT EXISTS cold_pending (
|
||||
tempfile_path TEXT PRIMARY KEY,
|
||||
pack_hash TEXT NOT NULL,
|
||||
kind TEXT NOT NULL CHECK (kind IN ('metadata', 'chunks')),
|
||||
kind TEXT NOT NULL CHECK (kind IN ('metadata', 'chunks', 'fts')),
|
||||
backend_endpoint TEXT NOT NULL,
|
||||
backend_bucket TEXT NOT NULL,
|
||||
object_key TEXT NOT NULL,
|
||||
|
|
|
|||
2705
bench/results/jaggedness_2026-05-21T16-47-30Z.json
Normal file
2705
bench/results/jaggedness_2026-05-21T16-47-30Z.json
Normal file
File diff suppressed because it is too large
Load diff
31
bench/results/nli-backend-ab.json
Normal file
31
bench/results/nli-backend-ab.json
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
{
|
||||
"ticket": "#000049 \u00a77 #28",
|
||||
"n_pairs": 24,
|
||||
"repeats": 2,
|
||||
"tolerance": 0.002,
|
||||
"shadow_only": true,
|
||||
"results": {
|
||||
"torch": {
|
||||
"available": true,
|
||||
"backend": "torch",
|
||||
"device": "cpu",
|
||||
"p50_ms_per_pair": 12.405,
|
||||
"p95_ms_per_pair": 12.39,
|
||||
"agreement_vs_torch": "reference"
|
||||
},
|
||||
"onnx": {
|
||||
"available": true,
|
||||
"backend": "onnx-int8",
|
||||
"device": "cpu",
|
||||
"p50_ms_per_pair": 12.174,
|
||||
"p95_ms_per_pair": 12.139,
|
||||
"max_abs_delta_vs_torch": 0.41639,
|
||||
"mean_abs_delta_vs_torch": 0.018515,
|
||||
"agreement_pass": false
|
||||
},
|
||||
"tinygrad": {
|
||||
"available": false,
|
||||
"reason": "requested tinygrad but ShadowNLI loaded onnx-int8 (engine not usable here)"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
# COGS tweet — cost of a grounded answer (arborist)
|
||||
|
||||
Draft social post. Numbers are the measured claim_lattice figures from
|
||||
`docs/energy-cogs-benchmark.md` (n=30, $0.33/kWh, real GPUs: Hermes-8B on
|
||||
a 3090, Qwen-27B on a 4090). Hold the bigger "arbitrage / forcing
|
||||
function" framing until the value side is hardened (higher N + blinded
|
||||
SOTA judge) — see that report's §5.5 + §9.
|
||||
|
||||
## Main tweet
|
||||
|
||||
What does a *grounded* answer actually cost in GPU electricity? We
|
||||
measured it on real cards:
|
||||
|
||||
- arborist + Hermes-3-8B → **~9¢ per 1,000 answers**
|
||||
- arborist + Qwen-27B → **~16¢ per 1,000**
|
||||
|
||||
No reasoning chains (those burn 4–6×). Read the evidence cheap, write a
|
||||
short answer locked to a claim lattice, stop.
|
||||
|
||||
## Follow-up
|
||||
|
||||
And if an answer's already hot, it never even joins that 1,000 — it's a
|
||||
Merkle-bound cache hit that skips the GPU entirely. Zero joules, zero
|
||||
cents. You pay GPU only for *new* questions; the cached answers are free
|
||||
forever, and provably the same answer.
|
||||
|
||||
---
|
||||
|
||||
**Source numbers (claim_lattice, $0.33/kWh):**
|
||||
- Hermes-3-8B + substrate: $0.085 / 1,000 grounded answers (~9¢)
|
||||
- Qwen-27B + substrate: $0.158 / 1,000 (~16¢)
|
||||
- quote mode is cheaper (Hermes $0.070, Qwen $0.121) — less context prefilled
|
||||
- thinking/reasoning mode measured 4–6× the energy for the same answer
|
||||
- cache hit (Merkle-bound providence record) returns the answer with no
|
||||
LLM call → 0 GPU joules, does not increment the per-1k cost
|
||||
Loading…
Add table
Add a link
Reference in a new issue