handle large sql text

datahub-project · Jan 18, 2024 · 14cf5e7 · treff7es · Jan 18, 2024 · treff7es
1 parent 19bfa2e
commit 14cf5e7
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 6 deletions.
diff --git a/metadata-ingestion/src/datahub/ingestion/source/redshift/query.py b/metadata-ingestion/src/datahub/ingestion/source/redshift/query.py
@@ -384,7 +384,8 @@ def list_insert_create_queries_sql(
                     target_schema,
                     target_table,
                     username,
-                    querytxt as ddl
+                    query as query_id,
+                    LISTAGG(CASE WHEN LEN(RTRIM(querytxt)) = 0 THEN querytxt ELSE RTRIM(querytxt) END) WITHIN GROUP (ORDER BY sequence) as ddl
                 from
                         (
                     select
@@ -393,27 +394,30 @@ def list_insert_create_queries_sql(
                         sti.table as target_table,
                         sti.database as cluster,
                         usename as username,
-                        querytxt,
+                        text as querytxt,
+                        sq.query,
+                        sequence,
                         si.starttime as starttime
                     from
                         stl_insert as si
                     join SVV_TABLE_INFO sti on
                         sti.table_id = tbl
                     left join svl_user_info sui on
                         si.userid = sui.usesysid
-                    left join stl_query sq on
+                    left join STL_QUERYTEXT sq on
                         si.query = sq.query
                     left join stl_load_commits slc on
                         slc.query = si.query
                     where
                         sui.usename <> 'rdsdb'
-                        and sq.aborted = 0
                         and slc.query IS NULL
                         and cluster = '{db_name}'
                         and si.starttime >= '{start_time}'
                         and si.starttime < '{end_time}'
+                        and cluster = 'dev'
                     ) as target_tables
-                    order by cluster, target_schema, target_table, starttime asc
+                    group by cluster, query_id, target_schema, target_table, username, starttime
+                    order by cluster, query_id, target_schema, target_table, starttime asc
                 """.format(
             # We need the original database name for filtering
             db_name=db_name,

diff --git a/metadata-ingestion/src/datahub/ingestion/source/redshift/redshift_schema.py b/metadata-ingestion/src/datahub/ingestion/source/redshift/redshift_schema.py
@@ -367,7 +367,9 @@ def get_lineage_rows(
                     target_table=row[field_names.index("target_table")]
                     if "target_table" in field_names
                     else None,
-                    ddl=row[field_names.index("ddl")] if "ddl" in field_names else None,
+                    ddl=row[field_names.index("ddl")].replace("\\n", "")
+                    if "ddl" in field_names
+                    else None,
                     filename=row[field_names.index("filename")]
                     if "filename" in field_names
                     else None,