postgres & redshift ddl and e2e fixes

ingestion/src/metadata/ingestion/source/database/redshift/metadata.py

@@ -95,6 +95,7 @@
     get_multi_columns,
     get_redshift_columns,
     get_table_comment,
+    get_temp_table_names,
     get_view_definition,
 )
 from metadata.utils import fqn
@@ -132,6 +133,7 @@
 RedshiftDialect.get_all_table_comments = get_all_table_comments
 RedshiftDialect.get_table_comment = get_table_comment
 RedshiftDialect.get_view_definition = get_view_definition
+RedshiftDialect.get_temp_table_names = get_temp_table_names
 RedshiftDialect._get_redshift_columns = get_redshift_columns
 RedshiftDialect._get_all_relation_info = (  # pylint: disable=protected-access
     _get_all_relation_info

ingestion/src/metadata/ingestion/source/database/redshift/utils.py

@@ -65,7 +65,7 @@
     self._has_native_hstore = False
 
 
 def _load_domains(self, connection, **kw):
     """
     Override to return empty dict since Redshift does not support user-created
     domains and pg_catalog.pg_collation does not exist in Redshift, causing a
@@ -74,6 +74,16 @@
     return {}
 
 
+def get_temp_table_names(self, connection, schema=None, **kw):
+    """
+    Override PGDialect's get_temp_table_names to avoid querying
+    pg_catalog.pg_class.relpersistence which does not exist in Redshift,
+    causing a ProgrammingError that aborts the transaction and breaks all
+    subsequent queries.
+    """
+    return []
+
+
 def get_multi_columns(self, connection, **kw):
     """
     Override PGDialect's get_multi_columns to avoid querying

ingestion/src/metadata/utils/sqlalchemy_utils.py

@@ -18,6 +18,7 @@
 
 from sqlalchemy import text
 from sqlalchemy.engine import Engine, reflection
+from sqlalchemy.exc import ProgrammingError
 from sqlalchemy.schema import CreateTable, MetaData
 
 from metadata.utils.logger import ingestion_logger
@@ -158,6 +159,14 @@ def get_all_table_ddls(
     except Exception as exc:
         logger.debug(traceback.format_exc())
         logger.debug(f"Failed to get table ddls for {schema_name}: {exc}")
+        # Roll back the aborted transaction so the connection remains usable
+        # for subsequent queries (e.g. get_table_comment). Without this,
+        # psycopg2 raises InFailedSqlTransaction on every query that follows.
+        if isinstance(exc, ProgrammingError):
+            try:
+                connection.rollback()
+            except Exception:
+                pass
 
 
 def get_table_ddl_wrapper(

ingestion/tests/cli_e2e/common_e2e_sqa_mixins.py

@@ -12,28 +12,27 @@
 """
 Add Common E2E Sqlalchemy Mixins
 """
-from sqlalchemy import text
 
 
 class SQACommonMethods:
     def create_table_and_view(self) -> None:
         with self.engine.begin() as connection:
-            connection.execute(text(self.create_table_query))
+            connection.exec_driver_sql(self.create_table_query)
             for insert_query in self.insert_data_queries:
-                connection.execute(text(insert_query))
-            connection.execute(text(self.create_view_query))
+                connection.exec_driver_sql(insert_query)
+            connection.exec_driver_sql(self.create_view_query)
 
     def delete_table_and_view(self) -> None:
         with self.engine.begin() as connection:
-            connection.execute(text(self.drop_view_query))
-            connection.execute(text(self.drop_table_query))
+            connection.exec_driver_sql(self.drop_view_query)
+            connection.exec_driver_sql(self.drop_table_query)
 
     def run_update_queries(self) -> None:
         with self.engine.begin() as connection:
             for update_query in self.update_queries():
-                connection.execute(text(update_query))
+                connection.exec_driver_sql(update_query)
 
     def run_delete_queries(self) -> None:
         with self.engine.begin() as connection:
             for drop_query in self.delete_queries():
-                connection.execute(text(drop_query))
+                connection.exec_driver_sql(drop_query)