vdk-trino: Optimize SCD1 Upsert template (#3454)

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

---------

Signed-off-by: Stefan Buldeev sbuldeev@vmware.com
Signed-off-by: Dako Dakov <ddakov@vmware.com>
Co-authored-by: Stefan Buldeev <stefan.buldeev@broadcom.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: ddakov <ddakov@vmware.com>

Author: 4 people

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(

projects/vdk-plugins/vdk-trino/tests/test_vdk_templates.py

@@ -672,7 +672,7 @@ def __scd_upsert_execute(
                     "run",
                     get_test_job_path(
                         pathlib.Path(os.path.dirname(os.path.abspath(__file__))),
-                        "load_dimension_scd_upsert_template_job",
+                        "load_dimension_scd1_upsert_template_job",
                     ),
                     "--arguments",
                     json.dumps(
@@ -696,7 +696,7 @@ def __scd_upsert_execute(
                     "run",
                     get_test_job_path(
                         pathlib.Path(os.path.dirname(os.path.abspath(__file__))),
-                        "load_dimension_scd_upsert_template_job",
+                        "load_dimension_scd1_upsert_template_job",
                     ),
                     "--arguments",
                     json.dumps(