Override check schema exists

.changes/unreleased/Features-20230407-104723.yaml

@@ -0,0 +1,5 @@
+kind: Features
+time: 2023-04-07T10:47:23.105369-07:00
+custom:
+  Author: jiezhen-chen
+  Issue: 555

dbt/adapters/redshift/connections.py

@@ -142,9 +142,6 @@ def _connection_keys(self):
             "schema",
             "sslmode",
             "region",
-            "sslmode",
-            "region",
-            "iam_profile",
             "autocreate",
             "db_groups",
             "ra3_node",

dbt/adapters/redshift/impl.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Optional, Set, Any, Dict, Type
+from typing import Optional, Set, Any, Dict, Type, List
 from collections import namedtuple
 from dbt.adapters.base import PythonJobHelper
 from dbt.adapters.base.impl import AdapterConfig, ConstraintSupport
@@ -115,6 +115,21 @@ def valid_incremental_strategies(self):
     def timestamp_add_sql(self, add_to: str, number: int = 1, interval: str = "hour") -> str:
         return f"{add_to} + interval '{number} {interval}'"
 
+    def _get_cursor(self):
+        return self.connections.get_thread_connection().handle.cursor()
+
+    def list_schemas(self, database: str, schema: Optional[str] = None) -> List[str]:
+        cursor = self._get_cursor()
+        results = []
+        for s in cursor.get_schemas(catalog=database, schema_pattern=schema):
+            results.append(s[0])
+        return results
+
+    @available
+    def check_schema_exists(self, database: str, schema: str) -> bool:
+        results = self.list_schemas(database=database, schema=schema)
+        return len(results) > 0
+
     def _link_cached_database_relations(self, schemas: Set[str]):
         """
         :param schemas: The set of schemas that should have links added.

tests/functional/adapter/test_adapter_methods.py

@@ -16,6 +16,30 @@
 
 """
 
+models__get_invalid_schema = """
+
+{% set upstream = ref('upstream') %}
+
+{% set existing = adapter.check_schema_exists(upstream.database, "doesnotexist") %}
+{% if existing == False %}
+select 2 as id
+{% else %}
+select 1 as id
+{% endif %}
+"""
+
+models__get_valid_schema = """
+
+{% set upstream = ref('upstream') %}
+
+{% set existing = adapter.check_schema_exists(upstream.database, upstream.schema) %}
+{% if existing == True %}
+select 2 as id
+{% else %}
+select 1 as id
+{% endif %}
+"""
+
 models__upstream_sql = """
 select 1 as id
 
@@ -84,6 +108,9 @@ def models(self):
             "base_view.sql": "{{ config(bind=True) }} select * from {{ ref('model') }}",
             "get_relation_type.sql": models__get_relation_type,
             "expected_type.sql": "select 1 as valid_type",
+            "get_invalid_schema.sql": models__get_invalid_schema,
+            "get_valid_schema.sql": models__get_invalid_schema,
+            "get_schema_expected.sql": "select 2 as id",
         }
 
     def project_files(
@@ -104,11 +131,13 @@ def project_config_update(self):
     def test_adapter_methods(self, project):
         run_dbt(["compile"])  # trigger any compile-time issues
         result = run_dbt()
-        assert len(result) == 7
+        assert len(result) == 10
 
         run_dbt(["test"])
         check_relations_equal(project.adapter, ["call_get_relation", "expected"])
         check_relations_equal(project.adapter, ["get_relation_type", "expected_type"])
+        check_relations_equal(project.adapter, ["get_invalid_schema", "get_schema_expected"])
+        check_relations_equal(project.adapter, ["get_valid_schema", "get_schema_expected"])
 
 
 class TestRedshiftAdapterMethod(RedshiftAdapterMethod):

tests/unit/test_redshift_adapter.py

@@ -587,6 +587,18 @@ def mock_cursor(self, mock_get_thread_conn):
         mock_handle.return_value = mock_cursor
         return mock_cursor
 
+    @mock.patch("dbt.adapters.redshift.impl.RedshiftAdapter._get_cursor")
+    def test_list_schemas(self, mock_cursor):
+        mock_cursor.return_value.get_schemas.return_value = ["schema1"], ["schema2"]
+        results = self.adapter.list_schemas(database="somedb", schema="someschema")
+        self.assertTrue(results == ["schema1", "schema2"])
+
+    @mock.patch("dbt.adapters.redshift.impl.RedshiftAdapter.list_schemas")
+    def test_check_schema_exists(self, mock_list_schemas):
+        mock_list_schemas.return_value = ["schema1", "schema2"]
+        results = self.adapter.check_schema_exists(database="somedb", schema="someschema")
+        self.assertTrue(results is True)
+
 
 class TestRedshiftAdapterConversions(TestAdapterConversions):
     def test_convert_text_type(self):