fix(redshift): describe temp relations from the driver when the catalog cannot see them (#2097)

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Tauhid Anjum <tauhidanjum@gmail.com>

Author: 3 people

dbt-redshift/.changes/unreleased/Fixes-20260728-120003.yaml

@@ -0,0 +1,8 @@
+kind: Fixes
+body: Describe temporary relations from the driver when the connection targets a datashare
+    consumer database, where they are invisible to all catalog views. Previously this made
+    on_schema_change='sync_all_columns' drop every column in the target table.
+time: 2026-07-28T12:00:03.029795-07:00
+custom:
+    Author: will-sargent-dbtlabs
+    Issue: "1947 1991"

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

@@ -42,6 +42,46 @@
 GET_RELATIONS_MACRO_NAME = "redshift__get_relations"
 SHOW_TABLES_FROM_SCHEMA_MACRO_NAME = "redshift__show_tables_from_schema"
 
+# Redshift type OID -> SQL data type name, as reported by both the legacy path's
+# information_schema.columns.data_type and SHOW COLUMNS. Used to describe temp relations
+# from the driver's cursor description (see get_columns_in_temp_relation); the names have
+# to match whichever of those two paths described the target relation, or schema comparison
+# reports a type change on every run. Verified against SHOW COLUMNS on Redshift 1.0.358853
+# -- note bpchar reports as "character", and varbyte as "binary varying".
+TYPE_OID_TO_DATA_TYPE: Dict[int, str] = {
+    16: "boolean",  # bool
+    20: "bigint",  # int8
+    21: "smallint",  # int2
+    23: "integer",  # int4
+    25: "character varying",  # text
+    700: "real",  # float4
+    701: "double precision",  # float8
+    1042: "character",  # bpchar
+    1043: "character varying",  # varchar
+    1082: "date",  # date
+    1083: "time without time zone",  # time
+    1114: "timestamp without time zone",  # timestamp
+    1184: "timestamp with time zone",  # timestamptz
+    1188: "interval year to month",  # intervaly2m
+    1190: "interval day to second",  # intervald2s
+    1266: "time with time zone",  # timetz
+    1700: "numeric",  # numeric
+    2935: "hllsketch",  # hllsketch
+    3000: "geometry",  # geometry
+    3001: "geography",  # geography
+    4000: "super",  # super
+    6551: "binary varying",  # varbyte
+}
+
+# information_schema reports its internal type name where SHOW COLUMNS reports the SQL name,
+# so a fallback column only compares equal if it follows whichever path described the target.
+# Everything else in the map above agrees across both; these are the exceptions, verified in
+# tests/functional/test_type_oid_mapping.py.
+TYPE_OID_TO_INFORMATION_SCHEMA_DATA_TYPE: Dict[int, str] = {
+    1188: "intervaly2m",  # SHOW COLUMNS: interval year to month
+    1190: "intervald2s",  # SHOW COLUMNS: interval day to second
+}
+
 REDSHIFT_SKIP_AUTOCOMMIT_TRANSACTION_STATEMENTS = BehaviorFlag(
     name="redshift_skip_autocommit_transaction_statements",
     default=False,
@@ -199,6 +239,111 @@ def use_show_apis(self) -> bool:
         """
         return bool(self.config.credentials.datasharing)
 
+    @available
+    def get_columns_in_temp_relation(self, relation: BaseRelation) -> List[Any]:
+        """Describe a temporary relation from the driver instead of the catalog.
+
+        Needed when the connection's database is a datashare consumer database: temp
+        relations stay queryable but are invisible to ``information_schema.columns``,
+        ``pg_attribute`` and ``svv_columns`` alike, and ``SHOW COLUMNS`` cannot address
+        them because they carry no database or schema.
+
+        Sizes come from ``cursor.ps["row_desc"]``'s ``type_modifier``, not from
+        ``cursor.description``: ``redshift_connector`` hardcodes the PEP 249
+        size/precision/scale fields to ``None`` (see its ``Cursor._getDescription``).
+        Without real sizes, every sized column looks changed against the target on the
+        next run and gets needlessly rewritten.
+        """
+        sql = f"select * from {self.quote(relation.identifier)} limit 0"
+        _, cursor = self.connections.add_select_query(sql)
+
+        # `cursor.ps["row_desc"]` is an internal, undocumented structure of
+        # `redshift_connector`, not a public API -- there is no supported alternative for
+        # reaching `type_modifier`. `_getDescription` builds `cursor.description` from this
+        # same list, in the same order, so index-aligned lookup here is safe.
+        try:
+            row_descriptions = cursor.ps["row_desc"]
+        except (AttributeError, KeyError, TypeError):
+            row_descriptions = []
+
+        columns = []
+        for i, description in enumerate(cursor.description or []):
+            # PEP 249: (name, type_code, display_size, internal_size, precision, scale, null_ok)
+            column_name = description[0]
+            type_code = description[1]
+
+            type_modifier = None
+            if i < len(row_descriptions):
+                type_modifier = row_descriptions[i].get("type_modifier")
+
+            data_type = self._temp_relation_data_type(type_code)
+
+            # Only string and exact-numeric types feed size into Column.data_type, which is
+            # what schema comparison comes down to. For every other type the size fields are
+            # ignored, so leave them unset.
+            char_size = None
+            numeric_precision = None
+            numeric_scale = None
+            if type_modifier is not None and type_modifier >= 0:
+                # Postgres wire-protocol convention: VARHDRSZ (4 bytes) is added to both a
+                # string's declared length and a numeric's packed (precision, scale) pair.
+                # -1 means "no modifier" (unconstrained/default size) and is left unset.
+                raw_modifier = type_modifier - 4
+                if data_type in ("character varying", "character"):
+                    char_size = raw_modifier if raw_modifier > 0 else None
+                elif data_type == "numeric":
+                    numeric_precision = raw_modifier >> 16
+                    numeric_scale = raw_modifier & 0xFFFF
+
+            columns.append(
+                self.Column(
+                    column=column_name,
+                    dtype=data_type,
+                    char_size=char_size,
+                    numeric_precision=numeric_precision,
+                    numeric_scale=numeric_scale,
+                )
+            )
+
+        return columns
+
+    def _temp_relation_data_type(self, type_code: Any) -> str:
+        """Map a driver type code onto the name the target relation's describer reports.
+
+        The target goes through SHOW COLUMNS when ``datasharing`` is on and through
+        information_schema when it is off, and the two disagree on a few type names.
+        """
+        if not self.use_show_apis():
+            data_type = TYPE_OID_TO_INFORMATION_SCHEMA_DATA_TYPE.get(type_code)
+            if data_type is not None:
+                return data_type
+
+        data_type = TYPE_OID_TO_DATA_TYPE.get(type_code)
+        if data_type is not None:
+            return data_type
+
+        # Unmapped OID. The driver's own label is closer than nothing, but its lookup is an
+        # IntEnum call that raises for codes it doesn't recognise either -- and a column
+        # whose type cannot be named at all is better reported than described with an
+        # invented name, which would only fail later as invalid DDL.
+        try:
+            fallback = str(self.connections.data_type_code_to_name(type_code)).lower()
+        except Exception as exc:
+            raise dbt_common.exceptions.DbtRuntimeError(
+                f"Cannot describe temporary relation column: Redshift returned type code "
+                f"{type_code}, which neither dbt-redshift nor redshift_connector "
+                f"recognises. Please report this at "
+                f"https://github.com/dbt-labs/dbt-adapters/issues"
+            ) from exc
+
+        logger.debug(
+            f"No known Redshift data type for type code {type_code}; "
+            f"falling back to {fallback!r}. Schema comparison for this column may be "
+            f"inaccurate -- please report this at "
+            f"https://github.com/dbt-labs/dbt-adapters/issues"
+        )
+        return fallback
+
     @available
     def drop_without_cascade(self) -> bool:
         """Whether to omit CASCADE from DROP statements.

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

@@ -130,7 +130,24 @@
   {% if redshift__use_show_apis() and relation.database and relation.schema %}
     {{ return(redshift__get_columns_in_relation_show(relation)) }}
   {% else %}
-    {{ return(redshift__get_columns_in_relation_legacy(relation)) }}
+    {%- 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) }}
   {% endif %}
 {% endmacro %}
 

dbt-redshift/tests/functional/adapter/datashare_consumer/__init__.py

@@ -0,0 +1,12 @@
+import pytest
+
+from tests.functional.adapter.datashare_consumer.fixtures import (
+    REDSHIFT_TEST_DATASHARE_DBNAME,
+)
+
+
+@pytest.fixture(autouse=True, scope="session")
+def _skip_without_datashare_db():
+    """Skip every test in this directory when the env var is not set."""
+    if not REDSHIFT_TEST_DATASHARE_DBNAME:
+        pytest.skip("REDSHIFT_TEST_DATASHARE_DBNAME not set")

dbt-redshift/tests/functional/adapter/datashare_consumer/conftest.py

@@ -0,0 +1,60 @@
+import os
+
+import pytest
+
+# Name of a database created FROM A DATASHARE on the cluster under test, i.e. one that
+# svv_redshift_databases reports with database_type = 'shared'. The connection's dbname is
+# pointed directly at it, which is the configuration these tests exist to cover.
+#
+# Setting one up (Redshift Serverless, two namespaces):
+#
+#   for role in producer consumer; do
+#     aws redshift-serverless create-namespace --namespace-name rs-$role \
+#       --admin-username admin --admin-user-password '<password>' --db-name dev
+#     aws redshift-serverless create-workgroup --workgroup-name rs-$role-wg \
+#       --namespace-name rs-$role --base-capacity 8 --publicly-accessible
+#   done
+#
+# Both workgroups need inbound 5439 from the test runner. The namespace GUIDs below come from
+# `get-namespace --query namespace.namespaceId`. Then:
+#
+#   -- producer namespace, database `dev`
+#   create schema shared_sch;
+#   create table shared_sch.orders (id int);
+#   create datashare repro_share;
+#   alter datashare repro_share add schema shared_sch;
+#   alter datashare repro_share add table shared_sch.orders;
+#   grant usage on datashare repro_share to namespace '<consumer-namespace-guid>';
+#
+#   -- consumer namespace
+#   create database ds_consumer from datashare repro_share of namespace '<producer-guid>';
+#
+# then set REDSHIFT_TEST_DATASHARE_DBNAME=ds_consumer and point the test profile's host at
+# the consumer workgroup.
+REDSHIFT_TEST_DATASHARE_DBNAME = os.getenv("REDSHIFT_TEST_DATASHARE_DBNAME", "")
+
+
+class DatashareConsumerMixin:
+    """Connect directly to a datashare consumer database, with datasharing enabled.
+
+    This is deliberately different from CrossDatabaseMixin, which leaves the connection on
+    the default database and only retargets models via `+database`. That arrangement works;
+    connecting straight at the consumer database is the one that breaks, because temporary
+    relations are then invisible to every catalog view.
+    """
+
+    @pytest.fixture(scope="class")
+    def profiles_config_update(self, dbt_profile_target, unique_schema):
+        return {
+            "test": {
+                "outputs": {
+                    "default": {
+                        **dbt_profile_target,
+                        "schema": unique_schema,
+                        "dbname": REDSHIFT_TEST_DATASHARE_DBNAME,
+                        "datasharing": True,
+                    }
+                },
+                "target": "default",
+            }
+        }

dbt-redshift/tests/functional/adapter/datashare_consumer/fixtures.py

@@ -0,0 +1,100 @@
+"""
+Incremental on_schema_change against a connection pointed directly at a datashare consumer
+database -- the configuration that reproduces dbt-labs/dbt-adapters#1947 and #1991.
+
+The existing coverage passes because it never exercises this arrangement:
+
+    | test                                       | connection dbname | model database        |
+    |--------------------------------------------|-------------------|-----------------------|
+    | TestIncrementalOnSchemaChangeWithDatasharing | default           | same as connection    |
+    | TestIncrementalCrossDatabase*                | default           | cross-db via +database|
+    | these tests                                  | consumer database | same as connection    |
+
+In the last row, temporary relations are created and stay queryable but are absent from
+information_schema.columns, pg_attribute and svv_columns, so column introspection returns
+nothing. Before the driver-based fallback, sync_all_columns read that as "the source has no
+columns" and dropped every column in the target.
+"""
+
+from dbt.tests.adapter.incremental.test_incremental_on_schema_change import (
+    BaseIncrementalOnSchemaChange,
+)
+from dbt.tests.util import get_connection, run_dbt
+import pytest
+
+from tests.functional.adapter.datashare_consumer.fixtures import DatashareConsumerMixin
+
+
+class TestIncrementalOnSchemaChangeDatashareConsumer(
+    DatashareConsumerMixin, BaseIncrementalOnSchemaChange
+):
+    """The full on_schema_change suite, run against a datashare consumer database."""
+
+
+_MODEL_SYNC_ALL_COLUMNS = """
+{{
+    config(
+        materialized='incremental',
+        unique_key='id',
+        on_schema_change='sync_all_columns'
+    )
+}}
+
+with source_data as (
+    select 1 as id, 'aaa' as field_1, 2.5::numeric(18,2) as field_2, 'x'::char(5) as field_3
+    union all select 2 as id, 'bbb' as field_1, 3.5::numeric(18,2) as field_2, 'y'::char(5) as field_3
+)
+
+select * from source_data
+
+{% if is_incremental() %}
+where id not in (select id from {{ this }})
+{% endif %}
+"""
+
+
+class TestIncrementalTempRelationColumnsPreserved(DatashareConsumerMixin):
+    """Targeted regression test: an incremental run must not drop the target's columns.
+
+    Also guards the type mapping: because the source relation is described from the driver
+    and the target from SHOW COLUMNS, any disagreement in reported data types shows up as a
+    perpetual schema change. Running twice and asserting a stable column set catches that --
+    char and numeric columns are included specifically because they are the ones whose
+    reported type carries a size.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"incremental_sync_all_columns.sql": _MODEL_SYNC_ALL_COLUMNS}
+
+    def _columns(self, project):
+        relation = project.adapter.Relation.create(
+            database=project.database,
+            schema=project.test_schema,
+            identifier="incremental_sync_all_columns",
+        )
+        with get_connection(project.adapter):
+            return project.adapter.get_columns_in_relation(relation)
+
+    def test_columns_survive_incremental_run(self, project):
+        run_dbt(["run", "--select", "incremental_sync_all_columns"])
+        before = self._columns(project)
+        assert [c.name for c in before] == ["id", "field_1", "field_2", "field_3"]
+
+        run_dbt(["run", "--select", "incremental_sync_all_columns"])
+        after = self._columns(project)
+
+        assert [c.name for c in after] == ["id", "field_1", "field_2", "field_3"]
+        # Data types must be stable too, otherwise the driver-described source and the
+        # SHOW COLUMNS-described target disagree and dbt alters the column type every run.
+        assert [c.dtype for c in after] == [c.dtype for c in before]
+
+    def test_third_run_is_still_stable(self, project):
+        """A type-mapping mismatch would keep re-triggering; a third run pins that down."""
+        run_dbt(["run", "--select", "incremental_sync_all_columns"])
+        run_dbt(["run", "--select", "incremental_sync_all_columns"])
+        first = self._columns(project)
+        run_dbt(["run", "--select", "incremental_sync_all_columns"])
+        assert [(c.name, c.dtype) for c in self._columns(project)] == [
+            (c.name, c.dtype) for c in first
+        ]