Repository: dbt-labs/dbt-core
Pull request: Feature: faster snowflake catalogs (#2009)
(Tentatively) fix #2009 

- Thread catalog queries across information schemas
- the `get_catalog` macro now takes an information schema and a list of schemas in that information schema
- filtering happens in the database instead of in python

To be honest, I'm not totally sure about the bigquery impact of these changes :( The bigquery code does some funky stuff to handle the fact that the information schema is sometimes per-schema on that database, and I could easily have screwed it up, in which case we'd issue too many queries.

Problem statement:
Reduce memory consumption in docs generation
### Describe the feature
The `dbt docs generate` command fetches all of the columns in all of the source and target schemas touched by dbt. As such, even moderately sized dbt projects can have catalogs containing hundreds of thousands of columns. In practice, we see that the `dbt docs generate` command can consume hundreds of megabytes of memory which causes challenges for tools which orchestrate dbt.

Let's do some profiling here to gauge the memory usage patterns of the `dbt docs generate` command and inspect parts of the codebase that might be responsible for ballooning memory usage. If there are any straightforward changes we can make to the dbt codebase to reduce the memory footprint of this command, we should try to implement them.

### Describe alternatives you've considered
Download more RAM

### Additional context
This doesn't appear to be database specific. The absolute numbers aren't so useful here, but I have seen a report of a catalog query which returns 140k records consuming over 600mb of memory. Some of the Agate operations that happen on the dataframe returned by the database might be likely culprits here.

Hints from issue discussion:

Commit message: PR feedback w/ improved catalog results behavior

Patch to review:
diff --git a/core/dbt/adapters/base/impl.py b/core/dbt/adapters/base/impl.py
index 297713e2c7b..5129d60d646 100644
--- a/core/dbt/adapters/base/impl.py
+++ b/core/dbt/adapters/base/impl.py
@@ -1,4 +1,6 @@
 import abc
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import Future  # noqa - we use this for typing only
 from contextlib import contextmanager
 from datetime import datetime
 from typing import (
@@ -21,6 +23,7 @@
 from dbt.contracts.graph.compiled import CompileResultNode, CompiledSeedNode
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.parsed import ParsedSeedNode
+from dbt.exceptions import warn_or_error
 from dbt.node_types import NodeType
 from dbt.logger import GLOBAL_LOGGER as logger
 from dbt.utils import filter_null_values
@@ -117,7 +120,7 @@ def _relation_name(rel: Optional[BaseRelation]) -> str:
         return str(rel)
 
 
-class SchemaSearchMap(dict):
+class SchemaSearchMap(Dict[InformationSchema, Set[Optional[str]]]):
     """A utility class to keep track of what information_schema tables to
     search for what schemas
     """
@@ -1009,29 +1012,56 @@ def _catalog_filter_table(
         """
         return table.where(_catalog_filter_schemas(manifest))
 
-    def _get_catalog_information_schemas(
-        self, manifest: Manifest
-    ) -> List[InformationSchema]:
-        return list(self._get_cache_schemas(manifest).keys())
+    def _get_one_catalog(
+        self,
+        information_schema: InformationSchema,
+        schemas: Set[str],
+        manifest: Manifest,
+    ) -> agate.Table:
 
-    def get_catalog(self, manifest: Manifest) -> agate.Table:
-        """Get the catalog for this manifest by running the get catalog macro.
-        Returns an agate.Table of catalog information.
-        """
-        information_schemas = self._get_catalog_information_schemas(manifest)
-        # make it a list so macros can index into it.
-        kwargs = {'information_schemas': information_schemas}
-        table = self.execute_macro(
-            GET_CATALOG_MACRO_NAME,
-            kwargs=kwargs,
-            release=True,
-            # pass in the full manifest so we get any local project overrides
-            manifest=manifest,
-        )
+        name = '.'.join([
+            str(information_schema.database),
+            'information_schema'
+        ])
+
+        # calculate the possible schemas for a given schema name
+        all_schema_names: Set[str] = set()
+        for schema in schemas:
+            all_schema_names.update({schema, schema.lower(), schema.upper()})
+
+        with self.connection_named(name):
+            kwargs = {
+                'information_schema': information_schema,
+                'schemas': all_schema_names
+            }
+            table = self.execute_macro(
+                GET_CATALOG_MACRO_NAME,
+                kwargs=kwargs,
+                release=True,
+                # pass in the full manifest so we get any local project
+                # overrides
+                manifest=manifest,
+            )
 
         results = self._catalog_filter_table(table, manifest)
         return results
 
+    def get_catalog(
+        self, manifest: Manifest
+    ) -> Tuple[agate.Table, List[Exception]]:
+        # snowflake is super slow. split it out into the specified threads
+        num_threads = self.config.threads
+        schema_map = self._get_cache_schemas(manifest)
+
+        with ThreadPoolExecutor(max_workers=num_threads) as executor:
+            futures = [
+                executor.submit(self._get_one_catalog, info, schemas, manifest)
+                for info, schemas in schema_map.items() if len(schemas) > 0
+            ]
+            catalogs, exceptions = catch_as_completed(futures)
+
+        return catalogs, exceptions
+
     def cancel_open_connections(self):
         """Cancel all open connections."""
         return self.connections.cancel_open()
@@ -1104,3 +1134,30 @@ def post_model_hook(self, config: Mapping[str, Any], context: Any) -> None:
         The second parameter is the value returned by pre_mdoel_hook.
         """
         pass
+
+
+def catch_as_completed(
+    futures  # typing: List[Future[agate.Table]]
+) -> Tuple[agate.Table, List[Exception]]:
+
+    catalogs: agate.Table = agate.Table(rows=[])
+    exceptions: List[Exception] = []
+
+    for future in as_completed(futures):
+        exc = future.exception()
+        # we want to re-raise on ctrl+c and BaseException
+        if exc is None:
+            catalog = future.result()
+            catalogs = agate.Table.merge([catalogs, catalog])
+        elif (
+            isinstance(exc, KeyboardInterrupt) or
+            not isinstance(exc, Exception)
+        ):
+            raise exc
+        else:
+            warn_or_error(
+                f'Encountered an error while generating catalog: {str(exc)}'
+            )
+            # exc is not None, derives from Exception, and isn't ctrl+c
+            exceptions.append(exc)
+    return catalogs, exceptions
diff --git a/core/dbt/contracts/results.py b/core/dbt/contracts/results.py
index 69226debd8a..ea42ad81ede 100644
--- a/core/dbt/contracts/results.py
+++ b/core/dbt/contracts/results.py
@@ -291,4 +291,5 @@ def key(self) -> CatalogKey:
 class CatalogResults(JsonSchemaMixin, Writable):
     nodes: Dict[str, CatalogTable]
     generated_at: datetime
+    errors: Optional[List[str]]
     _compile_results: Optional[Any] = None
diff --git a/core/dbt/contracts/rpc.py b/core/dbt/contracts/rpc.py
index 23b152f5d34..06e41e84c64 100644
--- a/core/dbt/contracts/rpc.py
+++ b/core/dbt/contracts/rpc.py
@@ -510,6 +510,7 @@ def from_result(
         return cls(
             nodes=base.nodes,
             generated_at=base.generated_at,
+            errors=base.errors,
             _compile_results=base._compile_results,
             logs=logs,
             tags=tags,
diff --git a/core/dbt/include/global_project/macros/adapters/common.sql b/core/dbt/include/global_project/macros/adapters/common.sql
index bb7a4620d30..1d2ec0ca250 100644
--- a/core/dbt/include/global_project/macros/adapters/common.sql
+++ b/core/dbt/include/global_project/macros/adapters/common.sql
@@ -94,11 +94,11 @@
 {% endmacro %}
 
 
-{% macro get_catalog(information_schemas) -%}
-  {{ return(adapter_macro('get_catalog', information_schemas)) }}
+{% macro get_catalog(information_schema, schemas) -%}
+  {{ return(adapter_macro('get_catalog', information_schema, schemas)) }}
 {%- endmacro %}
 
-{% macro default__get_catalog(information_schemas) -%}
+{% macro default__get_catalog(information_schema, schemas) -%}
 
   {% set typename = adapter.type() %}
   {% set msg -%}
diff --git a/core/dbt/task/generate.py b/core/dbt/task/generate.py
index 97c03a1589a..00e1a052cfe 100644
--- a/core/dbt/task/generate.py
+++ b/core/dbt/task/generate.py
@@ -14,6 +14,7 @@
 )
 from dbt.exceptions import InternalException
 from dbt.include.global_project import DOCS_INDEX_FILE_PATH
+from dbt.logger import GLOBAL_LOGGER as logger
 import dbt.ui.printer
 import dbt.utils
 import dbt.compilation
@@ -194,7 +195,9 @@ def run(self):
                 dbt.ui.printer.print_timestamped_line(
                     'compile failed, cannot generate docs'
                 )
-                return CatalogResults({}, datetime.utcnow(), compile_results)
+                return CatalogResults(
+                    {}, datetime.utcnow(), compile_results, None
+                )
 
         shutil.copyfile(
             DOCS_INDEX_FILE_PATH,
@@ -208,7 +211,7 @@ def run(self):
         adapter = get_adapter(self.config)
         with adapter.connection_named('generate_catalog'):
             dbt.ui.printer.print_timestamped_line("Building catalog")
-            catalog_table = adapter.get_catalog(self.manifest)
+            catalog_table, exceptions = adapter.get_catalog(self.manifest)
 
         catalog_data: List[PrimitiveDict] = [
             dict(zip(catalog_table.column_names, map(_coerce_decimal, row)))
@@ -216,10 +219,16 @@ def run(self):
         ]
 
         catalog = Catalog(catalog_data)
+
+        errors: Optional[List[str]] = None
+        if exceptions:
+            errors = [str(e) for e in exceptions]
+
         results = self.get_catalog_results(
             nodes=catalog.make_unique_id_map(self.manifest),
             generated_at=datetime.utcnow(),
             compile_results=compile_results,
+            errors=errors,
         )
 
         path = os.path.join(self.config.target_path, CATALOG_FILENAME)
@@ -229,21 +238,32 @@ def run(self):
         dbt.ui.printer.print_timestamped_line(
             'Catalog written to {}'.format(os.path.abspath(path))
         )
+
+        if exceptions:
+            logger.error(
+                'dbt encountered {} failure{} while writing the catalog'
+                .format(len(exceptions), (len(exceptions) == 1) * 's')
+            )
+
         return results
 
     def get_catalog_results(
         self,
         nodes: Dict[str, CatalogTable],
         generated_at: datetime,
-        compile_results: Optional[Any]
+        compile_results: Optional[Any],
+        errors: Optional[List[str]]
     ) -> CatalogResults:
         return CatalogResults(
             nodes=nodes,
             generated_at=generated_at,
             _compile_results=compile_results,
+            errors=errors,
         )
 
     def interpret_results(self, results):
+        if results.errors:
+            return False
         compile_results = results._compile_results
         if compile_results is None:
             return True
diff --git a/core/dbt/task/rpc/project_commands.py b/core/dbt/task/rpc/project_commands.py
index 0449f306847..d88f46b5371 100644
--- a/core/dbt/task/rpc/project_commands.py
+++ b/core/dbt/task/rpc/project_commands.py
@@ -110,12 +110,13 @@ def set_args(self, params: RPCDocsGenerateParameters) -> None:
         self.args.compile = params.compile
 
     def get_catalog_results(
-        self, nodes, generated_at, compile_results
+        self, nodes, generated_at, compile_results, errors
     ) -> RemoteCatalogResults:
         return RemoteCatalogResults(
             nodes=nodes,
             generated_at=datetime.utcnow(),
             _compile_results=compile_results,
+            errors=errors,
             logs=[],
         )
 
diff --git a/plugins/bigquery/dbt/adapters/bigquery/impl.py b/plugins/bigquery/dbt/adapters/bigquery/impl.py
index de02f5f2dab..1e58c73a5ad 100644
--- a/plugins/bigquery/dbt/adapters/bigquery/impl.py
+++ b/plugins/bigquery/dbt/adapters/bigquery/impl.py
@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, Set
 
 import dbt.deprecations
 import dbt.exceptions
@@ -7,6 +7,7 @@
 import dbt.clients.agate_helper
 
 from dbt.adapters.base import BaseAdapter, available, RelationType
+from dbt.adapters.base.impl import SchemaSearchMap
 from dbt.adapters.bigquery.relation import (
     BigQueryRelation, BigQueryInformationSchema
 )
@@ -492,23 +493,22 @@ def _catalog_filter_table(
         })
         return super()._catalog_filter_table(table, manifest)
 
-    def _get_catalog_information_schemas(
-        self, manifest: Manifest
-    ) -> List[BigQueryInformationSchema]:
+    def _get_cache_schemas(
+        self, manifest: Manifest, exec_only: bool = False
+    ) -> SchemaSearchMap:
+        candidates = super()._get_cache_schemas(manifest, exec_only)
+        db_schemas: Dict[str, Set[str]] = {}
+        result = SchemaSearchMap()
 
-        candidates = super()._get_catalog_information_schemas(manifest)
-        information_schemas = []
-        db_schemas = {}
-        for candidate in candidates:
+        for candidate, schemas in candidates.items():
             database = candidate.database
             if database not in db_schemas:
                 db_schemas[database] = set(self.list_schemas(database))
             if candidate.schema in db_schemas[database]:
-                information_schemas.append(candidate)
+                result[candidate] = schemas
             else:
                 logger.debug(
                     'Skipping catalog for {}.{} - schema does not exist'
                     .format(database, candidate.schema)
                 )
-
-        return information_schemas
+        return result
diff --git a/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql b/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql
index bc013b2bd37..d41b03604f4 100644
--- a/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql
+++ b/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql
@@ -1,225 +1,223 @@
 
-{% macro bigquery__get_catalog(information_schemas) -%}
+{% macro bigquery__get_catalog(information_schema, schemas) -%}
 
-  {%- if (information_schemas | length) == 0 -%}
+  {%- if (schemas | length) == 0 -%}
     {# Hopefully nothing cares about the columns we return when there are no rows #}
     {%- set query  = "select 1 as id limit 0" -%}
   {%- else -%}
 
   {%- set query -%}
-    {%- for information_schema in information_schemas -%}
-      (
-        with schemas as (
+    with schemas as (
 
-            select
-              catalog_name as table_database,
-              schema_name as table_schema,
-              location
-
-            from {{ information_schema.replace(information_schema_view='SCHEMATA') }}
-
-        ),
-
-        tables as (
-            select
-                project_id as table_database,
-                dataset_id as table_schema,
-                table_id as original_table_name,
-
-                concat(project_id, '.', dataset_id, '.', table_id) as relation_id,
-
-                row_count,
-                size_bytes as size_bytes,
-                case
-                    when type = 1 then 'table'
-                    when type = 2 then 'view'
-                    else 'external'
-                end as table_type,
+        select
+          catalog_name as table_database,
+          schema_name as table_schema,
+          location
+
+        from {{ information_schema.replace(information_schema_view='SCHEMATA') }}
+        where (
+          {%- for schema in schemas -%}
+            upper(schema_name) = upper('{{ schema }}'){%- if not loop.last %} or {% endif -%}
+          {%- endfor -%}
+        )
+    ),
 
-                REGEXP_CONTAINS(table_id, '^.+[0-9]{8}$') and coalesce(type, 0) = 1 as is_date_shard,
-                REGEXP_EXTRACT(table_id, '^(.+)[0-9]{8}$') as shard_base_name,
-                REGEXP_EXTRACT(table_id, '^.+([0-9]{8})$') as shard_name
+    tables as (
+        select
+            project_id as table_database,
+            dataset_id as table_schema,
+            table_id as original_table_name,
 
-            from {{ information_schema.replace(information_schema_view='__TABLES__') }}
+            concat(project_id, '.', dataset_id, '.', table_id) as relation_id,
 
-        ),
+            row_count,
+            size_bytes as size_bytes,
+            case
+                when type = 1 then 'table'
+                when type = 2 then 'view'
+                else 'external'
+            end as table_type,
 
-        extracted as (
+            REGEXP_CONTAINS(table_id, '^.+[0-9]{8}$') and coalesce(type, 0) = 1 as is_date_shard,
+            REGEXP_EXTRACT(table_id, '^(.+)[0-9]{8}$') as shard_base_name,
+            REGEXP_EXTRACT(table_id, '^.+([0-9]{8})$') as shard_name
 
-            select *,
-                case
-                    when is_date_shard then shard_base_name
-                    else original_table_name
-                end as table_name
+        from {{ information_schema.replace(information_schema_view='__TABLES__') }}
 
-            from tables
+    ),
 
-        ),
+    extracted as (
 
-        unsharded_tables as (
+        select *,
+            case
+                when is_date_shard then shard_base_name
+                else original_table_name
+            end as table_name
 
-            select
-                table_database,
-                table_schema,
-                table_name,
-                coalesce(table_type, 'external') as table_type,
-                is_date_shard,
+        from tables
 
-                struct(
-                    min(shard_name) as shard_min,
-                    max(shard_name) as shard_max,
-                    count(*) as shard_count
-                ) as table_shards,
+    ),
 
-                sum(size_bytes) as size_bytes,
-                sum(row_count) as row_count,
+    unsharded_tables as (
 
-                max(relation_id) as relation_id
+        select
+            table_database,
+            table_schema,
+            table_name,
+            coalesce(table_type, 'external') as table_type,
+            is_date_shard,
 
-            from extracted
-            group by 1,2,3,4,5
+            struct(
+                min(shard_name) as shard_min,
+                max(shard_name) as shard_max,
+                count(*) as shard_count
+            ) as table_shards,
 
-        ),
+            sum(size_bytes) as size_bytes,
+            sum(row_count) as row_count,
 
-        info_schema_columns as (
+            max(relation_id) as relation_id
 
-            select
-                concat(table_catalog, '.', table_schema, '.', table_name) as relation_id,
-                table_catalog as table_database,
-                table_schema,
-                table_name,
+        from extracted
+        group by 1,2,3,4,5
 
-                -- use the "real" column name from the paths query below
-                column_name as base_column_name,
-                ordinal_position as column_index,
-                cast(null as string) as column_comment,
+    ),
 
-                is_partitioning_column,
-                clustering_ordinal_position
+    info_schema_columns as (
 
-            from {{ information_schema.replace(information_schema_view='COLUMNS') }}
-            where ordinal_position is not null
+        select
+            concat(table_catalog, '.', table_schema, '.', table_name) as relation_id,
+            table_catalog as table_database,
+            table_schema,
+            table_name,
 
-        ),
+            -- use the "real" column name from the paths query below
+            column_name as base_column_name,
+            ordinal_position as column_index,
+            cast(null as string) as column_comment,
 
-        info_schema_column_paths as (
+            is_partitioning_column,
+            clustering_ordinal_position
 
-            select
-                concat(table_catalog, '.', table_schema, '.', table_name) as relation_id,
-                field_path as column_name,
-                data_type as column_type,
-                column_name as base_column_name
+        from {{ information_schema.replace(information_schema_view='COLUMNS') }}
+        where ordinal_position is not null
 
-            from {{ information_schema.replace(information_schema_view='COLUMN_FIELD_PATHS') }}
-            where data_type not like 'STRUCT%'
+    ),
 
-        ),
+    info_schema_column_paths as (
 
-        columns as (
+        select
+            concat(table_catalog, '.', table_schema, '.', table_name) as relation_id,
+            field_path as column_name,
+            data_type as column_type,
+            column_name as base_column_name
 
-            select * except (base_column_name)
-            from info_schema_columns
-            join info_schema_column_paths using (relation_id, base_column_name)
+        from {{ information_schema.replace(information_schema_view='COLUMN_FIELD_PATHS') }}
+        where data_type not like 'STRUCT%'
 
-        ),
+    ),
 
-        column_stats as (
+    columns as (
 
-            select
-                table_database,
-                table_schema,
-                table_name,
-                max(relation_id) as relation_id,
-                max(case when is_partitioning_column = 'YES' then 1 else 0 end) = 1 as is_partitioned,
-                max(case when is_partitioning_column = 'YES' then column_name else null end) as partition_column,
-                max(case when clustering_ordinal_position is not null then 1 else 0 end) = 1 as is_clustered,
-                array_to_string(
-                    array_agg(
-                        case
-                            when clustering_ordinal_position is not null then column_name
-                            else null
-                        end ignore nulls
-                        order by clustering_ordinal_position
-                    ), ', '
-                ) as clustering_columns
+        select * except (base_column_name)
+        from info_schema_columns
+        join info_schema_column_paths using (relation_id, base_column_name)
 
-            from columns
-            group by 1,2,3
+    ),
 
-        )
+    column_stats as (
 
         select
-            unsharded_tables.table_database,
-            unsharded_tables.table_schema,
-            case
-                when is_date_shard then concat(unsharded_tables.table_name, '*')
-                else unsharded_tables.table_name
-            end as table_name,
-            unsharded_tables.table_type,
-
-            -- coalesce name and type for External tables - these columns are not
-            -- present in the COLUMN_FIELD_PATHS resultset
-            coalesce(columns.column_name, '<unknown>') as column_name,
-            -- invent a row number to account for nested fields -- BQ does
-            -- not treat these nested properties as independent fields
-            row_number() over (
-                partition by relation_id
-                order by columns.column_index, columns.column_name
-            ) as column_index,
-            coalesce(columns.column_type, '<unknown>') as column_type,
-            columns.column_comment,
-
-            'Location' as `stats__location__label`,
-            location as `stats__location__value`,
-            'The geographic location of this table' as `stats__location__description`,
-            location is not null as `stats__location__include`,
-
-            'Shard count' as `stats__date_shards__label`,
-            table_shards.shard_count as `stats__date_shards__value`,
-            'The number of date shards in this table' as `stats__date_shards__description`,
-            is_date_shard as `stats__date_shards__include`,
-
-            'Shard (min)' as `stats__date_shard_min__label`,
-            table_shards.shard_min as `stats__date_shard_min__value`,
-            'The first date shard in this table' as `stats__date_shard_min__description`,
-            is_date_shard as `stats__date_shard_min__include`,
-
-            'Shard (max)' as `stats__date_shard_max__label`,
-            table_shards.shard_max as `stats__date_shard_max__value`,
-            'The last date shard in this table' as `stats__date_shard_max__description`,
-            is_date_shard as `stats__date_shard_max__include`,
-
-            '# Rows' as `stats__num_rows__label`,
-            row_count as `stats__num_rows__value`,
-            'Approximate count of rows in this table' as `stats__num_rows__description`,
-            (unsharded_tables.table_type = 'table') as `stats__num_rows__include`,
-
-            'Approximate Size' as `stats__num_bytes__label`,
-            size_bytes as `stats__num_bytes__value`,
-            'Approximate size of table as reported by BigQuery' as `stats__num_bytes__description`,
-            (unsharded_tables.table_type = 'table') as `stats__num_bytes__include`,
-
-            'Partitioned By' as `stats__partitioning_type__label`,
-            partition_column as `stats__partitioning_type__value`,
-            'The partitioning column for this table' as `stats__partitioning_type__description`,
-            is_partitioned as `stats__partitioning_type__include`,
-
-            'Clustered By' as `stats__clustering_fields__label`,
-            clustering_columns as `stats__clustering_fields__value`,
-            'The clustering columns for this table' as `stats__clustering_fields__description`,
-            is_clustered as `stats__clustering_fields__include`
-
-        -- join using relation_id (an actual relation, not a shard prefix) to make
-        -- sure that column metadata is picked up through the join. This will only
-        -- return the column information for the "max" table in a date-sharded table set
-        from unsharded_tables
-        left join schemas using(table_database, table_schema)
-        left join columns using (relation_id)
-        left join column_stats using (relation_id)
-      )
-
-      {% if not loop.last %} union all {% endif %}
-    {%- endfor -%}
+            table_database,
+            table_schema,
+            table_name,
+            max(relation_id) as relation_id,
+            max(case when is_partitioning_column = 'YES' then 1 else 0 end) = 1 as is_partitioned,
+            max(case when is_partitioning_column = 'YES' then column_name else null end) as partition_column,
+            max(case when clustering_ordinal_position is not null then 1 else 0 end) = 1 as is_clustered,
+            array_to_string(
+                array_agg(
+                    case
+                        when clustering_ordinal_position is not null then column_name
+                        else null
+                    end ignore nulls
+                    order by clustering_ordinal_position
+                ), ', '
+            ) as clustering_columns
+
+        from columns
+        group by 1,2,3
+
+    )
+
+    select
+        unsharded_tables.table_database,
+        unsharded_tables.table_schema,
+        case
+            when is_date_shard then concat(unsharded_tables.table_name, '*')
+            else unsharded_tables.table_name
+        end as table_name,
+        unsharded_tables.table_type,
+
+        -- coalesce name and type for External tables - these columns are not
+        -- present in the COLUMN_FIELD_PATHS resultset
+        coalesce(columns.column_name, '<unknown>') as column_name,
+        -- invent a row number to account for nested fields -- BQ does
+        -- not treat these nested properties as independent fields
+        row_number() over (
+            partition by relation_id
+            order by columns.column_index, columns.column_name
+        ) as column_index,
+        coalesce(columns.column_type, '<unknown>') as column_type,
+        columns.column_comment,
+
+        'Location' as `stats__location__label`,
+        location as `stats__location__value`,
+        'The geographic location of this table' as `stats__location__description`,
+        location is not null as `stats__location__include`,
+
+        'Shard count' as `stats__date_shards__label`,
+        table_shards.shard_count as `stats__date_shards__value`,
+        'The number of date shards in this table' as `stats__date_shards__description`,
+        is_date_shard as `stats__date_shards__include`,
+
+        'Shard (min)' as `stats__date_shard_min__label`,
+        table_shards.shard_min as `stats__date_shard_min__value`,
+        'The first date shard in this table' as `stats__date_shard_min__description`,
+        is_date_shard as `stats__date_shard_min__include`,
+
+        'Shard (max)' as `stats__date_shard_max__label`,
+        table_shards.shard_max as `stats__date_shard_max__value`,
+        'The last date shard in this table' as `stats__date_shard_max__description`,
+        is_date_shard as `stats__date_shard_max__include`,
+
+        '# Rows' as `stats__num_rows__label`,
+        row_count as `stats__num_rows__value`,
+        'Approximate count of rows in this table' as `stats__num_rows__description`,
+        (unsharded_tables.table_type = 'table') as `stats__num_rows__include`,
+
+        'Approximate Size' as `stats__num_bytes__label`,
+        size_bytes as `stats__num_bytes__value`,
+        'Approximate size of table as reported by BigQuery' as `stats__num_bytes__description`,
+        (unsharded_tables.table_type = 'table') as `stats__num_bytes__include`,
+
+        'Partitioned By' as `stats__partitioning_type__label`,
+        partition_column as `stats__partitioning_type__value`,
+        'The partitioning column for this table' as `stats__partitioning_type__description`,
+        is_partitioned as `stats__partitioning_type__include`,
+
+        'Clustered By' as `stats__clustering_fields__label`,
+        clustering_columns as `stats__clustering_fields__value`,
+        'The clustering columns for this table' as `stats__clustering_fields__description`,
+        is_clustered as `stats__clustering_fields__include`
+
+    -- join using relation_id (an actual relation, not a shard prefix) to make
+    -- sure that column metadata is picked up through the join. This will only
+    -- return the column information for the "max" table in a date-sharded table set
+    from unsharded_tables
+    left join schemas using(table_database, table_schema)
+    left join columns using (relation_id)
+    left join column_stats using (relation_id)
   {%- endset -%}
 
   {%- endif -%}
diff --git a/plugins/postgres/dbt/include/postgres/macros/catalog.sql b/plugins/postgres/dbt/include/postgres/macros/catalog.sql
index b7a4243a66f..377fdbd9ae9 100644
--- a/plugins/postgres/dbt/include/postgres/macros/catalog.sql
+++ b/plugins/postgres/dbt/include/postgres/macros/catalog.sql
@@ -1,12 +1,12 @@
 
-{% macro postgres__get_catalog(information_schemas) -%}
+{% macro postgres__get_catalog(information_schema, schemas) -%}
 
   {%- call statement('catalog', fetch_result=True) -%}
     {#
       If the user has multiple databases set and the first one is wrong, this will fail.
       But we won't fail in the case where there are multiple quoting-difference-only dbs, which is better.
     #}
-    {% set database = information_schemas[0].database %}
+    {% set database = information_schema.database %}
     {{ adapter.verify_database(database) }}
 
     select
@@ -28,8 +28,11 @@
     join pg_catalog.pg_class tbl on tbl.relnamespace = sch.oid
     join pg_catalog.pg_attribute col on col.attrelid = tbl.oid
 
-    where sch.nspname != 'information_schema'
-      and sch.nspname not like 'pg\_%' -- avoid postgres system schemas, '_' is a wildcard so escape it
+    where (
+        {%- for schema in schemas -%}
+          upper(sch.nspname) = upper('{{ schema }}'){%- if not loop.last %} or {% endif -%}
+        {%- endfor -%}
+      )
       and not pg_is_other_temp_schema(sch.oid) -- not a temporary schema belonging to another session
       and tbl.relpersistence = 'p' -- [p]ermanent table. Other values are [u]nlogged table, [t]emporary table
       and tbl.relkind in ('r', 'v', 'f', 'p') -- o[r]dinary table, [v]iew, [f]oreign table, [p]artitioned table. Other values are [i]ndex, [S]equence, [c]omposite type, [t]OAST table, [m]aterialized view
diff --git a/plugins/redshift/dbt/include/redshift/macros/catalog.sql b/plugins/redshift/dbt/include/redshift/macros/catalog.sql
index f8b40477c5a..83935f67a35 100644
--- a/plugins/redshift/dbt/include/redshift/macros/catalog.sql
+++ b/plugins/redshift/dbt/include/redshift/macros/catalog.sql
@@ -1,7 +1,7 @@
 
-{% macro redshift__get_base_catalog(information_schemas) -%}
+{% macro redshift__get_base_catalog(information_schema, schemas) -%}
   {%- call statement('base_catalog', fetch_result=True) -%}
-    {% set database = information_schemas[0].database %}
+    {% set database = information_schema.database %}
     {{ adapter.verify_database(database) }}
 
     with late_binding as (
@@ -94,8 +94,11 @@
     from unioned
     join table_owners using (table_database, table_schema, table_name)
 
-    where table_schema != 'information_schema'
-      and table_schema not like 'pg\_%'
+    where (
+        {%- for schema in schemas -%}
+          upper(table_schema) = upper('{{ schema }}'){%- if not loop.last %} or {% endif -%}
+        {%- endfor -%}
+      )
 
     order by "column_index"
   {%- endcall -%}
@@ -103,7 +106,7 @@
   {{ return(load_result('base_catalog').table) }}
 {%- endmacro %}
 
-{% macro redshift__get_extended_catalog(information_schemas) %}
+{% macro redshift__get_extended_catalog(schemas) %}
   {%- call statement('extended_catalog', fetch_result=True) -%}
 
     select
@@ -180,6 +183,11 @@
         (skew_rows is not null) as "stats:skew_rows:include"
 
     from svv_table_info
+    where (
+        {%- for schema in schemas -%}
+          upper(schema) = upper('{{ schema }}'){%- if not loop.last %} or {% endif -%}
+        {%- endfor -%}
+    )
 
   {%- endcall -%}
 
@@ -215,16 +223,16 @@
 {% endmacro %}
 
 
-{% macro redshift__get_catalog(information_schemas) %}
+{% macro redshift__get_catalog(information_schema, schemas) %}
 
     {#-- Compute a left-outer join in memory. Some Redshift queries are
       -- leader-only, and cannot be joined to other compute-based queries #}
 
-    {% set catalog = redshift__get_base_catalog(information_schemas) %}
+    {% set catalog = redshift__get_base_catalog(information_schema, schemas) %}
 
     {% set select_extended =  redshift__can_select_from('svv_table_info') %}
     {% if select_extended %}
-        {% set extended_catalog = redshift__get_extended_catalog() %}
+        {% set extended_catalog = redshift__get_extended_catalog(schemas) %}
         {% set catalog = catalog.join(extended_catalog, 'table_id') %}
     {% else %}
         {{ redshift__no_svv_table_info_warning() }}
diff --git a/plugins/snowflake/dbt/adapters/snowflake/impl.py b/plugins/snowflake/dbt/adapters/snowflake/impl.py
index 13af6dc73f2..86c96abbb55 100644
--- a/plugins/snowflake/dbt/adapters/snowflake/impl.py
+++ b/plugins/snowflake/dbt/adapters/snowflake/impl.py
@@ -1,11 +1,14 @@
 from typing import Mapping, Any, Optional
 
+import agate
+
 from dbt.adapters.sql import SQLAdapter
 from dbt.adapters.snowflake import SnowflakeConnectionManager
 from dbt.adapters.snowflake import SnowflakeRelation
 from dbt.adapters.snowflake import SnowflakeColumn
-from dbt.utils import filter_null_values
+from dbt.contracts.graph.manifest import Manifest
 from dbt.exceptions import RuntimeException
+from dbt.utils import filter_null_values
 
 
 class SnowflakeAdapter(SQLAdapter):
@@ -23,7 +26,9 @@ def date_function(cls):
         return "CURRENT_TIMESTAMP()"
 
     @classmethod
-    def _catalog_filter_table(cls, table, manifest):
+    def _catalog_filter_table(
+        cls, table: agate.Table, manifest: Manifest
+    ) -> agate.Table:
         # On snowflake, users can set QUOTED_IDENTIFIERS_IGNORE_CASE, so force
         # the column names to their lowercased forms.
         lowered = table.rename(
diff --git a/plugins/snowflake/dbt/include/snowflake/macros/catalog.sql b/plugins/snowflake/dbt/include/snowflake/macros/catalog.sql
index fe68cd2e46e..1d796a3251d 100644
--- a/plugins/snowflake/dbt/include/snowflake/macros/catalog.sql
+++ b/plugins/snowflake/dbt/include/snowflake/macros/catalog.sql
@@ -1,68 +1,62 @@
-
-{% macro snowflake__get_catalog(information_schemas) -%}
-
-    {%- call statement('catalog', fetch_result=True) -%}
-    {% for information_schema in information_schemas %}
-
-        (
-            with tables as (
-
-                select
-                    table_catalog as "table_database",
-                    table_schema as "table_schema",
-                    table_name as "table_name",
-                    table_type as "table_type",
-
-                    -- note: this is the _role_ that owns the table
-                    table_owner as "table_owner",
-
-                    'Clustering Key' as "stats:clustering_key:label",
-                    clustering_key as "stats:clustering_key:value",
-                    'The key used to cluster this table' as "stats:clustering_key:description",
-                    (clustering_key is not null) as "stats:clustering_key:include",
-
-                    'Row Count' as "stats:row_count:label",
-                    row_count as "stats:row_count:value",
-                    'An approximate count of rows in this table' as "stats:row_count:description",
-                    (row_count is not null) as "stats:row_count:include",
-
-                    'Approximate Size' as "stats:bytes:label",
-                    bytes as "stats:bytes:value",
-                    'Approximate size of the table as reported by Snowflake' as "stats:bytes:description",
-                    (bytes is not null) as "stats:bytes:include"
-
-                from {{ information_schema }}.tables
-
-            ),
-
-            columns as (
-
-                select
-                    table_catalog as "table_database",
-                    table_schema as "table_schema",
-                    table_name as "table_name",
-                    null as "table_comment",
-
-                    column_name as "column_name",
-                    ordinal_position as "column_index",
-                    data_type as "column_type",
-                    null as "column_comment"
-
-                from {{ information_schema }}.columns
-
-            )
-
-            select *
-            from tables
-            join columns using ("table_database", "table_schema", "table_name")
-            where "table_schema" != 'INFORMATION_SCHEMA'
-            order by "column_index"
-        )
-        {% if not loop.last %} union all {% endif %}
-
-    {% endfor %}
-  {%- endcall -%}
-
-  {{ return(load_result('catalog').table) }}
+{% macro snowflake__get_catalog(information_schema, schemas) -%}
+  {% set query %}
+      with tables as (
+
+          select
+              table_catalog as "table_database",
+              table_schema as "table_schema",
+              table_name as "table_name",
+              table_type as "table_type",
+
+              -- note: this is the _role_ that owns the table
+              table_owner as "table_owner",
+
+              'Clustering Key' as "stats:clustering_key:label",
+              clustering_key as "stats:clustering_key:value",
+              'The key used to cluster this table' as "stats:clustering_key:description",
+              (clustering_key is not null) as "stats:clustering_key:include",
+
+              'Row Count' as "stats:row_count:label",
+              row_count as "stats:row_count:value",
+              'An approximate count of rows in this table' as "stats:row_count:description",
+              (row_count is not null) as "stats:row_count:include",
+
+              'Approximate Size' as "stats:bytes:label",
+              bytes as "stats:bytes:value",
+              'Approximate size of the table as reported by Snowflake' as "stats:bytes:description",
+              (bytes is not null) as "stats:bytes:include"
+
+          from {{ information_schema }}.tables
+
+      ),
+
+      columns as (
+
+          select
+              table_catalog as "table_database",
+              table_schema as "table_schema",
+              table_name as "table_name",
+              null as "table_comment",
+
+              column_name as "column_name",
+              ordinal_position as "column_index",
+              data_type as "column_type",
+              null as "column_comment"
+
+          from {{ information_schema }}.columns
+      )
+
+      select *
+      from tables
+      join columns using ("table_database", "table_schema", "table_name")
+      where (
+        {%- for schema in schemas -%}
+          upper("table_schema") = upper('{{ schema }}'){%- if not loop.last %} or {% endif -%}
+        {%- endfor -%}
+      )
+      order by "column_index"
+    {%- endset -%}
+
+  {{ return(run_query(query)) }}
 
 {%- endmacro %}
diff --git a/test/integration/029_docs_generate_tests/fail_macros/failure.sql b/test/integration/029_docs_generate_tests/fail_macros/failure.sql
index 56c01eb7025..f0519ed245a 100644
--- a/test/integration/029_docs_generate_tests/fail_macros/failure.sql
+++ b/test/integration/029_docs_generate_tests/fail_macros/failure.sql
@@ -1,3 +1,3 @@
-{% macro get_catalog(information_schemas) %}
+{% macro get_catalog(information_schema, schemas) %}
 	{% do exceptions.raise_compiler_error('rejected: no catalogs for you') %}
 {% endmacro %}
diff --git a/test/unit/test_docs_generate.py b/test/unit/test_docs_generate.py
index ede163b5d81..79c60eab37f 100644
--- a/test/unit/test_docs_generate.py
+++ b/test/unit/test_docs_generate.py
@@ -29,6 +29,7 @@ def generate_catalog_dict(self, columns):
         result = generate.CatalogResults(
             nodes=generate.Catalog(columns).make_unique_id_map(self.manifest),
             generated_at=datetime.utcnow(),
+            errors=None,
         )
         return result.to_dict(omit_none=False)['nodes']
 
diff --git a/test/unit/test_postgres_adapter.py b/test/unit/test_postgres_adapter.py
index 5b5c6795dce..16aac9e8c42 100644
--- a/test/unit/test_postgres_adapter.py
+++ b/test/unit/test_postgres_adapter.py
@@ -180,7 +180,8 @@ def test_set_zero_keepalive(self, psycopg2):
             connect_timeout=10)
 
     @mock.patch.object(PostgresAdapter, 'execute_macro')
-    def test_get_catalog_various_schemas(self, mock_execute):
+    @mock.patch.object(PostgresAdapter, '_get_cache_schemas')
+    def test_get_catalog_various_schemas(self, mock_get_schemas, mock_execute):
         column_names = ['table_database', 'table_schema', 'table_name']
         rows = [
             ('dbt', 'foo', 'bar'),
@@ -192,15 +193,18 @@ def test_get_catalog_various_schemas(self, mock_execute):
         mock_execute.return_value = agate.Table(rows=rows,
                                                 column_names=column_names)
 
+        mock_get_schemas.return_value.items.return_value = [(mock.MagicMock(database='dbt'), {'foo', 'FOO', 'quux'})]
+
         mock_manifest = mock.MagicMock()
         mock_manifest.get_used_schemas.return_value = {('dbt', 'foo'),
                                                        ('dbt', 'quux')}
 
-        catalog = self.adapter.get_catalog(mock_manifest)
+        catalog, exceptions = self.adapter.get_catalog(mock_manifest)
         self.assertEqual(
             set(map(tuple, catalog)),
             {('dbt', 'foo', 'bar'), ('dbt', 'FOO', 'baz'), ('dbt', 'quux', 'bar')}
         )
+        self.assertEqual(exceptions, [])
 
 
 class TestConnectingPostgresAdapter(unittest.TestCase):
diff --git a/third-party-stubs/agate/__init__.pyi b/third-party-stubs/agate/__init__.pyi
index 7f21d3badf7..2d1f01f1020 100644
--- a/third-party-stubs/agate/__init__.pyi
+++ b/third-party-stubs/agate/__init__.pyi
@@ -57,6 +57,8 @@ class Table:
     def from_object(cls, obj: Iterable[Dict[str, Any]], *, column_types: Optional['TypeTester'] = None) -> 'Table': ...
     @classmethod
     def from_csv(cls, path: Iterable[str], *, column_types: Optional['TypeTester'] = None) -> 'Table': ...
+    @classmethod
+    def merge(cls, tables: Iterable['Table']) -> 'Table': ...
 
 
 class TypeTester:


Code review comments:
diff_hunk: @@ -1009,29 +1012,56 @@ def _catalog_filter_table(
         """
         return table.where(_catalog_filter_schemas(manifest))
 
-    def _get_catalog_information_schemas(
-        self, manifest: Manifest
-    ) -> List[InformationSchema]:
-        return list(self._get_cache_schemas(manifest).keys())
+    def _get_one_catalog(
+        self,
+        information_schema: InformationSchema,
+        schemas: Set[str],
+        manifest: Manifest,
+    ) -> agate.Table:
 
-    def get_catalog(self, manifest: Manifest) -> agate.Table:
-        """Get the catalog for this manifest by running the get catalog macro.
-        Returns an agate.Table of catalog information.
-        """
-        information_schemas = self._get_catalog_information_schemas(manifest)
-        # make it a list so macros can index into it.
-        kwargs = {'information_schemas': information_schemas}
-        table = self.execute_macro(
-            GET_CATALOG_MACRO_NAME,
-            kwargs=kwargs,
-            release=True,
-            # pass in the full manifest so we get any local project overrides
-            manifest=manifest,
-        )
+        name = '.'.join([
+            str(information_schema.database),
+            'information_schema'
+        ])
+
+        # calculate the possible schemas for a given schema name
+        all_schema_names: Set[str] = set()
+        for schema in schemas:
+            all_schema_names.update({schema, schema.lower(), schema.upper()})
path: core/dbt/adapters/base/impl.py
text: @user1:
It looks to me like we're doing case-insensitive schema filtering in the catalog queries for every plugin -- do we need to pass in a set of the lower/upper cased variants of the schema name here? In dev, I see catalog SQL like this (pg):

```
    where (upper(sch.nspname) = upper('SNAPSHOTS') or upper(sch.nspname) = upper('TEST_SCHEMA') or upper(sch.nspname) = upper('snapshots') or upper(sch.nspname) = upper('test_schema'))
```

@author:
Ooh, you're right - this is why the comparison was case-sensitive in catalog.sql before. I'll remove this.

@user1:
that makes a ton of sense - i didn't put those two things together!

diff_hunk: @@ -229,21 +238,32 @@ def run(self):
         dbt.ui.printer.print_timestamped_line(
             'Catalog written to {}'.format(os.path.abspath(path))
         )
+
+        if exceptions:
+            logger.error(
+                'dbt encountered {} failure{} while writing the catalog'
+                .format(len(exceptions), (len(exceptions) == 1) * 's')
path: core/dbt/task/generate.py
text: @user1:
Do we want `!=` 1 here?

@user1:
also, this is exceedingly clever

@user1:
also, this whole flow looks great in practice!

```
Encountered an error while generating catalog: Database Error
  Access Denied: Table the-psf:INFORMATION_SCHEMA.SCHEMATA: User does not have permission to query table the-psf:INFORMATION_SCHEMA.SCHEMATA.
21:39:33 | Catalog written to /Users/drew/fishtown/clients/bq/target/catalog.json
dbt encountered 1 failures while writing the catalog
```

Can we just move this log line to be up above the `Catalog written to...` line? I want to make it clear that the catalog has still been written despite the errors.

@author:
Hah, yes, it should be `!=1`. And it is exceedingly clever, just a stupid pluralization hack I remember from a long time ago.

diff_hunk: @@ -1104,3 +1134,30 @@ def post_model_hook(self, config: Mapping[str, Any], context: Any) -> None:
         The second parameter is the value returned by pre_mdoel_hook.
         """
         pass
+
+
+def catch_as_completed(
+    futures  # typing: List[Future[agate.Table]]
+) -> Tuple[agate.Table, List[Exception]]:
+
+    catalogs: agate.Table = agate.Table(rows=[])
+    exceptions: List[Exception] = []
+
+    for future in as_completed(futures):
+        exc = future.exception()
+        # we want to re-raise on ctrl+c and BaseException
+        if exc is None:
+            catalog = future.result()
+            catalogs = agate.Table.merge([catalogs, catalog])
path: core/dbt/adapters/base/impl.py
text: @user1:
Agate seems unhappy about this on Snowflake. I see:
```
2020-02-04 02:07:41,316840 (MainThread): Tables contain columns with the same names, but different types.
2020-02-04 02:07:41,322568 (MainThread): Traceback (most recent call last):
  File "/Users/drew/fishtown/dbt/core/dbt/main.py", line 81, in main
    results, succeeded = handle_and_check(args)
  File "/Users/drew/fishtown/dbt/core/dbt/main.py", line 159, in handle_and_check
    task, res = run_from_args(parsed)
  File "/Users/drew/fishtown/dbt/core/dbt/main.py", line 212, in run_from_args
    results = task.run()
  File "/Users/drew/fishtown/dbt/core/dbt/task/generate.py", line 214, in run
    catalog_table, exceptions = adapter.get_catalog(self.manifest)
  File "/Users/drew/fishtown/dbt/core/dbt/adapters/base/impl.py", line 1061, in get_catalog
    catalogs, exceptions = catch_as_completed(futures)
  File "/Users/drew/fishtown/dbt/core/dbt/adapters/base/impl.py", line 1151, in catch_as_completed
    catalogs = agate.Table.merge([catalogs, catalog])
  File "/Users/drew/fishtown/dbt/env/lib/python3.7/site-packages/agate/table/merge.py", line 45, in merge
    raise DataTypeError('Tables contain columns with the same names, but different types.')
agate.exceptions.DataTypeError: Tables contain columns with the same names, but different types.
```

In this case, the error is super annoying. One `information_schema` contains a table with a clustering key defined (of type `string`). The other `information_schema` that gets merged in does not have _any_ clustering keys on any tables, so every value in the `stats:clustering_key:value` columns is `None`, which I guess Agate assigned to an integery type?

I ran some debug code to see what was going wrong here, and I saw that the types for `catalog` and `catalogs` were:
```
stats:clustering_key:value': {<agate.data_types.number.Number object at 0x109d26a10>,
                                <agate.data_types.text.Text object at 0x10a02c390>}
```

While I saw this on Snowflake, I have to imagine we'd see similar things on other databases. Any good ideas about what to do about this one? My (not ideal) thinking is that we could push the raw results into a list, then convert that raw list of lists back into a Table...

@author:
I have a fix for this (roll my own merge function that tracks null-ness). I think it's actually pretty ok.