MINOR: Fix Redshift View Def regex_replace Error

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

@@ -83,6 +83,7 @@
     _get_schema_column_info,
     get_columns,
     get_table_comment,
+    get_view_definition,
 )
 from metadata.ingestion.source.database.stored_procedures_mixin import (
     QueryByProcedure,
@@ -122,6 +123,7 @@
 PGDialect._get_column_info = _get_pg_column_info  # pylint: disable=protected-access
 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_all_relation_info = (  # pylint: disable=protected-access
     _get_all_relation_info
 )

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

@@ -236,30 +236,6 @@
 # hence we are appending "create view <schema>.<table> as " to select query
 # to generate the column level lineage
 REDSHIFT_GET_ALL_RELATIONS = """
-  WITH view_defs AS (
-      SELECT
-          c.oid,
-          pg_catalog.pg_get_viewdef(c.oid, true) AS view_definition,
-          n.nspname,
-          c.relname
-      FROM
-          pg_catalog.pg_class c
-          LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
-      WHERE
-          c.relkind = 'v'
-    ),
-    adjusted_view_defs AS (
-        SELECT
-            oid,
-            CASE
-                WHEN view_definition LIKE '%WITH NO SCHEMA BINDING%' THEN
-                  REGEXP_REPLACE(view_definition, 'create view [^ ]+ as (.*WITH NO SCHEMA BINDING;?)', '\\1')
-                ELSE
-                  'CREATE VIEW ' || nspname || '.' || relname || ' AS ' || view_definition
-            END AS view_definition
-        FROM
-            view_defs
-    )
     SELECT
         c.relkind,
         n.oid as "schema_oid",
@@ -271,13 +247,12 @@
         AS "diststyle",
         c.relowner AS "owner_id",
         u.usename AS "owner_name",
-        avd.view_definition 
+        CAST(pg_catalog.pg_get_viewdef(c.oid, true) AS TEXT)
         AS "view_definition",
         pg_catalog.array_to_string(c.relacl, '\n') AS "privileges"
     FROM pg_catalog.pg_class c
             LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
             JOIN pg_catalog.pg_user u ON u.usesysid = c.relowner
-            LEFT JOIN adjusted_view_defs avd ON avd.oid = c.oid
     WHERE c.relkind IN ('r', 'v', 'm', 'S', 'f')
         AND n.nspname !~ '^pg_' {schema_clause} {table_clause}
     UNION

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

@@ -395,3 +395,22 @@ def _get_all_relation_info(self, connection, **kw):  # pylint: disable=unused-ar
         key = RelationKey(rel.relname, rel.schema, connection)
         relations[key] = rel
     return relations
+
+
+@reflection.cache
+def get_view_definition(self, connection, view_name, schema=None, **kw):
+    """Return view definition.
+    Given a :class:`.Connection`, a string `view_name`,
+    and an optional string `schema`, return the view definition.
+
+    Overrides interface
+    :meth:`~sqlalchemy.engine.interfaces.Dialect.get_view_definition`.
+    """
+    view = self._get_redshift_relation(connection, view_name, schema, **kw)
+    pattern = re.compile("WITH NO SCHEMA BINDING", re.IGNORECASE)
+    view_definition = str(sa.text(pattern.sub("", view.view_definition)))
+    if not view_definition.startswith("create"):
+        view_definition = (
+            f"CREATE VIEW {view.schema}.{view.relname} AS {view_definition}"
+        )
+    return view_definition