Repository: dbt-labs/dbt-core
Pull request: (#1576) use the information schema on BigQuery
Fixes #1576 

Work in progress. Use the BigQuery `INFORMATION_SCHEMA` to fetch the catalog. I actually cheat here and use `__TABLES__` (not `INFORMATION_SCHEMA.TABLES`) because the information in `__TABLES__` is a superset of the data in `TABLES` (ie. `row_count` and `size_bytes`).

Couple of things to verify here:
 - do we need to worry about quoting/casing here? I just want to preserve the existing behavior
 - it's hard to find good docs on `__TABLES__` -- is that appropriate for us to use?
 - make sure this fix actually addresses #1576 -- how does this hold up for datasets with many date-sharded tables?

To document:

Date sharded tables can be addressed using dbt sources by replacing the date shard suffix with a `*` in the source specification. When this source is referenced from a model, dbt will expand the wildcard to match all date shards in the table. Additionally, the auto-generated dbt documentation website will correctly collect statistics about _all_ of the date shards in this table.

```
sources:
    - name: source_schema
      tables:
          - name: events_*
```

Problem statement:
Generating docs take and long time
When running `dbt docs generate` is take over 10 mins to complete and uses lots of CPU resources producing the catalog.

When running with debug turned on the following gets written out:

```
2019-06-20 16:57:55,546 (MainThread): On "<None>": cache miss for schema "my_project.my_dataset", this is inefficient
```

along with:

```
2019-06-20 16:57:56,829 (MainThread): with schema=99439170, model_name=None, relations=[
<BigQueryRelation my_project.my_dataset.`cloudaudit_googleapis_com_data_access_20161114`>, 
<BigQueryRelation my_project.my_dataset.`cloudaudit_googleapis_com_data_access_20161115`>
.....
```

which is a date sharded table containing hundreds of shards.

The interesting thing is the `cloudaudit_googleapis_com_data_access` is the only one in the dataset that is not used or referenced by DBT. In the same dataset there is 2 other date sharded tables containing over a thousand shards, which are referenced as a source in DBT, but they are not printed out

```
sources:
  - name: google_analytics
    database: my_project
    schema: my_dataset
    loader: ga360
    tables:
      - name: ga_sessions_*
      - name: ga_sessions_intraday_*
```

Hints from issue discussion:
Thanks for the report @whittid4 - we'll check it out! My initial thinking:
- high cpu usage sounds like an `agate` problem? I recall this being an issue with seeds, but not sure if that's exactly what's happening here
- we're probably iterating over the individual date shards, which is probably undersirable
I have run into a similar problem. I also have DBT setup to read from a BQ database with a lot of date sharded tables. Running `dbt docs generate` takes over an hour for me. There is no real noticable cpu load though

In the dbt logs the following is repeated over 50,000 times: 
```
019-06-26 15:34:11,918 (MainThread): /home/venv/lib/python3.6/site-packages/dbt/adapters/bigquery/impl.py:465: PendingDeprecationWarning: This method will be deprecated in future versions. Please use Table.time_partitioning.type_ instead.
```


Thanks for the additional info @EricLeer!

It's pretty clear to me that we need dbt to be smarter about catalog generation with date sharded tables on BQ. I don't think it makes sense for dbt to know about every single individual date shard -- these are probably source tables, and the BQ interface is probably suited for exploring these shards than dbt docs is, at least currently.

Presently, dbt is using an API method to fetch all of the tables/views in every dataset that dbt touches. This means that dbt will fundamentally need to pull down every single one of these tables, then maybe distill the date-sharded tables down to a single source table in-memory.

Maybe an alternative approach is to use BQ's new-ish information schema? I think with some clever SQL, we can push a lot of this filtering into the BQ layer. This should make `dbt docs generate` a lot snappier, as well as bring BQ in line with the rest of dbt's plugins (for the most part).

I just played around with some code here, what do you guys think of something like this?
```sql
with base as (

  select *,
    REGEXP_CONTAINS(table_name, '^.+[0-9]{8}$') as is_date_shard,
    REGEXP_EXTRACT(table_name, '^(.+)[0-9]{8}$') as base_name,
    REGEXP_EXTRACT(table_name, '^.+([0-9]{8})$') as shard_name

  FROM dbt_dbanin.INFORMATION_SCHEMA.TABLES

),

extracted as (

  select *,
    coalesce(base_name, table_name) as root_table_name

  from base

),

unsharded as (

  select
    table_catalog,
    table_schema,
    root_table_name as table_name,
    
    row_number() over (partition by root_table_name order by shard_name desc) as shard_index,
    min(shard_name) over (partition by root_table_name) as first_shard,
    max(shard_name) over (partition by root_table_name) as last_shard,
    count(*) over (partition by root_table_name) as num_shards

  from extracted
  
)

select *
from unsharded
where shard_index = 1
order by table_name
```

This query will find all of the tables/views in a given dataset, then squash down date shards into a single "root" record. We can pluck out the individual shard names if necessary, maybe aggregating them into an array in SQL? That would be pretty slick.

Curious to hear what you think!

cc @whittid4 
I think this would work. For one of my datasets this would reduce the amount of tables searched from 40000 to 400. 

On the other hand I think you would still run into the same problem if the tables are sharded in a different way then yyyymmdd date. In principle with [querying a wildcard table](https://cloud.google.com/bigquery/docs/querying-wildcard-tables) any pattern is possible and I think it would be best if this behaviour is also supported with this solution. Maybe the correct information of on what to colapse the shards can be gathered from the schema.yml file where the source is defined? For instance I have defined my source as:

```
name: dataset_name
    tables:
      - name: table_name
        identifier: table_name_*
``` 
and thus I would expect the shards to colapse on `table_name_` .

Finally what is the reason that dbt fetches all tables/views in a dataset it touches? Wouldn't it make more sense to only fetch the tables/views that are actually defined as a source?
That's a really good point - I didn't realize that you could shard on strings other than an 8-char date suffix. I agree - generating some sort of glob query from the specified `identifier` is probably a good idea.

dbt tries to find all of the tables in a given BQ dataset that match up with models and sources defined in the active dbt project. By querying for all of the relations in a dataset at once, dbt can make one query per dataset referenced in the dbt project. If dbt instead queried for each source/model individually, then dbt would need to execute orders of magnitude more queries like this in the general case. Imagine you had 30 source tables defined in a single dataset --  all things considered, I'd rather make one query against that dataset than 30!
Sorry for the delay in getting back to you.

@drewbanin, the example SQL you shared above will work for our case as we are only using the standard data sharding method, but as @EricLeer points out it would be nice to make this solution fit most use cases


I tried to point to a specific table during doc generation with dbt 0.14.2 but it didn't help.
It takes hours to process a referenced dataset with thousand (date) sharded tables even if you only reference a specific non-wildcard table in that dataset.
In my case it's a way too long to generate docs for both `dataset_1` and `dataset_2` (please see example YAML below). Both contain more than a thousand sharded tables.
``` yaml
sources:

  - name: dataset_1
    database: main_database
    schema: dataset_1_{{ var("tenant_id") }}
    tables:

      - name: table_1
        identifier: table_1_{{ var("tenant_id") }}_{{ "*" if not var("doc_gen", False) else "20190901" }}

  - name: dataset_2
    database: main_database
    schema: dataset_2
    tables:

      - name: table_2
        identifier: table_2
        columns:
          - name: table_id
```
Hey @mr2dark - I spent some time yesterday pulling together a more completed "catalog" query for BigQuery which leverages the `INFORMATION_SCHEMA`. TBD if we're able to slot that in for the 0.15.0 release (we removed it from the milestone, for now) but I'd love to ship it if we can!

Are you able/interested to help test out a branch of dbt which introduces this logic?
@drewbanin I'll be glad to help. How can I test that?

BTW it looks like my estimation is wrong. It can take 30 minutes and the cases where I didn't receive results for hours were probably caused by power/network state issues when my laptop went to sleep and/or was disconnected from network for a while. I'll try to reproduce that when I'll have some spare time.
But anyway building of a catalog for 30 minutes when you actually use a couple of tables feels like an overkill. 
Cool! I'll follow up with a branch name and some more info when I have something to show for myself :)

Commit message: Include Location field in catalog from the SCHEMATA table

Patch to review:
diff --git a/plugins/bigquery/dbt/adapters/bigquery/impl.py b/plugins/bigquery/dbt/adapters/bigquery/impl.py
index 596aff2a66c..5ca6c2e570f 100644
--- a/plugins/bigquery/dbt/adapters/bigquery/impl.py
+++ b/plugins/bigquery/dbt/adapters/bigquery/impl.py
@@ -1,5 +1,3 @@
-import copy
-
 import dbt.deprecations
 import dbt.exceptions
 import dbt.flags as flags
@@ -25,6 +23,9 @@
 import agate
 
 
+GET_CATALOG_MACRO_NAME = 'get_catalog'
+
+
 def _stub_relation(*args, **kwargs):
     return BigQueryRelation.create(
         database='',
@@ -387,127 +388,31 @@ def load_dataframe(self, database, schema, table_name, agate_table,
         with self.connections.exception_handler("LOAD TABLE"):
             self.poll_until_job_completes(job, timeout)
 
-    ###
-    # The get_catalog implementation for bigquery
-    ###
-    def _flat_columns_in_table(self, table):
-        """An iterator over the flattened columns for a given schema and table.
-        Resolves child columns as having the name "parent.child".
-        """
-        for col in self._get_dbt_columns_from_bq_table(table):
-            yield from col.flatten()
-
-    @classmethod
-    def _get_stats_column_names(cls):
-        """Construct a tuple of the column names for stats. Each stat has 4
-        columns of data.
-        """
-        columns = []
-        stats = ('num_bytes', 'num_rows', 'location', 'partitioning_type',
-                 'clustering_fields')
-        stat_components = ('label', 'value', 'description', 'include')
-        for stat_id in stats:
-            for stat_component in stat_components:
-                columns.append('stats:{}:{}'.format(stat_id, stat_component))
-        return tuple(columns)
-
-    @classmethod
-    def _get_stats_columns(cls, table, relation_type):
-        """Given a table, return an iterator of key/value pairs for stats
-        column names/values.
-        """
-        column_names = cls._get_stats_column_names()
-
-        # agate does not handle the array of column names gracefully
-        clustering_value = None
-        if table.clustering_fields is not None:
-            clustering_value = ','.join(table.clustering_fields)
-        # cast num_bytes/num_rows to str before they get to agate, or else
-        # agate will incorrectly decide they are booleans.
-        column_values = (
-            'Number of bytes',
-            str(table.num_bytes),
-            'The number of bytes this table consumes',
-            relation_type == 'table',
-
-            'Number of rows',
-            str(table.num_rows),
-            'The number of rows in this table',
-            relation_type == 'table',
-
-            'Location',
-            table.location,
-            'The geographic location of this table',
-            True,
-
-            'Partitioning Type',
-            table.partitioning_type,
-            'The partitioning type used for this table',
-            relation_type == 'table',
-
-            'Clustering Fields',
-            clustering_value,
-            'The clustering fields for this table',
-            relation_type == 'table',
-        )
-        return zip(column_names, column_values)
-
     def get_catalog(self, manifest):
-        connection = self.connections.get_thread_connection()
-        client = connection.handle
-
-        schemas = manifest.get_used_schemas()
-
-        column_names = (
-            'table_database',
-            'table_schema',
-            'table_name',
-            'table_type',
-            'table_comment',
-            # does not exist in bigquery, but included for consistency
-            'table_owner',
-            'column_name',
-            'column_index',
-            'column_type',
-            'column_comment',
-        )
-        all_names = column_names + self._get_stats_column_names()
-        columns = []
+        """Get the catalog for this manifest by running the get catalog macro.
+        Returns an agate.Table of catalog information.
+        """
 
-        for database_name, schema_name in schemas:
-            relations = self.list_relations(database_name, schema_name)
-            for relation in relations:
-
-                # This relation contains a subset of the info we care about.
-                # Fetch the full table object here
-                table_ref = self.connections.table_ref(
-                    database_name,
-                    relation.schema,
-                    relation.identifier,
-                    connection
-                )
-                table = client.get_table(table_ref)
-
-                flattened = self._flat_columns_in_table(table)
-                relation_stats = dict(self._get_stats_columns(table,
-                                                              relation.type))
-
-                for index, column in enumerate(flattened, start=1):
-                    column_data = (
-                        relation.database,
-                        relation.schema,
-                        relation.name,
-                        relation.type,
-                        None,
-                        None,
-                        column.name,
-                        index,
-                        column.data_type,
-                        None,
-                    )
-                    column_dict = dict(zip(column_names, column_data))
-                    column_dict.update(copy.deepcopy(relation_stats))
-
-                    columns.append(column_dict)
-
-        return dbt.clients.agate_helper.table_from_data(columns, all_names)
+        information_schemas = []
+        for database, schema in manifest.get_used_schemas():
+            information_schemas.append(self.Relation.create(
+                database=database,
+                schema=schema,
+                quote_policy={
+                    'database': True,
+                    'schema': True
+                }
+            ))
+
+        # 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)
+
+        # BigQuery doesn't allow ":" chars in column names -- remap them here.
+        table = table.rename(column_names={
+            col.name: col.name.replace('__', ':') for col in table.columns
+        })
+
+        return self._catalog_filter_table(table, manifest)
diff --git a/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql b/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql
new file mode 100644
index 00000000000..d52d20a5c95
--- /dev/null
+++ b/plugins/bigquery/dbt/include/bigquery/macros/catalog.sql
@@ -0,0 +1,219 @@
+
+{% macro bigquery__get_catalog(information_schemas) -%}
+
+  {%- call statement('catalog', fetch_result=True) -%}
+    {% for information_schema in information_schemas %}
+      (
+        with schemas as (
+
+            select
+              catalog_name as table_database,
+              schema_name as table_schema,
+              location
+
+            from `{{ information_schema.database }}`.INFORMATION_SCHEMA.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 concat('unknown (', cast(type as string), ')')
+                end as table_type,
+
+                REGEXP_CONTAINS(table_id, '^.+[0-9]{8}$') and type = 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
+
+            from {{ information_schema }}.__TABLES__
+
+        ),
+
+        extracted as (
+
+            select *,
+                case
+                    when is_date_shard then shard_base_name
+                    else original_table_name
+                end as table_name
+
+            from tables
+
+        ),
+
+        unsharded_tables as (
+
+            select
+                table_database,
+                table_schema,
+                table_name,
+                table_type,
+                is_date_shard,
+
+                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,
+
+                max(relation_id) as relation_id
+
+            from extracted
+            group by 1,2,3,4,5
+
+        ),
+
+        info_schema_columns as (
+
+            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,
+
+                is_partitioning_column,
+                clustering_ordinal_position
+
+            from {{ information_schema }}.INFORMATION_SCHEMA.COLUMNS
+            where ordinal_position is not null
+
+        ),
+
+        info_schema_column_paths 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
+
+            from {{ information_schema }}.INFORMATION_SCHEMA.COLUMN_FIELD_PATHS
+            where data_type not like 'STRUCT%'
+
+        ),
+
+        columns as (
+
+            select * except (base_column_name)
+            from info_schema_columns
+            join info_schema_column_paths using (relation_id, base_column_name)
+
+        ),
+
+        column_stats 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
+
+            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,
+
+            columns.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,
+            columns.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 %}
+  {%- endcall -%}
+  {{ return(load_result('catalog').table) }}
+
+{% endmacro %}
diff --git a/test/integration/029_docs_generate_tests/test_docs_generate.py b/test/integration/029_docs_generate_tests/test_docs_generate.py
index 42c86b529a4..a0dbbfb82ae 100644
--- a/test/integration/029_docs_generate_tests/test_docs_generate.py
+++ b/test/integration/029_docs_generate_tests/test_docs_generate.py
@@ -5,7 +5,7 @@
 from unittest.mock import ANY, patch
 
 from test.integration.base import DBTIntegrationTest, use_profile, AnyFloat, \
-    AnyStringWith, normalize
+    AnyString, AnyStringWith, normalize
 
 
 def _read_file(path):
@@ -225,39 +225,48 @@ def _bigquery_stats(self, is_table, partition=None, cluster=None):
                 'value': 'US',
                 'description': 'The geographic location of this table',
                 'include': True,
-            },
+            }
         }
         if is_table:
             stats.update({
                 'num_bytes': {
                     'id': 'num_bytes',
-                    'label': 'Number of bytes',
+                    'label': AnyString(),
                     'value': AnyFloat(),
-                    'description': 'The number of bytes this table consumes',
+                    'description': AnyString(),
                     'include': True,
                 },
                 'num_rows': {
                     'id': 'num_rows',
-                    'label': 'Number of rows',
+                    'label': AnyString(),
                     'value': AnyFloat(),
-                    'description': 'The number of rows in this table',
+                    'description': AnyString(),
                     'include': True,
-                },
+                }
+            })
+
+        if partition is not None:
+            stats.update({
                 'partitioning_type': {
                     'id': 'partitioning_type',
-                    'label': 'Partitioning Type',
+                    'label': AnyString(),
                     'value': partition,
-                    'description': 'The partitioning type used for this table',
-                    'include': True,
-                },
+                    'description': AnyString(),
+                    'include': True
+                }
+            })
+
+        if cluster is not None:
+            stats.update({
                 'clustering_fields': {
                     'id': 'clustering_fields',
-                    'label': 'Clustering Fields',
+                    'label': AnyString(),
                     'value': cluster,
-                    'description': 'The clustering fields for this table',
-                    'include': True,
-                },
+                    'description': AnyString(),
+                    'include': True
+                }
             })
+
         return stats
 
     def _expected_catalog(self, id_type, text_type, time_type, view_type,
@@ -535,10 +544,10 @@ def expected_bigquery_complex_catalog(self):
         my_schema_name = self.unique_schema()
         role = self.get_role()
         table_stats = self._bigquery_stats(True)
-        clustering_stats = self._bigquery_stats(True, partition='DAY',
+        clustering_stats = self._bigquery_stats(True, partition='updated_at',
                                                 cluster='first_name')
-        multi_clustering_stats = self._bigquery_stats(True, partition='DAY',
-                                                      cluster='first_name,email')
+        multi_clustering_stats = self._bigquery_stats(True, partition='updated_at',
+                                                      cluster='first_name, email')
         nesting_columns = {
             'field_1': {
                 'name': 'field_1',
diff --git a/test/integration/base.py b/test/integration/base.py
index 28eb4f18ef4..e8a5a8984d7 100644
--- a/test/integration/base.py
+++ b/test/integration/base.py
@@ -1157,6 +1157,13 @@ def __eq__(self, other):
         return isinstance(other, float)
 
 
+class AnyString:
+    """Any string. Use this in assertEqual() calls to assert that it is a float.
+    """
+    def __eq__(self, other):
+        return isinstance(other, str)
+
+
 class AnyStringWith:
     def __init__(self, contains=None):
         self.contains = contains


Code review comments:
diff_hunk: @@ -387,127 +388,31 @@ def load_dataframe(self, database, schema, table_name, agate_table,
         with self.connections.exception_handler("LOAD TABLE"):
             self.poll_until_job_completes(job, timeout)
 
-    ###
-    # The get_catalog implementation for bigquery
-    ###
-    def _flat_columns_in_table(self, table):
-        """An iterator over the flattened columns for a given schema and table.
-        Resolves child columns as having the name "parent.child".
-        """
-        for col in self._get_dbt_columns_from_bq_table(table):
-            yield from col.flatten()
-
-    @classmethod
-    def _get_stats_column_names(cls):
-        """Construct a tuple of the column names for stats. Each stat has 4
-        columns of data.
-        """
-        columns = []
-        stats = ('num_bytes', 'num_rows', 'location', 'partitioning_type',
-                 'clustering_fields')
-        stat_components = ('label', 'value', 'description', 'include')
-        for stat_id in stats:
-            for stat_component in stat_components:
-                columns.append('stats:{}:{}'.format(stat_id, stat_component))
-        return tuple(columns)
-
-    @classmethod
-    def _get_stats_columns(cls, table, relation_type):
-        """Given a table, return an iterator of key/value pairs for stats
-        column names/values.
-        """
-        column_names = cls._get_stats_column_names()
-
-        # agate does not handle the array of column names gracefully
-        clustering_value = None
-        if table.clustering_fields is not None:
-            clustering_value = ','.join(table.clustering_fields)
-        # cast num_bytes/num_rows to str before they get to agate, or else
-        # agate will incorrectly decide they are booleans.
-        column_values = (
-            'Number of bytes',
-            str(table.num_bytes),
-            'The number of bytes this table consumes',
-            relation_type == 'table',
-
-            'Number of rows',
-            str(table.num_rows),
-            'The number of rows in this table',
-            relation_type == 'table',
-
-            'Location',
-            table.location,
-            'The geographic location of this table',
-            True,
-
-            'Partitioning Type',
-            table.partitioning_type,
-            'The partitioning type used for this table',
-            relation_type == 'table',
-
-            'Clustering Fields',
-            clustering_value,
-            'The clustering fields for this table',
-            relation_type == 'table',
-        )
-        return zip(column_names, column_values)
-
     def get_catalog(self, manifest):
-        connection = self.connections.get_thread_connection()
-        client = connection.handle
-
-        schemas = manifest.get_used_schemas()
-
-        column_names = (
-            'table_database',
-            'table_schema',
-            'table_name',
-            'table_type',
-            'table_comment',
-            # does not exist in bigquery, but included for consistency
-            'table_owner',
-            'column_name',
-            'column_index',
-            'column_type',
-            'column_comment',
-        )
-        all_names = column_names + self._get_stats_column_names()
-        columns = []
+        """Get the catalog for this manifest by running the get catalog macro.
+        Returns an agate.Table of catalog information.
+        """
 
-        for database_name, schema_name in schemas:
-            relations = self.list_relations(database_name, schema_name)
-            for relation in relations:
-
-                # This relation contains a subset of the info we care about.
-                # Fetch the full table object here
-                table_ref = self.connections.table_ref(
-                    database_name,
-                    relation.schema,
-                    relation.identifier,
-                    connection
-                )
-                table = client.get_table(table_ref)
-
-                flattened = self._flat_columns_in_table(table)
-                relation_stats = dict(self._get_stats_columns(table,
-                                                              relation.type))
-
-                for index, column in enumerate(flattened, start=1):
-                    column_data = (
-                        relation.database,
-                        relation.schema,
-                        relation.name,
-                        relation.type,
-                        None,
-                        None,
-                        column.name,
-                        index,
-                        column.data_type,
-                        None,
-                    )
-                    column_dict = dict(zip(column_names, column_data))
-                    column_dict.update(copy.deepcopy(relation_stats))
-
-                    columns.append(column_dict)
-
-        return dbt.clients.agate_helper.table_from_data(columns, all_names)
+        information_schemas = []
+        for database, schema in manifest.get_used_schemas():
+            information_schemas.append(self.Relation.create(
+                database=database,
+                schema=schema,
+                quote_policy={
+                    'database': True,
+                    'schema': True
+                }
+            ))
path: plugins/bigquery/dbt/adapters/bigquery/impl.py
text: @user1:
The existing `get_catalog`'s call to `_get_cache_schemas` should do this, and if it doesn't we should fix it for BigQuery! You might have to override the `BigQueryRelation.information_schema` method in some way.


@author:
The big difference from the other `get_catalog` implementations here is that the BigQuery information schema is (usually) addressed with:
```
`project-id`.`dataset`.INFORMATION_SCHEMA.COLUMNS
```

ie. the information_schema is affixed to a dataset (schema) not a project (database). 

The exception is `SCHEMATA` which is addressed at the project level:
```
`project-id`.INFORMATION_SCHEMA.SCHEMATA
```

Do you still think we should override `get_cache_schemas`? I think I might also need to override `SchemaSearchMap` to return information schema Relations that BigQuery is happy with.

@user1:
Hmm, ok. In that case for now how about just implementing `_get_cache_schemas` as:
```
for database, schema in manifest.get_used_schemas():
    yield self.Relation.create(
        database=database,
        schema=schema,
        quote_policy={
            'database': True,
            'schema': True
        }
    )
```

In the long run, I think we should probably extend bigquery's Relation subclass to fully account for this quirky interpretation of `information_schema`, but we can do that at a later date.

@author:
just a couple of great minds, thinking alike.

diff_hunk: @@ -387,127 +388,31 @@ def load_dataframe(self, database, schema, table_name, agate_table,
         with self.connections.exception_handler("LOAD TABLE"):
             self.poll_until_job_completes(job, timeout)
 
-    ###
-    # The get_catalog implementation for bigquery
-    ###
-    def _flat_columns_in_table(self, table):
-        """An iterator over the flattened columns for a given schema and table.
-        Resolves child columns as having the name "parent.child".
-        """
-        for col in self._get_dbt_columns_from_bq_table(table):
-            yield from col.flatten()
-
-    @classmethod
-    def _get_stats_column_names(cls):
-        """Construct a tuple of the column names for stats. Each stat has 4
-        columns of data.
-        """
-        columns = []
-        stats = ('num_bytes', 'num_rows', 'location', 'partitioning_type',
-                 'clustering_fields')
-        stat_components = ('label', 'value', 'description', 'include')
-        for stat_id in stats:
-            for stat_component in stat_components:
-                columns.append('stats:{}:{}'.format(stat_id, stat_component))
-        return tuple(columns)
-
-    @classmethod
-    def _get_stats_columns(cls, table, relation_type):
-        """Given a table, return an iterator of key/value pairs for stats
-        column names/values.
-        """
-        column_names = cls._get_stats_column_names()
-
-        # agate does not handle the array of column names gracefully
-        clustering_value = None
-        if table.clustering_fields is not None:
-            clustering_value = ','.join(table.clustering_fields)
-        # cast num_bytes/num_rows to str before they get to agate, or else
-        # agate will incorrectly decide they are booleans.
-        column_values = (
-            'Number of bytes',
-            str(table.num_bytes),
-            'The number of bytes this table consumes',
-            relation_type == 'table',
-
-            'Number of rows',
-            str(table.num_rows),
-            'The number of rows in this table',
-            relation_type == 'table',
-
-            'Location',
-            table.location,
-            'The geographic location of this table',
-            True,
-
-            'Partitioning Type',
-            table.partitioning_type,
-            'The partitioning type used for this table',
-            relation_type == 'table',
-
-            'Clustering Fields',
-            clustering_value,
-            'The clustering fields for this table',
-            relation_type == 'table',
-        )
-        return zip(column_names, column_values)
-
     def get_catalog(self, manifest):
path: plugins/bigquery/dbt/adapters/bigquery/impl.py
text: @user1:
Instead of overriding `BaseAdapter.get_catalog`, we should use its existing behavior and make BigQuery behave more like other adapters do! I've outlined my thoughts on that ~below~ above, in github's rendering.

diff_hunk: @@ -387,127 +388,31 @@ def load_dataframe(self, database, schema, table_name, agate_table,
         with self.connections.exception_handler("LOAD TABLE"):
             self.poll_until_job_completes(job, timeout)
 
-    ###
-    # The get_catalog implementation for bigquery
-    ###
-    def _flat_columns_in_table(self, table):
-        """An iterator over the flattened columns for a given schema and table.
-        Resolves child columns as having the name "parent.child".
-        """
-        for col in self._get_dbt_columns_from_bq_table(table):
-            yield from col.flatten()
-
-    @classmethod
-    def _get_stats_column_names(cls):
-        """Construct a tuple of the column names for stats. Each stat has 4
-        columns of data.
-        """
-        columns = []
-        stats = ('num_bytes', 'num_rows', 'location', 'partitioning_type',
-                 'clustering_fields')
-        stat_components = ('label', 'value', 'description', 'include')
-        for stat_id in stats:
-            for stat_component in stat_components:
-                columns.append('stats:{}:{}'.format(stat_id, stat_component))
-        return tuple(columns)
-
-    @classmethod
-    def _get_stats_columns(cls, table, relation_type):
-        """Given a table, return an iterator of key/value pairs for stats
-        column names/values.
-        """
-        column_names = cls._get_stats_column_names()
-
-        # agate does not handle the array of column names gracefully
-        clustering_value = None
-        if table.clustering_fields is not None:
-            clustering_value = ','.join(table.clustering_fields)
-        # cast num_bytes/num_rows to str before they get to agate, or else
-        # agate will incorrectly decide they are booleans.
-        column_values = (
-            'Number of bytes',
-            str(table.num_bytes),
-            'The number of bytes this table consumes',
-            relation_type == 'table',
-
-            'Number of rows',
-            str(table.num_rows),
-            'The number of rows in this table',
-            relation_type == 'table',
-
-            'Location',
-            table.location,
-            'The geographic location of this table',
-            True,
-
-            'Partitioning Type',
-            table.partitioning_type,
-            'The partitioning type used for this table',
-            relation_type == 'table',
-
-            'Clustering Fields',
-            clustering_value,
-            'The clustering fields for this table',
-            relation_type == 'table',
-        )
-        return zip(column_names, column_values)
-
     def get_catalog(self, manifest):
-        connection = self.connections.get_thread_connection()
-        client = connection.handle
-
-        schemas = manifest.get_used_schemas()
-
-        column_names = (
-            'table_database',
-            'table_schema',
-            'table_name',
-            'table_type',
-            'table_comment',
-            # does not exist in bigquery, but included for consistency
-            'table_owner',
-            'column_name',
-            'column_index',
-            'column_type',
-            'column_comment',
-        )
-        all_names = column_names + self._get_stats_column_names()
-        columns = []
+        """Get the catalog for this manifest by running the get catalog macro.
+        Returns an agate.Table of catalog information.
+        """
 
-        for database_name, schema_name in schemas:
-            relations = self.list_relations(database_name, schema_name)
-            for relation in relations:
-
-                # This relation contains a subset of the info we care about.
-                # Fetch the full table object here
-                table_ref = self.connections.table_ref(
-                    database_name,
-                    relation.schema,
-                    relation.identifier,
-                    connection
-                )
-                table = client.get_table(table_ref)
-
-                flattened = self._flat_columns_in_table(table)
-                relation_stats = dict(self._get_stats_columns(table,
-                                                              relation.type))
-
-                for index, column in enumerate(flattened, start=1):
-                    column_data = (
-                        relation.database,
-                        relation.schema,
-                        relation.name,
-                        relation.type,
-                        None,
-                        None,
-                        column.name,
-                        index,
-                        column.data_type,
-                        None,
-                    )
-                    column_dict = dict(zip(column_names, column_data))
-                    column_dict.update(copy.deepcopy(relation_stats))
-
-                    columns.append(column_dict)
-
-        return dbt.clients.agate_helper.table_from_data(columns, all_names)
+        information_schemas = []
+        for database, schema in manifest.get_used_schemas():
+            information_schemas.append(self.Relation.create(
+                database=database,
+                schema=schema,
+                quote_policy={
+                    'database': True,
+                    'schema': True
+                }
+            ))
+
+        # 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)
path: plugins/bigquery/dbt/adapters/bigquery/impl.py
text: @user1:
The base adapter already does this in `get_catalog`

diff_hunk: @@ -387,127 +388,31 @@ def load_dataframe(self, database, schema, table_name, agate_table,
         with self.connections.exception_handler("LOAD TABLE"):
             self.poll_until_job_completes(job, timeout)
 
-    ###
-    # The get_catalog implementation for bigquery
-    ###
-    def _flat_columns_in_table(self, table):
-        """An iterator over the flattened columns for a given schema and table.
-        Resolves child columns as having the name "parent.child".
-        """
-        for col in self._get_dbt_columns_from_bq_table(table):
-            yield from col.flatten()
-
-    @classmethod
-    def _get_stats_column_names(cls):
-        """Construct a tuple of the column names for stats. Each stat has 4
-        columns of data.
-        """
-        columns = []
-        stats = ('num_bytes', 'num_rows', 'location', 'partitioning_type',
-                 'clustering_fields')
-        stat_components = ('label', 'value', 'description', 'include')
-        for stat_id in stats:
-            for stat_component in stat_components:
-                columns.append('stats:{}:{}'.format(stat_id, stat_component))
-        return tuple(columns)
-
-    @classmethod
-    def _get_stats_columns(cls, table, relation_type):
-        """Given a table, return an iterator of key/value pairs for stats
-        column names/values.
-        """
-        column_names = cls._get_stats_column_names()
-
-        # agate does not handle the array of column names gracefully
-        clustering_value = None
-        if table.clustering_fields is not None:
-            clustering_value = ','.join(table.clustering_fields)
-        # cast num_bytes/num_rows to str before they get to agate, or else
-        # agate will incorrectly decide they are booleans.
-        column_values = (
-            'Number of bytes',
-            str(table.num_bytes),
-            'The number of bytes this table consumes',
-            relation_type == 'table',
-
-            'Number of rows',
-            str(table.num_rows),
-            'The number of rows in this table',
-            relation_type == 'table',
-
-            'Location',
-            table.location,
-            'The geographic location of this table',
-            True,
-
-            'Partitioning Type',
-            table.partitioning_type,
-            'The partitioning type used for this table',
-            relation_type == 'table',
-
-            'Clustering Fields',
-            clustering_value,
-            'The clustering fields for this table',
-            relation_type == 'table',
-        )
-        return zip(column_names, column_values)
-
     def get_catalog(self, manifest):
-        connection = self.connections.get_thread_connection()
-        client = connection.handle
-
-        schemas = manifest.get_used_schemas()
-
-        column_names = (
-            'table_database',
-            'table_schema',
-            'table_name',
-            'table_type',
-            'table_comment',
-            # does not exist in bigquery, but included for consistency
-            'table_owner',
-            'column_name',
-            'column_index',
-            'column_type',
-            'column_comment',
-        )
-        all_names = column_names + self._get_stats_column_names()
-        columns = []
+        """Get the catalog for this manifest by running the get catalog macro.
+        Returns an agate.Table of catalog information.
+        """
 
-        for database_name, schema_name in schemas:
-            relations = self.list_relations(database_name, schema_name)
-            for relation in relations:
-
-                # This relation contains a subset of the info we care about.
-                # Fetch the full table object here
-                table_ref = self.connections.table_ref(
-                    database_name,
-                    relation.schema,
-                    relation.identifier,
-                    connection
-                )
-                table = client.get_table(table_ref)
-
-                flattened = self._flat_columns_in_table(table)
-                relation_stats = dict(self._get_stats_columns(table,
-                                                              relation.type))
-
-                for index, column in enumerate(flattened, start=1):
-                    column_data = (
-                        relation.database,
-                        relation.schema,
-                        relation.name,
-                        relation.type,
-                        None,
-                        None,
-                        column.name,
-                        index,
-                        column.data_type,
-                        None,
-                    )
-                    column_dict = dict(zip(column_names, column_data))
-                    column_dict.update(copy.deepcopy(relation_stats))
-
-                    columns.append(column_dict)
-
-        return dbt.clients.agate_helper.table_from_data(columns, all_names)
+        information_schemas = []
+        for database, schema in manifest.get_used_schemas():
+            information_schemas.append(self.Relation.create(
+                database=database,
+                schema=schema,
+                quote_policy={
+                    'database': True,
+                    'schema': True
+                }
+            ))
+
+        # 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)
+
+        # BigQuery doesn't allow ":" chars in column names -- remap them here.
+        table = table.rename(column_names={
+            col.name: col.name.replace('__', ':') for col in table.columns
+        })
+
+        return self._catalog_filter_table(table, manifest)
path: plugins/bigquery/dbt/adapters/bigquery/impl.py
text: @user1:
```
def self._catalog_filter_table(self, table, manifest):
    # BigQuery doesn't allow ":" chars in column names -- remap them here.
    table = table.rename(column_names={
        col.name: col.name.replace('__', ':') for col in table.columns
    })

    return super()._catalog_filter_table(table, manifest)
```

diff_hunk: @@ -1157,6 +1157,13 @@ def __eq__(self, other):
         return isinstance(other, float)
 
 
+class AnyString:
+    """Any string. Use this in assertEqual() calls to assert that it is a float.
path: test/integration/base.py
text: @user1:
`... to assert that it is a str.`!