feat(connection): add datashare support, disabled by default

Author: Brooke-white

redshift_connector/__init__.py

@@ -110,6 +110,7 @@ def connect(
     force_lowercase: bool = False,
     allow_db_user_override: bool = False,
     client_protocol_version: int = DEFAULT_PROTOCOL_VERSION,
+    database_metadata_current_db_only: bool = True,
 ) -> Connection:
 
     info: RedshiftProperty = RedshiftProperty()
@@ -151,6 +152,7 @@ def connect(
         force_lowercase=force_lowercase,
         allow_db_user_override=allow_db_user_override,
         client_protocol_version=client_protocol_version,
+        database_metadata_current_db_only=database_metadata_current_db_only,
     )
 
     return Connection(
@@ -169,6 +171,7 @@ def connect(
         application_name=info.application_name,
         replication=info.replication,
         client_protocol_version=info.client_protocol_version,
+        database_metadata_current_db_only=database_metadata_current_db_only,
     )
 
 

redshift_connector/config.py

@@ -103,7 +103,7 @@ def get_name(cls, i: int) -> str:
     "unicode": "utf-8",  # Needed for Amazon Redshift
 }
 
-table_type_clauses: dict = {
+table_type_clauses: typing.Dict[str, typing.Optional[typing.Dict[str, str]]] = {
     "TABLE": {
         "SCHEMAS": "c.relkind = 'r' AND n.nspname !~ '^pg_' AND n.nspname <> 'information_schema'",
         "NOSCHEMAS": "c.relkind = 'r' AND c.relname !~ '^pg_'",
@@ -158,4 +158,5 @@ def get_name(cls, i: int) -> str:
         "NOSCHEMAS": "c.relkind = 'S' AND c.relname ~ '^pg_temp_' ",
     },
     "EXTERNAL TABLE": None,
+    "SHARED TABLE": None,
 }

redshift_connector/core.py

@@ -364,6 +364,7 @@ def __init__(
         application_name: typing.Optional[str] = None,
         replication: typing.Optional[str] = None,
         client_protocol_version: int = DEFAULT_PROTOCOL_VERSION,
+        database_metadata_current_db_only: bool = True,
     ):
 
         self.merge_socket_read = False
@@ -384,6 +385,8 @@ def __init__(
         self.max_prepared_statements: int = int(max_prepared_statements)
         self._run_cursor: Cursor = Cursor(self, paramstyle="named")
         self._client_protocol_version: int = client_protocol_version
+        self._database = database
+        self._database_metadata_current_db_only: bool = database_metadata_current_db_only
 
         if user is None:
             raise InterfaceError("The 'user' connection parameter cannot be None")
@@ -565,6 +568,18 @@ def __init__(
 
         self.in_transaction = False
 
+    @property
+    def _is_multi_databases_catalog_enable_in_server(self: "Connection") -> bool:
+        if (b"datashare_enabled", str("on").encode()) in self.parameter_statuses:
+            return True
+        else:
+            # if we don't receive this param from the server, we do not support
+            return False
+
+    @property
+    def is_single_database_metadata(self):
+        return self._database_metadata_current_db_only or not self._is_multi_databases_catalog_enable_in_server
+
     def handle_ERROR_RESPONSE(self: "Connection", data, ps):
         msg: typing.Dict[str, str] = dict(
             (s[:1].decode(_client_encoding), s[1:].decode(_client_encoding)) for s in data.split(NULL_BYTE) if s != b""