Skip to content

Commit 56478cc

Browse files
author
Even Wei
authored
Merge pull request #1334 from DataRecce/feature/drc-3295-pr1-per-node-db-emitter
feat: precompute per_node.db in recce init --cloud [DRC-3295]
2 parents ab86991 + 6766eca commit 56478cc

6 files changed

Lines changed: 2144 additions & 61 deletions

File tree

‎pyproject.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ profile = "black"
105105
[tool.pytest.ini_options]
106106
markers = [
107107
"asyncio: mark test as an async test (used by pytest-asyncio)",
108+
"scale: mark test as a scale/perf validation (skipped unless RECCE_SCALE_TESTS=1)",
108109
]
109110
asyncio_default_fixture_loop_scope = "function"
110111
addopts = "--ignore=tests/adapter/dbt_adapter/test_proj/dbt_packages"

‎recce/cli.py‎

Lines changed: 181 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import os
22
from pathlib import Path
3-
from typing import List
3+
from typing import List, Optional
44

55
import click
66

@@ -318,16 +318,23 @@ def init(cache_db, **kwargs):
318318

319319
import json
320320
import logging
321+
import shutil
321322
import tempfile
322323
import time
323324

324325
import requests
325326
from rich.console import Console
326327
from rich.progress import Progress
327328

329+
from recce import __version__
328330
from recce.adapter.dbt_adapter import DbtAdapter
329331
from recce.core import load_context
330332
from recce.util.cll import _DEFAULT_DB_PATH, CllCache, get_cll_cache, set_cll_cache
333+
from recce.util.per_node_db import SCHEMA_VERSION as PER_NODE_DB_SCHEMA_VERSION
334+
from recce.util.per_node_db import (
335+
PerNodeDbWriter,
336+
extract_rows_from_artifacts,
337+
)
331338

332339
logger = logging.getLogger("recce")
333340
console = Console()
@@ -671,73 +678,187 @@ def _stream_download_to_file(url: str, dest: Path) -> int:
671678
stats = cache.stats
672679
console.print(f"\nCache saved to [bold]{cache_db}[/bold] ({stats['entries']} entries)")
673680

674-
# Upload results to Cloud if in cloud mode
675-
if is_cloud and cloud_client:
676-
console.print("\n[bold]Uploading results to Cloud...[/bold]")
677-
upload_failures = []
681+
# In cloud mode, emit per_node.db — a pure-artifact SQLite that Cloud
682+
# streams to serve lineage without proxying to an ephemeral Recce instance.
683+
# The scratch dir is always cleaned up, even on upload failure, so
684+
# long-lived Cloud deploys don't accumulate recce-per-node-* directories
685+
# in /tmp on retries.
686+
if is_cloud:
687+
per_node_scratch = Path(tempfile.mkdtemp(prefix="recce-per-node-"))
678688
try:
679-
upload_urls = cloud_client.get_upload_urls_by_session_id(cloud_org_id, cloud_project_id, session_id)
680-
681-
# Upload CLL map
682-
cll_map_upload_url = upload_urls.get("cll_map_url")
683-
if cll_map_upload_url and cll_map_path.is_file():
689+
if cloud_client:
690+
console.print("\n[bold]Uploading results to Cloud...[/bold]")
691+
upload_failures: list[str] = []
692+
upload_urls: Optional[dict] = None
684693
try:
685-
with open(cll_map_path, "rb") as f:
686-
resp = requests.put(
687-
cll_map_upload_url,
688-
data=f,
689-
headers={"Content-Type": "application/json"},
690-
timeout=_UPLOAD_TIMEOUT,
691-
)
692-
if resp.status_code in (200, 204):
693-
console.print(f" Uploaded cll_map.json ({cll_map_path.stat().st_size / 1024 / 1024:.1f} MB)")
694+
upload_urls = cloud_client.get_upload_urls_by_session_id(cloud_org_id, cloud_project_id, session_id)
695+
except Exception as e:
696+
logger.warning("[recce init] Cloud upload failed: %s", e)
697+
console.print(f" [[yellow]Warning[/yellow]] Cloud upload failed: {e}")
698+
699+
if upload_urls is not None:
700+
# Emit per_node.db only when Cloud declares support for it.
701+
# Against an older Cloud without per_node_db_url, emitting
702+
# the SQLite file is pure waste — it is a cloud-only
703+
# artifact with no local consumer.
704+
per_node_db_upload_url = upload_urls.get("per_node_db_url")
705+
per_node_db_path: Optional[Path] = None
706+
if per_node_db_upload_url:
707+
per_node_db_path = per_node_scratch / "per_node.db"
708+
console.print("\n[bold]Emitting per-node SQLite...[/bold]")
709+
t_pn_start = time.perf_counter()
710+
try:
711+
with PerNodeDbWriter(per_node_db_path) as writer:
712+
writer.write_meta(
713+
schema_version=str(PER_NODE_DB_SCHEMA_VERSION),
714+
session_id=session_id or "",
715+
recce_version=__version__,
716+
generated_at=str(int(time.time())),
717+
)
718+
# Use has_target / has_base to match the CLL
719+
# cache loop above. When only one env has
720+
# artifacts, context_kwargs path-swaps the
721+
# missing path to the present one so
722+
# load_context doesn't fail — so both manifests
723+
# are non-None but represent the SAME env. The
724+
# flags are the only truth about which env
725+
# actually has artifacts.
726+
envs_to_emit = []
727+
if has_target:
728+
envs_to_emit.append(
729+
("current", dbt_adapter.curr_manifest, dbt_adapter.curr_catalog)
730+
)
731+
if has_base:
732+
envs_to_emit.append(("base", dbt_adapter.base_manifest, dbt_adapter.base_catalog))
733+
for env_name, manifest, catalog in envs_to_emit:
734+
if manifest is None:
735+
continue
736+
manifest_dict = manifest.to_dict() if hasattr(manifest, "to_dict") else manifest
737+
catalog_dict = (
738+
catalog.to_dict()
739+
if (catalog is not None and hasattr(catalog, "to_dict"))
740+
else catalog
741+
)
742+
node_rows, column_rows, edge_rows, test_rows = extract_rows_from_artifacts(
743+
manifest_dict, catalog_dict, env_name
744+
)
745+
writer.write_nodes(node_rows)
746+
writer.write_columns(column_rows)
747+
writer.write_edges(edge_rows)
748+
writer.write_tests(test_rows)
749+
pn_elapsed = time.perf_counter() - t_pn_start
750+
pn_size_mb = per_node_db_path.stat().st_size / 1024 / 1024
751+
console.print(
752+
f" per_node.db saved to [bold]{per_node_db_path}[/bold] "
753+
f"({pn_size_mb:.1f} MB, {pn_elapsed:.1f}s)"
754+
)
755+
except Exception as e:
756+
logger.warning("[recce init] Failed to emit per_node.db: %s", e)
757+
console.print(f" [[yellow]Warning[/yellow]] Failed to emit per_node.db: {e}")
758+
per_node_db_path = None
694759
else:
695-
upload_failures.append("cll_map.json")
696760
console.print(
697-
f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: HTTP {resp.status_code}"
761+
" [[yellow]Warning[/yellow]] No per_node_db_url in upload URLs "
762+
"(Cloud server may need update) — skipping per_node.db emit"
698763
)
699-
except requests.RequestException as e:
700-
upload_failures.append("cll_map.json")
701-
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: {e}")
702-
elif not cll_map_upload_url:
703-
console.print(
704-
" [[yellow]Warning[/yellow]] No cll_map_url in upload URLs (Cloud server may need update)"
705-
)
706-
707-
# Upload CLL cache
708-
cll_cache_upload_url = upload_urls.get("cll_cache_url")
709-
if cll_cache_upload_url and Path(cache_db).is_file():
710-
try:
711-
with open(cache_db, "rb") as f:
712-
resp = requests.put(
713-
cll_cache_upload_url,
714-
data=f,
715-
headers={"Content-Type": "application/octet-stream"},
716-
timeout=_UPLOAD_TIMEOUT,
764+
765+
# Upload CLL map
766+
cll_map_upload_url = upload_urls.get("cll_map_url")
767+
if cll_map_upload_url and cll_map_path.is_file():
768+
try:
769+
with open(cll_map_path, "rb") as f:
770+
resp = requests.put(
771+
cll_map_upload_url,
772+
data=f,
773+
headers={"Content-Type": "application/json"},
774+
timeout=_UPLOAD_TIMEOUT,
775+
)
776+
if resp.status_code in (200, 204):
777+
console.print(
778+
f" Uploaded cll_map.json ({cll_map_path.stat().st_size / 1024 / 1024:.1f} MB)"
779+
)
780+
else:
781+
upload_failures.append("cll_map.json")
782+
console.print(
783+
f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: "
784+
f"HTTP {resp.status_code}"
785+
)
786+
except requests.RequestException as e:
787+
upload_failures.append("cll_map.json")
788+
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_map.json: {e}")
789+
elif not cll_map_upload_url:
790+
console.print(
791+
" [[yellow]Warning[/yellow]] No cll_map_url in upload URLs "
792+
"(Cloud server may need update)"
717793
)
718-
if resp.status_code in (200, 204):
719-
console.print(f" Uploaded cll_cache.db ({Path(cache_db).stat().st_size / 1024 / 1024:.1f} MB)")
720-
else:
721-
upload_failures.append("cll_cache.db")
794+
795+
# Upload per_node.db (only when Cloud supports it AND we emitted).
796+
if per_node_db_upload_url and per_node_db_path and per_node_db_path.is_file():
797+
try:
798+
with open(per_node_db_path, "rb") as f:
799+
resp = requests.put(
800+
per_node_db_upload_url,
801+
data=f,
802+
headers={"Content-Type": "application/octet-stream"},
803+
timeout=_UPLOAD_TIMEOUT,
804+
)
805+
if resp.status_code in (200, 204):
806+
console.print(
807+
f" Uploaded per_node.db "
808+
f"({per_node_db_path.stat().st_size / 1024 / 1024:.1f} MB)"
809+
)
810+
else:
811+
upload_failures.append("per_node.db")
812+
console.print(
813+
f" [[yellow]Warning[/yellow]] Failed to upload per_node.db: "
814+
f"HTTP {resp.status_code}"
815+
)
816+
except requests.RequestException as e:
817+
upload_failures.append("per_node.db")
818+
console.print(f" [[yellow]Warning[/yellow]] Failed to upload per_node.db: {e}")
819+
820+
# Upload CLL cache. cll_cache.db is load-bearing across sessions —
821+
# build_full_cll_map reuses its warm entries on subsequent runs —
822+
# so Cloud uploads it alongside per_node.db.
823+
cll_cache_upload_url = upload_urls.get("cll_cache_url")
824+
if cll_cache_upload_url and Path(cache_db).is_file():
825+
try:
826+
with open(cache_db, "rb") as f:
827+
resp = requests.put(
828+
cll_cache_upload_url,
829+
data=f,
830+
headers={"Content-Type": "application/octet-stream"},
831+
timeout=_UPLOAD_TIMEOUT,
832+
)
833+
if resp.status_code in (200, 204):
834+
console.print(
835+
f" Uploaded cll_cache.db "
836+
f"({Path(cache_db).stat().st_size / 1024 / 1024:.1f} MB)"
837+
)
838+
else:
839+
upload_failures.append("cll_cache.db")
840+
console.print(
841+
f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: "
842+
f"HTTP {resp.status_code}"
843+
)
844+
except requests.RequestException as e:
845+
upload_failures.append("cll_cache.db")
846+
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: {e}")
847+
elif not cll_cache_upload_url:
848+
logger.debug("No cll_cache_url in upload URLs — cache upload not supported yet")
849+
850+
if upload_failures:
722851
console.print(
723-
f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: HTTP {resp.status_code}"
852+
f"[bold yellow]Cloud upload completed with warnings[/bold yellow] "
853+
f"(failed: {', '.join(upload_failures)})"
724854
)
725-
except requests.RequestException as e:
726-
upload_failures.append("cll_cache.db")
727-
console.print(f" [[yellow]Warning[/yellow]] Failed to upload cll_cache.db: {e}")
728-
elif not cll_cache_upload_url:
729-
logger.debug("No cll_cache_url in upload URLs — cache upload not supported yet")
730-
731-
if upload_failures:
732-
console.print(
733-
f"[bold yellow]Cloud upload completed with warnings[/bold yellow] "
734-
f"(failed: {', '.join(upload_failures)})"
735-
)
736-
else:
737-
console.print("[bold green]Cloud upload complete.[/bold green]")
738-
except Exception as e:
739-
logger.warning("[recce init] Cloud upload failed: %s", e)
740-
console.print(f" [[yellow]Warning[/yellow]] Cloud upload failed: {e}")
855+
else:
856+
console.print("[bold green]Cloud upload complete.[/bold green]")
857+
finally:
858+
# Always remove the per_node.db scratch dir — it is throwaway per
859+
# invocation. cll_cache.db lives at ~/.recce/cll_cache.db (or the
860+
# user-provided --cache-db) and is NOT touched here.
861+
shutil.rmtree(per_node_scratch, ignore_errors=True)
741862
else:
742863
console.print("Run [bold]recce server --enable-cll-cache[/bold] to use the cached lineage.")
743864

0 commit comments

Comments
 (0)