vdk-trino:

What:
Optimize template by splitting staging table population in two step to reduce memory footprint.

Why:
For bigger tables the template struggles to create the staging table with unions between the view data and current table data.

Signed-off-by: Stefan Buldeev sbuldeev@vmware.com

Author: Stefan Buldeev

projects/vdk-plugins/vdk-trino/src/vdk/plugin/trino/templates/load/dimension/scd1_upsert/02-requisite-sql-scripts/02-create-table-and-insert-data.sql

@@ -1,12 +1 @@
-CREATE TABLE "{target_schema_staging}"."{target_table_staging}" AS
-(
-SELECT t.*
-FROM "{target_schema}"."{target_table}" AS t
-LEFT JOIN "{source_schema}"."{source_view}" AS s ON s."{id_column}" = t."{id_column}"
-WHERE s."{id_column}" IS NULL
-)
-UNION ALL
-(
-SELECT *
-FROM "{source_schema}"."{source_view}"
-)
+CREATE TABLE "{target_schema_staging}"."{target_table_staging}" AS (SELECT * FROM "{source_schema}"."{source_view}")

projects/vdk-plugins/vdk-trino/src/vdk/plugin/trino/templates/load/dimension/scd1_upsert/03-handle-quality-checks_and_move_data.py

@@ -58,23 +58,30 @@ def run(job_input: IJobInput):
     )
     job_input.execute_query(drop_table)
 
-    # create staging table and insert data
+    # create staging table and insert the source (new) data
     create_table_and_insert_data_query = CommonUtilities.get_file_content(
         SQL_FILES_FOLDER, "02-create-table-and-insert-data.sql"
     )
-    create_staging_table_and_insert_data = create_table_and_insert_data_query.format(
-        target_schema=target_schema,
-        target_table=target_table,
+    create_stg_tbl_and_insert_new_data = create_table_and_insert_data_query.format(
         source_schema=source_schema,
         source_view=source_view,
         target_schema_staging=staging_schema,
         target_table_staging=staging_table,
-        id_column=id_column,
     )
-    job_input.execute_query(create_staging_table_and_insert_data)
+    job_input.execute_query(create_stg_tbl_and_insert_new_data)
 
     staging_table_full_name = f"{staging_schema}.{staging_table}"
 
+    # append the target-only (old) data to the staging table
+    append_old_to_stg = f'''
+        INSERT INTO {staging_table_full_name}
+        SELECT t.*
+        FROM {target_schema}.{target_table} AS t
+        LEFT JOIN {staging_table_full_name} AS s ON s."{id_column}" = t."{id_column}"
+        WHERE s."{id_column}" IS NULL
+    '''
+    job_input.execute_query(append_old_to_stg)
+
     # copy the data if there's no quality check configure or if it passes
     if not check or check(staging_table_full_name):
         copy_staging_table_to_target_table(