refactor(redshift): move temp relation fallback into get_columns_in_relation

Follow-up to #2097. That PR reached the driver-based temp relation
fallback through a new @available method called from
redshift__get_columns_in_relation. A new adapter-specific method has no
record type behind it, so the conformance harness cannot record or
replay it.

Move the dispatch into an override of get_columns_in_relation instead.
@supports_replay re-applies AdapterGetColumnsInRelationRecord to
subclass overrides, so the fallback gets record/replay coverage without
a new record type, and porting to v2 is an override of an existing
interface method rather than a Redshift-only addition.

get_columns_in_temp_relation keeps its name but loses @available: it is
now reached only from get_columns_in_relation, which is where the call
is recorded. Behaviour is unchanged -- the SHOW branch already required
a qualified relation, so unqualified relations took the legacy branch
either way.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: tauhid621

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

@@ -11,6 +11,7 @@
 from collections import namedtuple
 from dbt.adapters.base import PythonJobHelper
 from dbt.adapters.base.impl import AdapterConfig, ConstraintSupport, FreshnessResponse
+from dbt.adapters.base.column import Column as BaseColumn
 from dbt.adapters.base.meta import available
 from dbt.adapters.base.relation import BaseRelation
 from dbt.adapters.capability import (
@@ -239,8 +240,26 @@ def use_show_apis(self) -> bool:
         """
         return bool(self.config.credentials.datasharing)
 
-    @available
-    def get_columns_in_temp_relation(self, relation: BaseRelation) -> List[Any]:
+    def get_columns_in_relation(self, relation: BaseRelation) -> List[BaseColumn]:
+        """Describe a relation, falling back to the driver when the catalog cannot see it.
+
+        On a datashare consumer database, temp relations stay queryable but are absent from
+        ``information_schema.columns``, ``pg_attribute`` and ``svv_columns``; the empty
+        result makes ``on_schema_change='sync_all_columns'`` drop every column in the
+        target. Only unqualified relations can take the fallback, since the driver query
+        has no database or schema to qualify itself with.
+
+        In an override rather than in the macro so that ``@supports_replay`` records the
+        fallback under the existing ``AdapterGetColumnsInRelationRecord``.
+        """
+        columns = super().get_columns_in_relation(relation)
+
+        if not columns and not relation.database and not relation.schema:
+            return self.get_columns_in_temp_relation(relation)
+
+        return columns
+
+    def get_columns_in_temp_relation(self, relation: BaseRelation) -> List[BaseColumn]:
         """Describe a temporary relation from the driver instead of the catalog.
 
         Needed when the connection's database is a datashare consumer database: temp

dbt-redshift/src/dbt/include/redshift/macros/adapters.sql

@@ -130,24 +130,7 @@
   {% if redshift__use_show_apis() and relation.database and relation.schema %}
     {{ return(redshift__get_columns_in_relation_show(relation)) }}
   {% else %}
-    {%- set columns = redshift__get_columns_in_relation_legacy(relation) -%}
-
-    {#-
-      A temp relation with no columns means the catalog could not see it, not that it has
-      none: when the connection's database is a datashare consumer database, temp relations
-      stay queryable but are absent from information_schema.columns, pg_attribute and
-      svv_columns. Left unhandled, an empty list makes on_schema_change='sync_all_columns'
-      treat every column in the target as removed and drop it.
-
-      Only temp relations are affected, and only they can be described this way: the driver
-      query is unqualified, so it resolves to the right relation only when there is no
-      database or schema to qualify it with.
-    -#}
-    {%- if columns | length == 0 and not relation.database and not relation.schema -%}
-      {{ return(adapter.get_columns_in_temp_relation(relation)) }}
-    {%- endif -%}
-
-    {{ return(columns) }}
+    {{ return(redshift__get_columns_in_relation_legacy(relation)) }}
   {% endif %}
 {% endmacro %}
 

dbt-redshift/tests/unit/test_temp_relation_columns.py

@@ -18,6 +18,7 @@
 import pytest
 from dbt_common.exceptions import DbtRuntimeError
 
+from dbt.adapters.sql import SQLAdapter
 from dbt.adapters.redshift.impl import (
     TYPE_OID_TO_DATA_TYPE,
     TYPE_OID_TO_INFORMATION_SCHEMA_DATA_TYPE,
@@ -140,6 +141,51 @@ def quote(identifier):
     _temp_relation_data_type = RedshiftAdapter._temp_relation_data_type
 
 
+class TestGetColumnsInRelationFallback:
+    """When the catalog result falls back to the driver, and when it must not."""
+
+    def _adapter(self):
+        # Bypass __init__: the override needs only a real RedshiftAdapter for zero-arg super().
+        adapter = RedshiftAdapter.__new__(RedshiftAdapter)
+        adapter.get_columns_in_temp_relation = mock.Mock(return_value=["from_driver"])
+        return adapter
+
+    def test_catalog_result_is_returned_without_touching_the_driver(self):
+        adapter = self._adapter()
+        relation = mock.Mock(database=None, schema=None, identifier="model__dbt_tmp123")
+
+        with mock.patch.object(
+            SQLAdapter, "get_columns_in_relation", return_value=["from_catalog"]
+        ):
+            assert RedshiftAdapter.get_columns_in_relation(adapter, relation) == ["from_catalog"]
+
+        adapter.get_columns_in_temp_relation.assert_not_called()
+
+    def test_invisible_temp_relation_falls_back_to_the_driver(self):
+        adapter = self._adapter()
+        relation = mock.Mock(database=None, schema=None, identifier="model__dbt_tmp123")
+
+        with mock.patch.object(SQLAdapter, "get_columns_in_relation", return_value=[]):
+            assert RedshiftAdapter.get_columns_in_relation(adapter, relation) == ["from_driver"]
+
+        adapter.get_columns_in_temp_relation.assert_called_once_with(relation)
+
+    def test_qualified_relation_with_no_columns_does_not_fall_back(self):
+        # The driver query is unqualified, so it would resolve to the wrong relation.
+        adapter = self._adapter()
+        relation = mock.Mock(database="db", schema="sch", identifier="my_model")
+
+        with mock.patch.object(SQLAdapter, "get_columns_in_relation", return_value=[]):
+            assert RedshiftAdapter.get_columns_in_relation(adapter, relation) == []
+
+        adapter.get_columns_in_temp_relation.assert_not_called()
+
+    def test_driver_fallback_is_not_exposed_to_jinja(self):
+        # @available here would put a method with no record type back in the Jinja context.
+        assert "get_columns_in_temp_relation" not in RedshiftAdapter._available_
+        assert "get_columns_in_relation" in RedshiftAdapter._available_
+
+
 class TestGetColumnsInTempRelation:
     def _adapter_with_columns(self, columns):
         adapter = _StubAdapter(columns)