Implement relation filtering on get_catalog macro

.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

@@ -23,7 +23,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_common.clients.agate_helper
 from dbt.adapters.contracts.connection import AdapterResponse
 from dbt.adapters.contracts.macros import MacroResolverProtocol
@@ -126,6 +126,7 @@ class BigQueryAdapter(BaseAdapter):
     _capabilities: CapabilityDict = CapabilityDict(
         {
             Capability.TableLastModifiedMetadata: CapabilitySupport(support=Support.Full),
+            Capability.SchemaMetadataByRelations: CapabilitySupport(support=Support.Full),
         }
     )
 

dbt/adapters/bigquery/relation_configs/_partition.py

@@ -111,7 +111,7 @@ def parse_model_node(cls, relation_config: RelationConfig) -> Dict[str, Any]:
             This doesn't currently collect `time_ingestion_partitioning` and `copy_partitions`
             because this was built for materialized views, which do not support those settings.
         """
-        config_dict: Dict[str, str] = relation_config.config.extra.get("partition_by")  # type: ignore
+        config_dict: Dict[str, Any] = relation_config.config.extra.get("partition_by")  # type: ignore
         if "time_ingestion_partitioning" in config_dict:
             del config_dict["time_ingestion_partitioning"]
         if "copy_partitions" in config_dict:

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 %}