allow for adapters to have only a database or only a schema in a catalog query

core/dbt/adapters/base/impl.py

@@ -107,13 +107,28 @@ def _catalog_filter_schemas(manifest: Manifest) -> Callable[[agate.Row], bool]:
     schemas = frozenset((d.lower(), s.lower()) for d, s in manifest.get_used_schemas())
 
     def test(row: agate.Row) -> bool:
-        table_database = _expect_row_value("table_database", row)
-        table_schema = _expect_row_value("table_schema", row)
-        # the schema may be present but None, which is not an error and should
-        # be filtered out
-        if table_schema is None:
+        if "table_database" in row.keys():
+            table_database = _expect_row_value("table_database", row)
+        else:
+            table_database = None
+
+        if "table_schema" in row.keys():
+            table_schema = _expect_row_value("table_schema", row)
+            # the schema may be present but None, which is not an error and should
+            # be filtered out
+            if table_schema is None:
+                return False
+        else:
+            table_schema = None
+
+        if table_database and table_schema:
+            return (table_database.lower(), table_schema.lower()) in schemas
+        elif table_schema:
+            return table_schema in {s for _, s in schemas}
+        elif table_database:
+            return table_database in {d for d, _ in schemas}
+        else:
             return False
-        return (table_database.lower(), table_schema.lower()) in schemas
 
     return test
 
@@ -1179,12 +1194,21 @@ class BaseAdapter(metaclass=AdapterMeta):
             }
 
             def in_map(row: agate.Row):
-                d = _expect_row_value("table_database", row)
-                s = _expect_row_value("table_schema", row)
+                if "table_database" in row.keys():
+                    d = _expect_row_value("table_database", row)
+                    d = d.casefold() if d is not None else None
+                else:
+                    d = None
+
+                if "table_schema" in row.keys():
+                    s = _expect_row_value("table_schema", row)
+                    s = s.casefold() if s is not None else None
+                else:
+                    s = None
+
                 i = _expect_row_value("table_name", row)
-                d = d.casefold() if d is not None else None
-                s = s.casefold() if s is not None else None
                 i = i.casefold() if i is not None else None
+
                 return (d, s, i) in relation_map
 
             catalogs = catalogs.where(in_map)

tests/adapter/dbt/tests/adapter/catalog/files.py

@@ -0,0 +1,16 @@
+MODELS__VIEW = """
+select 1 as id
+"""
+
+
+MACROS__GET_CATALOG = """
+{% macro default__get_catalog(information_schema, schemas) -%}
+
+  {% set typename = adapter.type() %}
+  {% set msg -%}
+    get_catalog not implemented for {{ typename }}
+  {%- endset %}
+
+  {{ exceptions.raise_compiler_error(msg) }}
+{% endmacro %}
+"""

tests/adapter/dbt/tests/adapter/catalog/test_include_policy.py

@@ -0,0 +1,28 @@
+import pytest
+
+from dbt.tests.util import run_dbt
+from tests.adapter.dbt.tests.adapter.catalog import files
+
+
+class CatalogRecognizesIncludePolicy:
+    """
+    This test addresses: https://github.com/dbt-labs/dbt-core/issues/9013
+
+    To implement this, overwrite `macros` with the version of `get_catalog` that is
+    specific to your adapter. Remember to remove database, schema, or both to test the flexibility.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"my_view.sql": files.MODELS__VIEW}
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"default__get_catalog.sql": files.MACROS__GET_CATALOG}
+
+    def test_include_policy_recognized_during_docs_generate(self, project):
+        """
+        Running successfully is passing
+        """
+        run_dbt(["run"])
+        run_dbt(["docs", "generate"])

tests/functional/catalog_tests/files.py

@@ -31,3 +31,64 @@ MY_MATERIALIZED_VIEW = """
 select *
 from {{ ref('my_seed') }}
 """
+
+
+# this is the same as `postgres__get_catalog_relations`, but with `table_database` removed
+MACROS__GET_CATALOG_RELATIONS = """
+{% macro postgres__get_catalog_relations(information_schema, relations) -%}
+  {%- 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_schema.database %}
+    {{ adapter.verify_database(database) }}
+
+    select
+        sch.nspname as table_schema,
+        tbl.relname as table_name,
+        case tbl.relkind
+            when 'v' then 'VIEW'
+            when 'm' then 'MATERIALIZED VIEW'
+            else 'BASE TABLE'
+        end as table_type,
+        tbl_desc.description as table_comment,
+        col.attname as column_name,
+        col.attnum as column_index,
+        pg_catalog.format_type(col.atttypid, col.atttypmod) as column_type,
+        col_desc.description as column_comment,
+        pg_get_userbyid(tbl.relowner) as table_owner
+
+    from pg_catalog.pg_namespace sch
+    join pg_catalog.pg_class tbl on tbl.relnamespace = sch.oid
+    join pg_catalog.pg_attribute col on col.attrelid = tbl.oid
+    left outer join pg_catalog.pg_description tbl_desc on (tbl_desc.objoid = tbl.oid and tbl_desc.objsubid = 0)
+    left outer join pg_catalog.pg_description col_desc on (col_desc.objoid = tbl.oid and col_desc.objsubid = col.attnum)
+    where (
+      {%- for relation in relations -%}
+        {%- if relation.identifier -%}
+          (upper(sch.nspname) = upper('{{ relation.schema }}') and
+           upper(tbl.relname) = upper('{{ relation.identifier }}'))
+        {%- else-%}
+          upper(sch.nspname) = upper('{{ relation.schema }}')
+        {%- endif -%}
+        {%- 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 in ('p', 'u') -- [p]ermanent table or [u]nlogged table. Exclude [t]emporary tables
+      and tbl.relkind in ('r', 'v', 'f', 'p', 'm') -- o[r]dinary table, [v]iew, [f]oreign table, [p]artitioned table, [m]aterialized view. Other values are [i]ndex, [S]equence, [c]omposite type, [t]OAST table
+      and col.attnum > 0 -- negative numbers are used for system columns such as oid
+      and not col.attisdropped -- column as not been dropped
+
+    order by
+        sch.nspname,
+        tbl.relname,
+        col.attnum
+
+  {%- endcall -%}
+
+  {{ return(load_result('catalog').table) }}
+{%- endmacro %}
+"""

tests/functional/catalog_tests/test_include_policy.py

@@ -0,0 +1,12 @@
+import pytest
+
+from tests.adapter.dbt.tests.adapter.catalog.test_include_policy import (
+    CatalogRecognizesIncludePolicy,
+)
+from tests.functional.catalog_tests import files
+
+
+class TestCatalogRecognizesIncludePolicy(CatalogRecognizesIncludePolicy):
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"postgres__get_catalog_relations.sql": files.MACROS__GET_CATALOG_RELATIONS}