Add support for pseudocolumns in unit tests (#1526)

Co-authored-by: Colin Rogers <111200756+colin-rogers-dbt@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 3 people

dbt-adapters/.changes/unreleased/Features-20260118-111437.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Update unit test column retrieval to support pseudocolumns
+time: 2026-01-18T11:14:37.820863-05:00
+custom:
+    Author: katieclaiborne-duet
+    Issue: "1517"

dbt-adapters/src/dbt/adapters/base/impl.py

@@ -33,6 +33,7 @@
     AdapterStandardizeGrantsDictRecord,
     AdapterListRelationsWithoutCachingRecord,
     AdapterGetColumnsInRelationRecord,
+    AdapterGetPseudocolumnsForRelationRecord,
     SubmitPythonJobRecord,
 )
 from dbt_common.behavior_flags import Behavior, BehaviorFlag
@@ -820,6 +821,27 @@ def get_columns_in_relation(self, relation: BaseRelation) -> List[BaseColumn]:
         """Get a list of the columns in the given Relation."""
         raise NotImplementedError("`get_columns_in_relation` is not implemented for this adapter!")
 
+    @record_function(
+        AdapterGetPseudocolumnsForRelationRecord,
+        method=True,
+        index_on_thread_id=True,
+        id_field_name="thread_id",
+    )
+    @available.parse_list
+    def get_pseudocolumns_for_relation(self, relation: BaseRelation) -> List[BaseColumn]:
+        """Get a list of queryable pseudocolumns for the given relation.
+
+        Pseudocolumns are system-generated columns that can be queried but don't
+        appear in the information schema (e.g., BigQuery's _FILE_NAME for external tables).
+
+        Default implementation returns an empty list. Adapters should override this
+        to provide pseudocolumns specific to their platform.
+
+        :param relation: The relation to get pseudocolumns for
+        :return: List of Column objects representing queryable pseudocolumns
+        """
+        return []
+
     def get_catalog_for_single_relation(self, relation: BaseRelation) -> Optional[CatalogTable]:
         """Get catalog information including table-level and column-level metadata for a single relation."""
         raise NotImplementedError(

dbt-adapters/src/dbt/adapters/record/base.py

@@ -321,3 +321,47 @@ class AdapterGetColumnsInRelationRecord(Record):
     params_cls = AdapterGetColumnsInRelationParams
     result_cls = AdapterGetColumnsInRelationResult
     group = "Available"
+
+
+@dataclasses.dataclass
+class AdapterGetPseudocolumnsForRelationParams:
+    thread_id: str
+    relation: "BaseRelation"
+
+    def _to_dict(self):
+        from dbt.adapters.record.serialization import serialize_base_relation
+
+        return {
+            "thread_id": self.thread_id,
+            "relation": serialize_base_relation(self.relation),
+        }
+
+    def _from_dict(self, data: Dict[str, Any]):
+        from dbt.adapters.record.serialization import deserialize_base_relation
+
+        self.thread_id = data["thread_id"]
+        self.relation = deserialize_base_relation(data["relation"])
+
+
+@dataclasses.dataclass
+class AdapterGetPseudocolumnsForRelationResult:
+    return_val: List["BaseColumn"]
+
+    def _to_dict(self):
+        from dbt.adapters.record.serialization import serialize_base_column_list
+
+        return {"return_val": serialize_base_column_list(self.return_val)}
+
+    def _from_dict(self, data: Dict[str, Any]):
+        from dbt.adapters.record.serialization import deserialize_base_column_list
+
+        self.return_val = deserialize_base_column_list(data["return_val"])
+
+
+@Recorder.register_record_type
+class AdapterGetPseudocolumnsForRelationRecord(Record):
+    """Implements record/replay support for the BaseAdapter.get_pseudocolumns_for_relation() method."""
+
+    params_cls = AdapterGetPseudocolumnsForRelationParams
+    result_cls = AdapterGetPseudocolumnsForRelationResult
+    group = "Available"

dbt-adapters/src/dbt/include/global_project/macros/adapters/columns.sql

@@ -16,6 +16,23 @@
   {{ return(columns) }}
 {% endmacro %}
 
+{% macro get_columns_for_unit_tests(relation) -%}
+  {{ return(adapter.dispatch('get_columns_for_unit_tests', 'dbt')(relation)) }}
+{% endmacro %}
+
+{% macro default__get_columns_for_unit_tests(relation) -%}
+  {#-- Get native columns from information schema --#}
+  {% set columns = adapter.get_columns_in_relation(relation) %}
+
+  {#-- Add pseudocolumns that are queryable but not in information schema --#}
+  {% set pseudocolumns = adapter.get_pseudocolumns_for_relation(relation) %}
+
+  {#-- Merge two lists --#}
+  {% set all_columns = columns + pseudocolumns %}
+
+  {{ return(all_columns) }}
+{% endmacro %}
+
 {%- macro get_list_of_column_names(columns) -%}
   {% set col_names = [] %}
   {% for col in columns %}

dbt-adapters/src/dbt/include/global_project/macros/materializations/tests/unit.sql

@@ -10,7 +10,7 @@
   {%- set target_relation = this.incorporate(type='table') -%}
   {%- set temp_relation = make_temp_relation(target_relation)-%}
   {% do run_query(get_create_table_as_sql(True, temp_relation, get_empty_subquery_sql(sql))) %}
-  {%- set columns_in_relation = adapter.get_columns_in_relation(temp_relation) -%}
+  {%- set columns_in_relation = get_columns_for_unit_tests(temp_relation) -%}
   {%- set column_name_to_data_types = {} -%}
   {%- set column_name_to_quoted = {} -%}
   {%- for column in columns_in_relation -%}

dbt-adapters/src/dbt/include/global_project/macros/unit_test_sql/get_fixture_sql.sql

@@ -5,7 +5,7 @@
 {%- if not column_name_to_data_types -%}
 {#-- Use defer_relation IFF it is available in the manifest and 'this' is missing from the database --#}
 {%-   set this_or_defer_relation = defer_relation if (defer_relation and not load_relation(this)) else this -%}
-{%-   set columns_in_relation = adapter.get_columns_in_relation(this_or_defer_relation) -%}
+{%-   set columns_in_relation = get_columns_for_unit_tests(this_or_defer_relation) -%}
 
 {%-   set column_name_to_data_types = {} -%}
 {%-   set column_name_to_quoted = {} -%}

dbt-bigquery/.changes/unreleased/Features-20260106-090826.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support for pseudocolumns in unit tests
+time: 2026-01-06T09:08:26.620597-05:00
+custom:
+    Author: katieclaiborne-duet
+    Issue: "1517"

dbt-bigquery/src/dbt/adapters/bigquery/impl.py

@@ -372,6 +372,23 @@ def get_columns_in_relation(self, relation: BigQueryRelation) -> List[BigQueryCo
             logger.debug("get_columns_in_relation error: {}".format(e))
             return []
 
+    @available.parse_list
+    def get_columns_and_pseudocolumns_for_relation(
+        self, relation: BigQueryRelation
+    ) -> List[BigQueryColumn]:
+        """Return all columns (regular + pseudocolumns) for a relation in a single API call."""
+        try:
+            table = self.connections.get_bq_table(
+                database=relation.database, schema=relation.schema, identifier=relation.identifier
+            )
+            columns = self._get_dbt_columns_from_bq_table(table)
+            if table.table_type == "EXTERNAL":
+                columns.append(BigQueryColumn("_FILE_NAME", "STRING"))
+            return columns
+        except (ValueError, google.cloud.exceptions.NotFound) as e:
+            logger.debug("get_columns_and_pseudocolumns_for_relation error: {}".format(e))
+            return []
+
     @available.parse(lambda *a, **k: [])
     def add_time_ingestion_partition_column(self, partition_by, columns) -> List[BigQueryColumn]:
         """Add time ingestion partition column to columns list"""

dbt-bigquery/src/dbt/include/bigquery/macros/adapters/columns.sql

@@ -1,3 +1,7 @@
+{% macro bigquery__get_columns_for_unit_tests(relation) -%}
+  {{ return(adapter.get_columns_and_pseudocolumns_for_relation(relation)) }}
+{%- endmacro %}
+
 {% macro bigquery__get_empty_subquery_sql(select_sql, select_sql_header=none) %}
     {%- if select_sql_header is not none -%}
     {{ select_sql_header }}

dbt-bigquery/tests/functional/adapter/unit_testing/test_pseudocolumns.py

@@ -0,0 +1,94 @@
+import pytest
+from unittest.mock import patch
+
+from dbt.tests.adapter.unit_testing.test_pseudocolumns import (
+    BasePseudocolumnUnitTest,
+    external_table_csv,
+)
+
+
+# Model that uses BigQuery's _FILE_NAME pseudocolumn
+my_model_sql = """
+select
+    id,
+    _FILE_NAME as file_name
+from {{ source('test_source', 'external_table') }}
+"""
+
+# Source definition for external table
+schema_sources_yml = """
+sources:
+  - name: test_source
+    schema: "{{ target.schema }}"
+    tables:
+      - name: external_table
+        columns:
+          - name: id
+"""
+
+# Unit test that includes _FILE_NAME pseudocolumn in the fixture
+test_my_model_yml = """
+unit_tests:
+  - name: test_bigquery_file_name_pseudocolumn
+    model: my_model
+    given:
+      - input: source('test_source', 'external_table')
+        rows:
+          - {id: 1, _FILE_NAME: 'gs://bucket/file1.csv'}
+          - {id: 2, _FILE_NAME: 'gs://bucket/file2.csv'}
+    expect:
+      rows:
+        - {id: 1, file_name: 'gs://bucket/file1.csv'}
+        - {id: 2, file_name: 'gs://bucket/file2.csv'}
+"""
+
+
+class TestBigQueryPseudocolumns(BasePseudocolumnUnitTest):
+    """Test BigQuery's _FILE_NAME pseudocolumn support in unit tests.
+
+    Uses BigQuery-specific SQL (_FILE_NAME as file_name) and validates that the
+    pseudocolumn framework works with BigQueryColumn types.
+    """
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        """Seed data."""
+        return {
+            "external_table.csv": external_table_csv,
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        """Models using BigQuery's _FILE_NAME pseudocolumn."""
+        return {
+            "my_model.sql": my_model_sql,
+            "sources.yml": schema_sources_yml + test_my_model_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def setup_pseudocolumn_override(self, project):
+        """Patch BigQueryAdapter at the class level to return columns + _FILE_NAME for external_table.
+
+        Instance-level patching (project.adapter.X) doesn't work because dbtRunner
+        creates its own adapter instance. Class-level patching affects all instances.
+        """
+        from dbt.adapters.bigquery.column import BigQueryColumn
+        from dbt.adapters.bigquery.impl import BigQueryAdapter
+
+        original_method = BigQueryAdapter.get_columns_and_pseudocolumns_for_relation
+
+        def mock_columns_and_pseudocolumns(_self, relation):
+            if relation.identifier == "external_table":
+                return [BigQueryColumn("id", "INTEGER"), BigQueryColumn("_FILE_NAME", "STRING")]
+            # Fall through to the real method for temp relations (model output schema lookup)
+            return original_method(_self, relation)
+
+        mock_columns_and_pseudocolumns._is_available_ = True
+        mock_columns_and_pseudocolumns._parse_replacement_ = lambda *a, **k: []
+
+        with patch.object(
+            BigQueryAdapter,
+            "get_columns_and_pseudocolumns_for_relation",
+            mock_columns_and_pseudocolumns,
+        ):
+            yield