MINOR: Redshift - better log for view with no schema binding (#20005)

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

@@ -84,6 +84,7 @@ from metadata.ingestion.source.database.redshift.utils import (
     _get_pg_column_info,
     _get_schema_column_info,
     get_columns,
+    get_redshift_columns,
     get_table_comment,
     get_view_definition,
 )
@@ -109,18 +110,15 @@ STANDARD_TABLE_TYPES = {
     "v": TableType.View,
 }
 
-
-RedshiftDialectMixin._get_column_info = (  # pylint: disable=protected-access
-    _get_column_info
-)
-RedshiftDialectMixin._get_schema_column_info = (  # pylint: disable=protected-access
-    _get_schema_column_info
-)
+# pylint: disable=protected-access
+RedshiftDialectMixin._get_column_info = _get_column_info
+RedshiftDialectMixin._get_schema_column_info = _get_schema_column_info
 RedshiftDialectMixin.get_columns = get_columns
-PGDialect._get_column_info = _get_pg_column_info  # pylint: disable=protected-access
+PGDialect._get_column_info = _get_pg_column_info
 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_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

@@ -34,6 +34,7 @@ from metadata.ingestion.source.database.redshift.queries import (
     REDSHIFT_TABLE_COMMENTS,
 )
 from metadata.utils.execution_time_tracker import calculate_execution_time
+from metadata.utils.logger import ingestion_logger
 from metadata.utils.sqlalchemy_utils import get_table_comment_wrapper
 
 sa_version = Version(sa.__version__)
@@ -45,6 +46,9 @@ ischema_names.update({"binary varying": sqltypes.VARBINARY})
 ischema_names.update(REDSHIFT_ISCHEMA_NAMES)
 
 
+logger = ingestion_logger()
+
+
 # pylint: disable=protected-access
 @calculate_execution_time()
 @reflection.cache
@@ -414,3 +418,25 @@ def get_view_definition(self, connection, view_name, schema=None, **kw):
             f"CREATE VIEW {view.schema}.{view.relname} AS {view_definition}"
         )
     return view_definition
+
+
+def get_redshift_columns(self, connection, table_name, schema=None, **kw):
+    try:
+        info_cache = kw.get("info_cache")
+        all_schema_columns = self._get_schema_column_info(
+            connection,
+            schema,
+            info_cache=info_cache,
+        )
+        key = RelationKey(table_name, schema, connection)
+        if key not in all_schema_columns.keys():
+            key = key.unquoted()
+        return all_schema_columns[key]
+    except KeyError:
+        schema_name = schema or "public"
+        logger.error(
+            f"Fetching columns for table {schema_name}.{table_name} failed,"
+            " if this is a view with no schema binding, please make sure user has"
+            f' USAGE privilege on schema "{schema_name}"'
+        )
+        return []