Implement relation filtering on get_catalog macro (#1073)

* changelog

* turn on get_catalog by relations

* add get_catalog_relations, break apart get_catalog for reuse in get_catalog_relations

---------

Co-authored-by: colin-rogers-dbt <[email protected]>

.changes/unreleased/Features-20231219-201203.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support limiting get_catalog by object name
+time: 2023-12-19T20:12:03.990725-05:00
+custom:
+  Author: mikealfare
+  Issue: "950"

dbt/adapters/bigquery/impl.py

@@ -19,7 +19,7 @@
 )
 from dbt.adapters.base.impl import FreshnessResponse
 from dbt.adapters.cache import _make_ref_key_dict  # type: ignore
-from dbt.adapters.capability import CapabilityDict, CapabilitySupport, Support, Capability
+from dbt.adapters.capability import Capability, CapabilityDict, CapabilitySupport, Support
 import dbt.clients.agate_helper
 from dbt.contracts.connection import AdapterResponse
 from dbt.contracts.graph.manifest import Manifest
@@ -120,10 +120,8 @@ class BigQueryAdapter(BaseAdapter):
         ConstraintType.foreign_key: ConstraintSupport.ENFORCED,
     }
 
-    _capabilities: CapabilityDict = CapabilityDict(
-        {
-            Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
-        }
+    _capabilities = CapabilityDict(
+        {Capability.SchemaMetadataByRelations: CapabilitySupport(support=Support.Full)}
     )
 
     def __init__(self, config) -> None:

dbt/include/bigquery/macros/catalog/by_relation.sql

@@ -0,0 +1,36 @@
+{% macro bigquery__get_catalog_relations(information_schema, relations) -%}
+
+    {%- if (relations | 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 -%}
+            with
+                table_shards_stage as ({{ _bigquery__get_table_shards_sql(information_schema) }}),
+                table_shards as (
+                    select * from table_shards_stage
+                    where (
+                        {%- for relation in relations -%}
+                            (
+                                upper(table_schema) = upper('{{ relation.schema }}')
+                            and upper(table_name) = upper('{{ relation.identifier }}')
+                            )
+                            {%- if not loop.last %} or {% endif -%}
+                        {%- endfor -%}
+                    )
+                ),
+                tables as ({{ _bigquery__get_tables_sql() }}),
+                table_stats as ({{ _bigquery__get_table_stats_sql() }}),
+
+                columns as ({{ _bigquery__get_columns_sql(information_schema) }}),
+                column_stats as ({{ _bigquery__get_column_stats_sql() }})
+
+            {{ _bigquery__get_extended_catalog_sql() }}
+        {%- endset -%}
+
+    {%- endif -%}
+
+    {{ return(run_query(query)) }}
+
+{%- endmacro %}

dbt/include/bigquery/macros/catalog/by_schema.sql

@@ -0,0 +1,32 @@
+{% macro bigquery__get_catalog(information_schema, schemas) -%}
+
+    {%- 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 -%}
+            with
+                table_shards as (
+                    {{ _bigquery__get_table_shards_sql(information_schema) }}
+                    where (
+                        {%- for schema in schemas -%}
+                            upper(tables.dataset_id) = upper('{{ schema }}')
+                            {%- if not loop.last %} or {% endif -%}
+                        {%- endfor -%}
+                    )
+                ),
+                tables as ({{ _bigquery__get_tables_sql() }}),
+                table_stats as ({{ _bigquery__get_table_stats_sql() }}),
+
+                columns as ({{ _bigquery__get_columns_sql(information_schema) }}),
+                column_stats as ({{ _bigquery__get_column_stats_sql() }})
+
+            {{ _bigquery__get_extended_catalog_sql() }}
+        {%- endset -%}
+
+    {%- endif -%}
+
+    {{ return(run_query(query)) }}
+
+{%- endmacro %}