manifests: compact per-version deltas (full lists reconstructed from lake_files); restore drill verifies against the restored DB
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
3 changed files +42 −16
modified
services/ingest/src/findata_ingest/cli.py
+14 −0
@@ -181,6 +181,20 @@ def create_key(email: str, name: str, scopes: str = "read:market,read:fundamenta | |||||
| 181 | typer.echo(json.dumps(create_api_key_for_email(email, name, scopes.split(",")), indent=2)) | 181 | typer.echo(json.dumps(create_api_key_for_email(email, name, scopes.split(",")), indent=2)) | ||
| 182 | 182 | ||||
| 183 | 183 | ||||
| 184 | +@app.command() | ||||
| 185 | +def manifest(dataset: str, version: int | None = None, out: Path | None = None) -> None: | ||||
| 186 | + """Write the full file manifest of a dataset version (reconstructed from lake_files) as JSON.""" | ||||
| 187 | + from findata_ingest.lake import Lake | ||||
| 188 | + files = Lake().snapshot_files(dataset, version) | ||||
| 189 | + doc = {"dataset_id": dataset, "version": version, "file_count": len(files), "rows": sum(f["rows"] for f in files), "files": files} | ||||
| 190 | + text = json.dumps(doc, indent=1) | ||||
| 191 | + if out: | ||||
| 192 | + out.write_text(text) | ||||
| 193 | + typer.echo(f"{out} ({len(files)} files)") | ||||
| 194 | + else: | ||||
| 195 | + typer.echo(text) | ||||
| 196 | + | ||||
| 197 | + | ||||
| 184 | @app.command() | 198 | @app.command() | ||
| 185 | def status() -> None: | 199 | def status() -> None: | ||
| 186 | from findata_core.db import fetch_all, fetch_one | 200 | from findata_core.db import fetch_all, fetch_one | ||
modified
services/ingest/src/findata_ingest/lake.py
+18 −3
@@ -222,13 +222,18 @@ class Lake: | |||||
| 222 | conn.execute("INSERT INTO dataset_versions(dataset_id, version, snapshot_id, file_count, row_count, bytes, ingest_run_id, note) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", | 222 | conn.execute("INSERT INTO dataset_versions(dataset_id, version, snapshot_id, file_count, row_count, bytes, ingest_run_id, note) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", | ||
| 223 | (dataset_id, version, snapshot_id, file_count, row_count, nbytes, ingest_run_id, note)) | 223 | (dataset_id, version, snapshot_id, file_count, row_count, nbytes, ingest_run_id, note)) | ||
| 224 | conn.execute("UPDATE datasets SET current_version=%s, updated_at=now() WHERE id=%s", (version, dataset_id)) | 224 | conn.execute("UPDATE datasets SET current_version=%s, updated_at=now() WHERE id=%s", (version, dataset_id)) | ||
| 225 | + # Manifest = compact delta (added/retired) + totals; the full file list of any version is reconstructed | ||||
| 226 | + # from lake_files (version_added/version_removed) with `Lake.snapshot_files()`. Writing the full list on | ||||
| 227 | + # every publish produced 83 GB of JSON during the bulk load. | ||||
| 225 | manifest = {"dataset_id": dataset_id, "version": version, "snapshot_id": snapshot_id, | 228 | manifest = {"dataset_id": dataset_id, "version": version, "snapshot_id": snapshot_id, | ||
| 226 | "created_at": datetime.now(UTC).isoformat(), "ingest_run_id": ingest_run_id, "note": note, | 229 | "created_at": datetime.now(UTC).isoformat(), "ingest_run_id": ingest_run_id, "note": note, | ||
| 227 | − "files": [{"path": r["path"], "sha256": r["sha256"], "rows": int(r["row_count"]), "bytes": int(r["bytes"])} for r in live]} | 230 | + "file_count": file_count, "row_count": row_count, "bytes": nbytes, | ||
| 231 | + "added": [{"path": f.rel_path, "sha256": f.sha256, "rows": f.row_count, "bytes": f.bytes} for f in added], | ||||
| 232 | + "retired": list(retire_paths)} | ||||
| 228 | mpath = self.root / "manifests" / dataset_id / f"v{version}.json" | 233 | mpath = self.root / "manifests" / dataset_id / f"v{version}.json" | ||
| 229 | − mpath.parent.mkdir(parents=True, exist_ok=True) | 234 | + mkdir_retry(mpath.parent) | ||
| 230 | tmp = mpath.with_suffix(".json.tmp") | 235 | tmp = mpath.with_suffix(".json.tmp") | ||
| 231 | − tmp.write_text(json.dumps(manifest, indent=1, default=str)) | 236 | + tmp.write_text(json.dumps(manifest, separators=(",", ":"), default=str)) | ||
| 232 | os.replace(tmp, mpath) | 237 | os.replace(tmp, mpath) | ||
| 233 | for p in retire_paths: | 238 | for p in retire_paths: | ||
| 234 | src = self.abs(p) | 239 | src = self.abs(p) | ||
@@ -239,6 +244,16 @@ class Lake: | |||||
| 239 | shutil.move(str(src), str(dst)) | 244 | shutil.move(str(src), str(dst)) | ||
| 240 | return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes} | 245 | return {"version": version, "snapshot_id": snapshot_id, "file_count": file_count, "row_count": row_count, "bytes": nbytes} | ||
| 241 | 246 | ||||
| 247 | + def snapshot_files(self, dataset_id: str, version: int | None = None) -> list[dict[str, Any]]: | ||||
| 248 | + """Full file list (path, sha256, rows, bytes) of a dataset at `version` (default: current), from lake_files.""" | ||||
| 249 | + from findata_core.db import connection as _conn | ||||
| 250 | + with _conn() as conn: | ||||
| 251 | + if version is None: | ||||
| 252 | + version = conn.execute("SELECT current_version FROM datasets WHERE id=%s", (dataset_id,)).fetchone()["current_version"] | ||||
| 253 | + rows = conn.execute("SELECT path, sha256, row_count, bytes FROM lake_files WHERE dataset_id=%s AND version_added <= %s " | ||||
| 254 | + "AND (version_removed IS NULL OR version_removed > %s) ORDER BY path", (dataset_id, version, version)).fetchall() | ||||
| 255 | + return [{"path": r["path"], "sha256": r["sha256"], "rows": int(r["row_count"]), "bytes": int(r["bytes"])} for r in rows] | ||||
| 256 | + | ||||
| 242 | def gc_orphans(self, older_than_hours: int = 2) -> int: | 257 | def gc_orphans(self, older_than_hours: int = 2) -> int: | ||
| 243 | """Delete curated files that are not registered in lake_files (left by a killed job before publication).""" | 258 | """Delete curated files that are not registered in lake_files (left by a killed job before publication).""" | ||
| 244 | from findata_core.db import connection as _conn | 259 | from findata_core.db import connection as _conn | ||
modified
services/ingest/src/findata_ingest/tasks/backup.py
+10 −13
@@ -65,12 +65,10 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 65 | def restore_drill(target: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw") -> dict[str, Any]: | 65 | def restore_drill(target: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw") -> dict[str, Any]: | ||
| 66 | """Restore the latest pg_dump into a scratch database and the lake into `dest`, then verify manifest hashes.""" | 66 | """Restore the latest pg_dump into a scratch database and the lake into `dest`, then verify manifest hashes.""" | ||
| 67 | import hashlib | 67 | import hashlib | ||
| 68 | − import json | ||||
| 69 | s = get_settings() | 68 | s = get_settings() | ||
| 70 | host, _, remote_dir = target.partition(":") | 69 | host, _, remote_dir = target.partition(":") | ||
| 71 | dest.mkdir(parents=True, exist_ok=True) | 70 | dest.mkdir(parents=True, exist_ok=True) | ||
| 72 | subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/curated/{dataset_id}/", f"{dest}/curated/{dataset_id}/"], check=True, env=_env(), timeout=3600) | 71 | subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/curated/{dataset_id}/", f"{dest}/curated/{dataset_id}/"], check=True, env=_env(), timeout=3600) | ||
| 73 | − subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/manifests/{dataset_id}/", f"{dest}/manifests/{dataset_id}/"], check=True, env=_env(), timeout=600) | ||||
| 74 | latest = subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"ls -1 {remote_dir}/pgdump | sort | tail -1"], capture_output=True, text=True, env=_env(), timeout=60).stdout.strip() | 72 | latest = subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"ls -1 {remote_dir}/pgdump | sort | tail -1"], capture_output=True, text=True, env=_env(), timeout=60).stdout.strip() | ||
| 75 | subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/pgdump/{latest}", f"{dest}/"], check=True, env=_env(), timeout=600) | 73 | subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/pgdump/{latest}", f"{dest}/"], check=True, env=_env(), timeout=600) | ||
| 76 | base_url = s.database_url.get_secret_value().rsplit("/", 1)[0] | 74 | base_url = s.database_url.get_secret_value().rsplit("/", 1)[0] | ||
@@ -89,18 +87,17 @@ def restore_drill(target: str, dest: Path, dataset_id: str = "market_bars.stock. | |||||
| 89 | 87 | ||||
| 90 | _sh('psql "$FD_PG_ADMIN" -qc "CREATE DATABASE \"$(basename "$FD_PG_SCRATCH")\"" && zstd -dc "$FD_DUMP" | psql -q "$FD_PG_SCRATCH" >/dev/null', 3600) | 88 | _sh('psql "$FD_PG_ADMIN" -qc "CREATE DATABASE \"$(basename "$FD_PG_SCRATCH")\"" && zstd -dc "$FD_DUMP" | psql -q "$FD_PG_SCRATCH" >/dev/null', 3600) | ||
| 91 | out = _sh(f'psql -tA "$FD_PG_SCRATCH" -c "SELECT count(*) FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 120).stdout.strip() | 89 | out = _sh(f'psql -tA "$FD_PG_SCRATCH" -c "SELECT count(*) FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 120).stdout.strip() | ||
| 92 | − manifests = sorted((dest / "manifests" / dataset_id).glob("v*.json"), key=lambda p: int(p.stem[1:])) | 90 | + listing = _sh(f'psql -tA -F "|" "$FD_PG_SCRATCH" -c "SELECT path, sha256 FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 600).stdout | ||
| 93 | verified = 0 | 91 | verified = 0 | ||
| 94 | missing = 0 | 92 | missing = 0 | ||
| 95 | − if manifests: | 93 | + files = [ln.split("|") for ln in listing.strip().splitlines() if "|" in ln] | ||
| 96 | − m = json.loads(manifests[-1].read_text()) | 94 | + for path, sha in files: | ||
| 97 | − for f in m["files"]: | 95 | + p = dest / path | ||
| 98 | − p = dest / f["path"] | 96 | + if not p.exists(): | ||
| 99 | − if not p.exists(): | 97 | + missing += 1 | ||
| 100 | − missing += 1 | 98 | + continue | ||
| 101 | − continue | 99 | + h = hashlib.sha256(p.read_bytes()).hexdigest() | ||
| 102 | − h = hashlib.sha256(p.read_bytes()).hexdigest() | 100 | + verified += int(h == sha) | ||
| 103 | − verified += int(h == f["sha256"]) | ||||
| 104 | subprocess.run(["bash", "-c", 'psql "$FD_PG_ADMIN" -qc "DROP DATABASE \"$(basename "$FD_PG_SCRATCH")\""'], env=env, timeout=120, capture_output=True) | 101 | subprocess.run(["bash", "-c", 'psql "$FD_PG_ADMIN" -qc "DROP DATABASE \"$(basename "$FD_PG_SCRATCH")\""'], env=env, timeout=120, capture_output=True) | ||
| 105 | − return {"pg_dump": latest, "restored_lake_files_in_db": out, "manifest_version": manifests[-1].stem if manifests else None, | 102 | + return {"pg_dump": latest, "restored_lake_files_in_db": out, "files_listed": len(files), | ||
| 106 | "files_hash_verified": verified, "files_missing": missing, "dest": str(dest)} | 103 | "files_hash_verified": verified, "files_missing": missing, "dest": str(dest)} | ||
| 107 | 104 | ||||