Add new get_catalog_relations macro, Supporting Changes

core/dbt/adapters/base/impl.py

@@ -73,7 +73,7 @@
 from dbt.adapters.cache import RelationsCache, _make_ref_key_dict
 from dbt import deprecations
 
-GET_CATALOG_MACRO_NAME = "get_catalog"
+GET_CATALOG_RELATIONS_MACRO_NAME = "get_catalog_relations"
 FRESHNESS_MACRO_NAME = "collect_freshness"
 
 
@@ -1080,12 +1080,24 @@ def _get_one_catalog(
         information_schema: InformationSchema,
         schemas: Set[str],
         manifest: Manifest,
+        relations_by_schema: Optional[Dict[str, Optional[List[BaseRelation]]]] = None,
     ) -> agate.Table:
-        kwargs = {"information_schema": information_schema, "schemas": schemas}
+
+        if relations_by_schema is None:
+            # The caller has not specified which relations they would like included
+            # in the results, so we specify None for each schema, which indicates
+            # to the get_catalog_relations macro that all relations for the schemas
+            # in the map should be returned.
+            relations_by_schema = {schema: None for schema in schemas}
+
+        kwargs = {
+            "information_schema": information_schema,
+            "relations_by_schema": relations_by_schema,
+        }
         table = self.execute_macro(
-            GET_CATALOG_MACRO_NAME,
+            GET_CATALOG_RELATIONS_MACRO_NAME,
             kwargs=kwargs,
-            # pass in the full manifest so we get any local project
+            # pass in the full manifest, so we get any local project
             # overrides
             manifest=manifest,
         )

core/dbt/adapters/base/relation.py

@@ -454,7 +454,7 @@ def search(self) -> Iterator[Tuple[InformationSchema, Optional[str]]]:
             for schema in schemas:
                 yield information_schema_name, schema
 
-    def flatten(self, allow_multiple_databases: bool = False):
+    def flatten(self, allow_multiple_databases: bool = False) -> "SchemaSearchMap":
         new = self.__class__()
 
         # make sure we don't have multiple databases if allow_multiple_databases is set to False

core/dbt/include/global_project/macros/adapters/metadata.sql

@@ -1,3 +1,19 @@
+{% macro get_catalog_relations(information_schema, relations_by_schema) -%}
+  {{ return(adapter.dispatch('get_catalog_relations', 'dbt')(information_schema, relations_by_schema)) }}
+{%- endmacro %}
+
+{#
+  The following default implementation simply relies on the more general
+  get_catalog macro. This is potentially less efficient than returning only the
+  relations the caller has asked for, but returning more results than were
+  requested is intentionally permitted by dbt, as long as the requested
+  relations are present.
+#}
+{% macro default__get_catalog_relations(information_schema, relations_by_schema) -%}
+    {%- set schemas = relations_by_schema | list -%}
+    {{ return(adapter.dispatch('get_catalog', 'dbt')(information_schema, schemas)) }}
+{% endmacro %}
+
 {% macro get_catalog(information_schema, schemas) -%}
   {{ return(adapter.dispatch('get_catalog', 'dbt')(information_schema, schemas)) }}
 {%- endmacro %}

tests/functional/artifacts/test_override.py

@@ -7,7 +7,7 @@
 """
 
 fail_macros__failure_sql = """
-{% macro get_catalog(information_schema, schemas) %}
+{% macro get_catalog_relations(information_schema, relations_by_schema) %}
     {% do exceptions.raise_compiler_error('rejected: no catalogs for you') %}
 {% endmacro %}