explore where/how to ingest the catalog config

core/dbt/config/runtime.py

@@ -15,7 +15,7 @@
     Type,
 )
 
-from dbt_config.external_config import ExternalCatalogConfig
+from dbt_config.catalog_config import ExternalCatalogConfig
 
 from dbt import tracking
 from dbt.adapters.contracts.connection import (
@@ -274,7 +274,8 @@ def collect_parts(
         )
         flags = get_flags()
         project = load_project(project_root, bool(flags.VERSION_CHECK), profile, cli_vars)
-        catalogs = load_external_catalog_config(project)
+        catalog_yml = load_external_catalog_config(project)
+        catalogs = ExternalCatalogConfig.model_validate(catalog_yml) if catalog_yml else None
         return project, profile, catalogs
 
     # Called in task/base.py, in BaseTask.from_args

core/dbt/context/providers.py

@@ -1295,6 +1295,7 @@ def api(self) -> Dict[str, Any]:
         return {
             "Relation": self.db_wrapper.Relation,
             "Column": self.adapter.Column,
+            # "Catalog": self.adapter.Catalog,
         }
 
     @contextproperty()

core/dbt/contracts/graph/manifest.py

@@ -20,7 +20,6 @@
     Union,
 )
 
-from dbt_config.external_config import ExternalCatalog
 from typing_extensions import Protocol
 
 import dbt_common.exceptions
@@ -845,7 +844,7 @@ class Manifest(MacroMethods, dbtClassMixin):
     unit_tests: MutableMapping[str, UnitTestDefinition] = field(default_factory=dict)
     saved_queries: MutableMapping[str, SavedQuery] = field(default_factory=dict)
     fixtures: MutableMapping[str, UnitTestFileFixture] = field(default_factory=dict)
-    catalogs: MutableMapping[str, ExternalCatalog] = field(default_factory=dict)
+    catalogs: MutableMapping[str, str] = field(default_factory=dict)
 
     _doc_lookup: Optional[DocLookup] = field(
         default=None, metadata={"serialize": lambda x: None, "deserialize": lambda x: None}

core/dbt/parser/manifest.py

@@ -10,6 +10,7 @@
 from typing import Any, Callable, Dict, List, Mapping, Optional, Set, Tuple, Type, Union
 
 import msgpack
+from dbt_config.catalog_config import ExternalCatalogConfig
 
 import dbt.deprecations
 import dbt.exceptions
@@ -29,6 +30,7 @@
 from dbt.clients.jinja import MacroStack, get_rendered
 from dbt.clients.jinja_static import statically_extract_macro_calls
 from dbt.config import Project, RuntimeConfig
+from dbt.config.external_config import load_external_catalog_config
 from dbt.constants import (
     MANIFEST_FILE_NAME,
     PARTIAL_PARSE_FILE_NAME,
@@ -443,7 +445,12 @@ def load(self) -> Manifest:
             patcher.construct_sources()
             self.manifest.sources = patcher.sources
             self._perf_info.patch_sources_elapsed = time.perf_counter() - start_patch
-
+            raw_catalog = load_external_catalog_config(self.root_project)
+            if raw_catalog:
+                catalog_config = ExternalCatalogConfig.model_validate(raw_catalog)
+                self.manifest.catalogs = {
+                    c.name: c.model_dump_json() for c in catalog_config.catalogs
+                }
             # We need to rebuild disabled in order to include disabled sources
             self.manifest.rebuild_disabled_lookup()
 
@@ -466,6 +473,7 @@ def load(self) -> Manifest:
             self.process_docs(self.root_project)
             self.process_metrics(self.root_project)
             self.process_saved_queries(self.root_project)
+            self.process_catalog(self.root_project)
             self.process_model_inferred_primary_keys()
             self.check_valid_group_config()
             self.check_valid_access_property()
@@ -1140,6 +1148,11 @@ def process_metrics(self, config: RuntimeConfig):
                 continue
             _process_metrics_for_node(self.manifest, current_project, exposure)
 
+    def process_catalog(self, config: RuntimeConfig):
+        if config.catalogs:
+            for catalog in config.catalogs.catalogs:
+                self.manifest.catalogs[catalog.name] = catalog.model_dump_json()
+
     def process_saved_queries(self, config: RuntimeConfig):
         """Processes SavedQuery nodes to populate their `depends_on`."""
         # Note: This will also capture various nodes which have been re-parsed

tests/functional/test_external_catalog.py

@@ -1,5 +1,6 @@
 import pytest
 import yaml
+from dbt_config.catalog_config import ExternalCatalog
 
 from dbt.tests.util import run_dbt, write_file
 from tests.fixtures.jaffle_shop import JaffleShopProject
@@ -8,30 +9,31 @@
 @pytest.fixture(scope="class", autouse=True)
 def dbt_catalog_config(project_root):
     config = {
-        "name": "my_project",
-        "version": "0.1",
-        "config-version": 2,
-        "external-catalog": {
-            "name": "my_external_catalog",
-            "type": "iceberg",
-            "configuration": {
-                "table_format": "parquet",
-                "namespace": "dbt",
-                "external_location": "s3://my-bucket/my-path",
-            },
-            "management": {
-                "enabled": True,
-                "create_if_not_exists": False,
-                "alter_if_different": False,
-                "read_only": True,
-                "refresh": "on_change",
-            },
-        },
+        "catalogs": [
+            {
+                "name": "my_external_catalog",
+                "type": "iceberg",
+                "configuration": {
+                    "table_format": "parquet",
+                    "namespace": "dbt",
+                    "external_location": "s3://my-bucket/my-path",
+                },
+                "management": {
+                    "enabled": True,
+                    "create_if_not_exists": False,
+                    "alter_if_different": False,
+                    "read_only": True,
+                    "refresh": "on_change",
+                },
+            }
+        ],
     }
     write_file(yaml.safe_dump(config), project_root, "catalog.yml")
 
 
 class TestCatalogConfig(JaffleShopProject):
 
     def test_supplying_external_catalog(self, project):
-        run_dbt(["build"])
+        manifest = run_dbt(["parse"])
+        assert manifest.catalogs != {}
+        ExternalCatalog.model_validate_json(manifest.catalogs["my_external_catalog"])

tests/unit/utils/__init__.py

@@ -11,7 +11,7 @@
 
 import agate
 import pytest
-from dbt_config.external_config import ExternalCatalogConfig
+from dbt_config.catalog_config import ExternalCatalogConfig
 
 from dbt.config.project import PartialProject
 from dbt.config.renderer import CatalogRenderer