Add catalogs tracking (#12747)

* add tracking for catalog count

* add catalog_type tracking

* add changie

* fix failing ut

* fix type hints

Author: ash2shukla

.changes/unreleased/Features-20260330-195959.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add catalogs.yml usage tracking
+time: 2026-03-30T19:59:59.334539+05:30
+custom:
+    Author: ash2shukla
+    Issue: "1122"

core/dbt/cli/main.py

@@ -213,6 +213,7 @@ def build(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -281,6 +282,7 @@ def docs_generate(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -352,6 +354,7 @@ def compile(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -402,6 +405,7 @@ def show(ctx, **kwargs):
             ctx.obj["flags"],
             ctx.obj["runtime_config"],
             ctx.obj["manifest"],
+            catalogs=ctx.obj.get("catalogs"),
         )
 
     results = task.run()
@@ -527,6 +531,7 @@ def list(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -593,6 +598,7 @@ def run(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -659,6 +665,7 @@ def clone(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -691,6 +698,7 @@ def run_operation(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -727,6 +735,7 @@ def seed(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
     results = task.run()
     success = task.interpret_results(results)
@@ -761,6 +770,7 @@ def snapshot(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -804,6 +814,7 @@ def freshness(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()
@@ -847,6 +858,7 @@ def test(ctx, **kwargs):
         ctx.obj["flags"],
         ctx.obj["runtime_config"],
         ctx.obj["manifest"],
+        catalogs=ctx.obj.get("catalogs"),
     )
 
     results = task.run()

core/dbt/cli/requires.py

@@ -411,6 +411,8 @@ def setup_manifest(ctx: Context, write: bool = True, write_perf_info: bool = Fal
     catalogs = load_catalogs(flags.PROJECT_DIR, ctx.obj["project"].project_name, flags.VARS)
     active_integrations = [get_active_write_integration(catalog) for catalog in catalogs]
 
+    ctx.obj["catalogs"] = catalogs
+
     # if a manifest has already been set on the context, don't overwrite it
     if ctx.obj.get("manifest") is None:
         ctx.obj["manifest"] = parse_manifest(

core/dbt/compilation.py

@@ -3,7 +3,18 @@
 import os
 import pickle
 from collections import defaultdict, deque
-from typing import Any, Dict, Iterable, List, Optional, Set, Tuple
+from typing import (
+    Any,
+    Dict,
+    Iterable,
+    List,
+    Literal,
+    Optional,
+    Sequence,
+    Set,
+    Tuple,
+    Union,
+)
 
 import networkx as nx  # type: ignore
 import sqlparse
@@ -50,11 +61,18 @@
 
 graph_file_name = "graph.gpickle"
 
+StatsKeyType = Union[NodeType, Literal["catalogs"]]
+
 
-def print_compile_stats(stats: Dict[NodeType, int]):
+def print_compile_stats(stats: Dict[StatsKeyType, int]):
     # create tracking event for resource_counts
     if dbt.tracking.active_user is not None:
-        resource_counts = {k.pluralize(): v for k, v in stats.items()}
+        resource_counts = {}
+        for k, v in stats.items():
+            if isinstance(k, NodeType):
+                resource_counts[k.pluralize()] = v
+            else:
+                resource_counts[k] = v
         dbt.tracking.track_resource_counts(resource_counts)
 
     # do not include resource types that are not actually defined in the project
@@ -72,8 +90,10 @@ def _node_enabled(node: ManifestNode):
         return True
 
 
-def _generate_stats(manifest: Manifest) -> Dict[NodeType, int]:
-    stats: Dict[NodeType, int] = defaultdict(int)
+def _generate_stats(
+    manifest: Manifest, catalogs: Optional[Sequence[Any]] = None
+) -> Dict[StatsKeyType, int]:
+    stats: Dict[StatsKeyType, int] = defaultdict(int)
     for node in manifest.nodes.values():
         if _node_enabled(node):
             stats[node.resource_type] += 1
@@ -87,6 +107,8 @@ def _generate_stats(manifest: Manifest) -> Dict[NodeType, int]:
     stats[NodeType.SemanticModel] += len(manifest.semantic_models)
     stats[NodeType.SavedQuery] += len(manifest.saved_queries)
     stats[NodeType.Unit] += len(manifest.unit_tests)
+    if catalogs is not None:
+        stats["catalogs"] = len(catalogs)
 
     # TODO: should we be counting dimensions + entities?
 
@@ -705,7 +727,13 @@ def _compile_relation_for_foreign_key_constraint_to(
     # This method doesn't actually "compile" any of the nodes. That is done by the
     # "compile_node" method. This creates a Linker and builds the networkx graph,
     # writes out the graph.gpickle file, and prints the stats, returning a Graph object.
-    def compile(self, manifest: Manifest, write=True, add_test_edges=False) -> Graph:
+    def compile(
+        self,
+        manifest: Manifest,
+        write=True,
+        add_test_edges=False,
+        catalogs: Optional[Sequence[Any]] = None,
+    ) -> Graph:
         self.initialize()
         linker = Linker()
         linker.link_graph(manifest)
@@ -737,14 +765,12 @@ def compile(self, manifest: Manifest, write=True, add_test_edges=False) -> Graph
                     )
                 )
 
-        stats = _generate_stats(manifest)
-
         if write:
             self.write_graph_file(linker, manifest)
 
         # Do not print these for list command
         if self.config.args.which != "list":
-            stats = _generate_stats(manifest)
+            stats = _generate_stats(manifest, catalogs)
             print_compile_stats(stats)
 
         return Graph(linker.graph)

core/dbt/task/base.py

@@ -11,6 +11,7 @@
 import dbt.exceptions
 import dbt_common.exceptions.base
 from dbt import tracking
+from dbt.artifacts.resources import Catalog
 from dbt.artifacts.resources.types import NodeType
 from dbt.artifacts.schemas.results import (
     NodeStatus,
@@ -119,12 +120,17 @@ def move_to_nearest_project_dir(project_dir: Optional[str]) -> Path:
 # holding a manifest, and moving direcories.
 class ConfiguredTask(BaseTask):
     def __init__(
-        self, args: Flags, config: RuntimeConfig, manifest: Optional[Manifest] = None
+        self,
+        args: Flags,
+        config: RuntimeConfig,
+        manifest: Optional[Manifest] = None,
+        catalogs: Optional[List[Catalog]] = None,
     ) -> None:
         super().__init__(args)
         self.config = config
         self.graph: Optional[Graph] = None
         self.manifest = manifest
+        self.catalogs = catalogs
         self.compiler = Compiler(self.config)
 
     def compile_manifest(self) -> None:
@@ -133,7 +139,7 @@ def compile_manifest(self) -> None:
 
         start_compile_manifest = time.perf_counter()
 
-        self.graph = self.compiler.compile(self.manifest)
+        self.graph = self.compiler.compile(self.manifest, catalogs=self.catalogs)
 
         compile_time = time.perf_counter() - start_compile_manifest
         if dbt.tracking.active_user is not None:

core/dbt/task/build.py

@@ -1,6 +1,7 @@
 from typing import Dict, Iterable, List, Optional, Set, Type
 
 from dbt.adapters.base import BaseRelation
+from dbt.artifacts.resources import Catalog
 from dbt.artifacts.schemas.results import NodeStatus
 from dbt.artifacts.schemas.run import RunResult
 from dbt.cli.flags import Flags
@@ -49,8 +50,14 @@ class BuildTask(RunTask):
     }
     ALL_RESOURCE_VALUES = frozenset({x for x in RUNNER_MAP.keys()})
 
-    def __init__(self, args: Flags, config: RuntimeConfig, manifest: Manifest) -> None:
-        super().__init__(args, config, manifest)
+    def __init__(
+        self,
+        args: Flags,
+        config: RuntimeConfig,
+        manifest: Manifest,
+        catalogs: Optional[List[Catalog]] = None,
+    ) -> None:
+        super().__init__(args, config, manifest, catalogs=catalogs)
         self.selected_unit_tests: Set = set()
         self.model_to_unit_test_map: Dict[str, List] = {}
 
@@ -214,4 +221,6 @@ def get_runner_type(self, node) -> Optional[Type[BaseRunner]]:
     def compile_manifest(self) -> None:
         if self.manifest is None:
             raise DbtInternalError("compile_manifest called before manifest was loaded")
-        self.graph: Graph = self.compiler.compile(self.manifest, add_test_edges=True)
+        self.graph: Graph = self.compiler.compile(
+            self.manifest, add_test_edges=True, catalogs=self.catalogs
+        )

core/dbt/task/freshness.py

@@ -197,8 +197,8 @@ def node_is_match(self, node):
 
 
 class FreshnessTask(RunTask):
-    def __init__(self, args, config, manifest) -> None:
-        super().__init__(args, config, manifest)
+    def __init__(self, args, config, manifest, catalogs) -> None:
+        super().__init__(args, config, manifest, catalogs)
 
         if self.args.output:
             deprecations.warn(

core/dbt/task/list.py

@@ -1,6 +1,7 @@
 import json
-from typing import Iterator, List
+from typing import Iterator, List, Optional
 
+from dbt.artifacts.resources import Catalog
 from dbt.cli.flags import Flags
 from dbt.config.runtime import RuntimeConfig
 from dbt.contracts.graph.manifest import Manifest
@@ -56,8 +57,14 @@ class ListTask(GraphRunnableTask):
         )
     )
 
-    def __init__(self, args: Flags, config: RuntimeConfig, manifest: Manifest) -> None:
-        super().__init__(args, config, manifest)
+    def __init__(
+        self,
+        args: Flags,
+        config: RuntimeConfig,
+        manifest: Manifest,
+        catalogs: Optional[List[Catalog]] = None,
+    ) -> None:
+        super().__init__(args, config, manifest, catalogs=catalogs)
         if self.args.models:
             if self.args.select:
                 raise DbtRuntimeError('"models" and "select" are mutually exclusive arguments')

core/dbt/task/retry.py

@@ -7,6 +7,7 @@
 from dbt.cli.flags import Flags
 from dbt.cli.types import Command as CliCommand
 from dbt.config import RuntimeConfig
+from dbt.config.catalogs import load_catalogs
 from dbt.constants import RUN_RESULTS_FILE_NAME
 from dbt.contracts.state import load_result_state
 from dbt.flags import get_flags, set_flags
@@ -118,7 +119,12 @@ def __init__(self, args: Flags, config: RuntimeConfig) -> None:
 
         # Parse manifest using resolved config/flags
         manifest = parse_manifest(retry_config, False, True, retry_flags.write_json, [])  # type: ignore
-        super().__init__(args, retry_config, manifest)
+        catalogs = load_catalogs(
+            str(retry_config.project_root),
+            retry_config.project_name,
+            retry_config.cli_vars,
+        )
+        super().__init__(args, retry_config, manifest, catalogs=catalogs)
         self.task_class = TASK_DICT.get(self.previous_command_name)  # type: ignore
 
     def run(self):
@@ -169,6 +175,7 @@ def get_graph_queue(self):
             get_flags(),
             self.config,
             self.manifest,
+            catalogs=self.catalogs,
         )
 
         if self.task_class == RunTask or self.task_class == BuildTask: