fix(cli): address round 2 review feedback on per_node.db emitter

even-wei · claude · even-wei · commit 6766eca56a10 · 2026-04-24T14:15:58.000+08:00
Critical:
- Phantom env='base' rows when only one env has artifacts. The emit loop
  keyed off `manifest is None`, but in the single-env path-swap case both
  curr_manifest and base_manifest are loaded pointing at the SAME env —
  so the missing-env guard never fired and we shipped duplicate rows under
  a fabricated env label. The emit loop now uses `has_target` / `has_base`,
  matching the CLL cache loop's truth source. Added a CLI test that
  reproduces the path-swap scenario and asserts no env='base' rows ship.

Addressed Copilot suggestion:
- Skip per_node.db emission entirely when Cloud doesn't advertise
  `per_node_db_url`. Fetch upload URLs first; only emit the SQLite file
  when there's somewhere to upload it. Saves 1-2s of pure-waste emit work
  against older Cloud servers during the migration window.

Docs:
- `extract_rows_from_artifacts` docstring no longer says "tests-as-columns"
  — columns come from each node's manifest `columns` metadata; tests are
  derived separately into `node_tests` via `_derive_tests`.
- Updated PR description: schema is rollback-journal (not WAL) mode.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
Signed-off-by: even-wei &lt;evenwei@infuseai.io&gt;
diff --git a/recce/cli.py b/recce/cli.py
@@ -686,51 +686,81 @@ def _stream_download_to_file(url: str, dest: Path) -> int:
     if is_cloud:
         per_node_scratch = Path(tempfile.mkdtemp(prefix="recce-per-node-"))
         try:
-            per_node_db_path: Optional[Path] = per_node_scratch / "per_node.db"
-            console.print("\n[bold]Emitting per-node SQLite...[/bold]")
-            t_pn_start = time.perf_counter()
-            try:
-                with PerNodeDbWriter(per_node_db_path) as writer:
-                    writer.write_meta(
-                        schema_version=str(PER_NODE_DB_SCHEMA_VERSION),
-                        session_id=session_id or "",
-                        recce_version=__version__,
-                        generated_at=str(int(time.time())),
-                    )
-                    envs_to_emit = [
-                        ("current", dbt_adapter.curr_manifest, dbt_adapter.curr_catalog),
-                        ("base", dbt_adapter.base_manifest, dbt_adapter.base_catalog),
-                    ]
-                    for env_name, manifest, catalog in envs_to_emit:
-                        if manifest is None:
-                            continue
-                        manifest_dict = manifest.to_dict() if hasattr(manifest, "to_dict") else manifest
-                        catalog_dict = (
-                            catalog.to_dict() if (catalog is not None and hasattr(catalog, "to_dict")) else catalog
-                        )
-                        node_rows, column_rows, edge_rows, test_rows = extract_rows_from_artifacts(
-                            manifest_dict, catalog_dict, env_name
-                        )
-                        writer.write_nodes(node_rows)
-                        writer.write_columns(column_rows)
-                        writer.write_edges(edge_rows)
-                        writer.write_tests(test_rows)
-                pn_elapsed = time.perf_counter() - t_pn_start
-                pn_size_mb = per_node_db_path.stat().st_size / 1024 / 1024
-                console.print(
-                    f"  per_node.db saved to [bold]{per_node_db_path}[/bold] ({pn_size_mb:.1f} MB, {pn_elapsed:.1f}s)"
-                )
-            except Exception as e:
-                logger.warning("[recce init] Failed to emit per_node.db: %s", e)
-                console.print(f"  [[yellow]Warning[/yellow]] Failed to emit per_node.db: {e}")
-                per_node_db_path = None
-
-            # Upload results to Cloud.
             if cloud_client:
                 console.print("\n[bold]Uploading results to Cloud...[/bold]")
                 upload_failures: list[str] = []
+                upload_urls: Optional[dict] = None
                 try:
                     upload_urls = cloud_client.get_upload_urls_by_session_id(cloud_org_id, cloud_project_id, session_id)
+                except Exception as e:
+                    logger.warning("[recce init] Cloud upload failed: %s", e)
+                    console.print(f"  [[yellow]Warning[/yellow]] Cloud upload failed: {e}")
+
+                if upload_urls is not None:
+                    # Emit per_node.db only when Cloud declares support for it.
+                    # Against an older Cloud without per_node_db_url, emitting
+                    # the SQLite file is pure waste — it is a cloud-only
+                    # artifact with no local consumer.
+                    per_node_db_upload_url = upload_urls.get("per_node_db_url")
+                    per_node_db_path: Optional[Path] = None
+                    if per_node_db_upload_url:
+                        per_node_db_path = per_node_scratch / "per_node.db"
+                        console.print("\n[bold]Emitting per-node SQLite...[/bold]")
+                        t_pn_start = time.perf_counter()
+                        try:
+                            with PerNodeDbWriter(per_node_db_path) as writer:
+                                writer.write_meta(
+                                    schema_version=str(PER_NODE_DB_SCHEMA_VERSION),
+                                    session_id=session_id or "",
+                                    recce_version=__version__,
+                                    generated_at=str(int(time.time())),
+                                )
+                                # Use has_target / has_base to match the CLL
+                                # cache loop above. When only one env has
+                                # artifacts, context_kwargs path-swaps the
+                                # missing path to the present one so
+                                # load_context doesn't fail — so both manifests
+                                # are non-None but represent the SAME env. The
+                                # flags are the only truth about which env
+                                # actually has artifacts.
+                                envs_to_emit = []
+                                if has_target:
+                                    envs_to_emit.append(
+                                        ("current", dbt_adapter.curr_manifest, dbt_adapter.curr_catalog)
+                                    )
+                                if has_base:
+                                    envs_to_emit.append(("base", dbt_adapter.base_manifest, dbt_adapter.base_catalog))
+                                for env_name, manifest, catalog in envs_to_emit:
+                                    if manifest is None:
+                                        continue
+                                    manifest_dict = manifest.to_dict() if hasattr(manifest, "to_dict") else manifest
+                                    catalog_dict = (
+                                        catalog.to_dict()
+                                        if (catalog is not None and hasattr(catalog, "to_dict"))
+                                        else catalog
+                                    )
+                                    node_rows, column_rows, edge_rows, test_rows = extract_rows_from_artifacts(
+                                        manifest_dict, catalog_dict, env_name
+                                    )
+                                    writer.write_nodes(node_rows)
+                                    writer.write_columns(column_rows)
+                                    writer.write_edges(edge_rows)
+                                    writer.write_tests(test_rows)
+                            pn_elapsed = time.perf_counter() - t_pn_start
+                            pn_size_mb = per_node_db_path.stat().st_size / 1024 / 1024
+                            console.print(
+                                f"  per_node.db saved to [bold]{per_node_db_path}[/bold] "
+                                f"({pn_size_mb:.1f} MB, {pn_elapsed:.1f}s)"
+                            )
+                        except Exception as e:
+                            logger.warning("[recce init] Failed to emit per_node.db: %s", e)
+                            console.print(f"  [[yellow]Warning[/yellow]] Failed to emit per_node.db: {e}")
+                            per_node_db_path = None
+                    else:
+                        console.print(
+                            "  [[yellow]Warning[/yellow]] No per_node_db_url in upload URLs "
+                            "(Cloud server may need update) — skipping per_node.db emit"
+                        )
 
                     # Upload CLL map
                     cll_map_upload_url = upload_urls.get("cll_map_url")
@@ -762,10 +792,7 @@ def _stream_download_to_file(url: str, dest: Path) -> int:
                             "(Cloud server may need update)"
                         )
 
-                    # Upload per_node.db. Graceful degradation: if Cloud hasn't added
-                    # the per_node_db_url key yet, log a warning and continue — this
-                    # keeps old CLI versions compatible with new Cloud and vice versa.
-                    per_node_db_upload_url = upload_urls.get("per_node_db_url")
+                    # Upload per_node.db (only when Cloud supports it AND we emitted).
                     if per_node_db_upload_url and per_node_db_path and per_node_db_path.is_file():
                         try:
                             with open(per_node_db_path, "rb") as f:
@@ -789,11 +816,6 @@ def _stream_download_to_file(url: str, dest: Path) -> int:
                         except requests.RequestException as e:
                             upload_failures.append("per_node.db")
                             console.print(f"  [[yellow]Warning[/yellow]] Failed to upload per_node.db: {e}")
-                    elif not per_node_db_upload_url:
-                        console.print(
-                            "  [[yellow]Warning[/yellow]] No per_node_db_url in upload URLs "
-                            "(Cloud server may need update)"
-                        )
 
                     # Upload CLL cache. cll_cache.db is load-bearing across sessions —
                     # build_full_cll_map reuses its warm entries on subsequent runs —
@@ -832,9 +854,6 @@ def _stream_download_to_file(url: str, dest: Path) -> int:
                         )
                     else:
                         console.print("[bold green]Cloud upload complete.[/bold green]")
-                except Exception as e:
-                    logger.warning("[recce init] Cloud upload failed: %s", e)
-                    console.print(f"  [[yellow]Warning[/yellow]] Cloud upload failed: {e}")
         finally:
             # Always remove the per_node.db scratch dir — it is throwaway per
             # invocation. cll_cache.db lives at ~/.recce/cll_cache.db (or the
diff --git a/recce/util/per_node_db.py b/recce/util/per_node_db.py
@@ -332,8 +332,10 @@ def extract_rows_from_artifacts(
 
     - ``manifest`` must be a raw dict (e.g. ``json.load(manifest.json)``).
     - ``catalog`` may be ``None``; when absent, all ``data_type`` are ``None``.
-    - Column names come from the catalog when available; otherwise from the
-      manifest's ``columns`` section (tests-as-columns).
+    - Column names come from the catalog when available; otherwise from each
+      node's ``columns`` metadata in the manifest (model/source column
+      definitions). Tests are NOT part of columns — they are derived
+      separately into ``node_tests`` rows via ``_derive_tests``.
     """
     catalog_nodes = ((catalog or {}).get("nodes") or {}) if catalog else {}
     node_rows: list[NodeRow] = []
diff --git a/tests/test_cli_per_node_db.py b/tests/test_cli_per_node_db.py
@@ -559,6 +559,118 @@ def __init__(self, *_a, **_kw):
         assert any("cll_map.json" in url for url in put_calls), put_calls
         assert any("cll_cache.db" in url for url in put_calls), put_calls
 
+    @patch("recce.core.load_context")
+    def test_no_base_rows_when_only_target_has_artifacts(self, mock_load_context, runner, tmp_path, tmp_db):
+        """Phantom env='base' guard: when only target/ is populated, the
+        earlier path-swap in ``init`` points target-base at target so
+        ``load_context`` doesn't fail — both manifests end up loaded but
+        represent the SAME (current) env. The emit loop must key off
+        has_target / has_base, not ``manifest is None``, otherwise the
+        current env's rows would ship in per_node.db under the fabricated
+        env='base' label.
+        """
+        import sqlite3 as _sqlite3
+
+        # Only current-env manifest is downloaded. base_download_urls is empty
+        # → target-base/manifest.json is never written → has_base == False.
+        manifest_bytes = b'{"nodes": {}, "child_map": {}}'
+        mock_client = _make_mock_cloud_client(
+            download_urls={"manifest_url": "https://s3.example.com/manifest.json"},
+            base_download_urls={},
+            upload_urls={
+                "cll_map_url": "https://s3.example.com/upload/cll_map.json",
+                "per_node_db_url": "https://s3.example.com/upload/per_node.db",
+                "cll_cache_url": "https://s3.example.com/upload/cll_cache.db",
+            },
+        )
+
+        nodes = {"model.test.a": _make_mock_node(raw_code="SELECT a FROM src")}
+        adapter = _make_mock_adapter(nodes)
+        adapter.get_cll_cached.return_value = MagicMock()
+        mock_cll_map = MagicMock()
+        mock_cll_map.nodes = {}
+        mock_cll_map.columns = {}
+        mock_cll_map.model_dump.return_value = {"nodes": {}, "columns": {}}
+        adapter.build_full_cll_map.return_value = mock_cll_map
+
+        # Simulate the path-swap outcome: both manifests are loaded, both
+        # point at the SAME object (the current-env manifest). `.nodes` on the
+        # manifest object must carry `.resource_type` attributes because the
+        # CLL cache loop iterates it; `.to_dict()` is what our emitter sees.
+        shared_manifest_dict = {
+            "nodes": {
+                "model.test.a": {
+                    "name": "a",
+                    "resource_type": "model",
+                    "package_name": "test",
+                    "columns": {"id": {"name": "id"}},
+                }
+            },
+            "child_map": {"model.test.a": []},
+        }
+        shared_manifest = MagicMock()
+        shared_manifest.to_dict.return_value = shared_manifest_dict
+        shared_manifest.nodes = {"model.test.a": _make_mock_node(raw_code="SELECT a FROM src")}
+        shared_manifest.metadata.adapter_type = "duckdb"
+        adapter.curr_manifest = shared_manifest
+        adapter.base_manifest = shared_manifest  # Same object — what the path-swap produces.
+        adapter.curr_catalog = None
+        adapter.base_catalog = None
+
+        mock_ctx = MagicMock()
+        mock_ctx.adapter = adapter
+        mock_load_context.return_value = mock_ctx
+
+        captured: dict[str, bytes] = {}
+
+        def capture_put(url, data=None, **kwargs):
+            if "per_node.db" in url:
+                captured["per_node_db"] = data.read() if hasattr(data, "read") else (data or b"")
+            return _make_mock_response(200)
+
+        with (
+            patch("recce.util.recce_cloud.RecceCloud", return_value=mock_client),
+            patch("requests.get", side_effect=lambda *a, **kw: _make_mock_response(200, manifest_bytes)),
+            patch("requests.put", side_effect=capture_put),
+            patch(
+                "recce.adapter.dbt_adapter.DbtAdapter._serialize_cll_data",
+                return_value='{"nodes":{}, "columns":{}, "parent_map":{}}',
+            ),
+        ):
+            result = runner.invoke(
+                cli,
+                [
+                    "init",
+                    "--cloud",
+                    "--cloud-token",
+                    "ghp_testtoken",
+                    "--session-id",
+                    "sess-1",
+                    "--cache-db",
+                    tmp_db,
+                    "--project-dir",
+                    str(tmp_path),
+                ],
+                catch_exceptions=False,
+            )
+
+        assert result.exit_code == 0, result.output
+        assert "per_node_db" in captured, "per_node.db was never PUT"
+
+        # Inspect the uploaded bytes directly — the scratch dir is already
+        # cleaned up by the CLI's finally block.
+        capture_path = tmp_path / "uploaded_per_node.db"
+        capture_path.write_bytes(captured["per_node_db"])
+        conn = _sqlite3.connect(str(capture_path))
+        try:
+            base_count = conn.execute("SELECT COUNT(*) FROM nodes WHERE env = 'base'").fetchone()[0]
+            current_count = conn.execute("SELECT COUNT(*) FROM nodes WHERE env = 'current'").fetchone()[0]
+        finally:
+            conn.close()
+
+        assert base_count == 0, f"phantom base rows: got {base_count}, expected 0"
+        assert current_count >= 1, f"expected ≥1 current-env node, got {current_count}"
+
 
 # ---------------------------------------------------------------------------
 # 2. Local-mode non-regression
