From 544bc9a6d905e34729a1d83c05112b09f4aae182 Mon Sep 17 00:00:00 2001
From: Ryan Rymarczyk <rymarczy@gmail.com>
Date: Thu, 12 Dec 2024 12:28:51 -0500
Subject: [PATCH] FEAT: ODS Drop Primary Keys from NEW Tables (#34)

This change alters the way new tables are created. A standard Postgres primary key is no longer created for each table. Instead, an Index using primary key columns is created, which can be nullable.

This change should have no impact on existing tables and will only impact newly added tables. With this change we can now import the EDW.PAYMENT_SUMMARY table.
---
 src/cubic_loader/qlik/ods_tables.py |  1 +
 src/cubic_loader/qlik/rds_utils.py  | 15 +++++++++++----
 2 files changed, 12 insertions(+), 4 deletions(-)

diff --git a/src/cubic_loader/qlik/ods_tables.py b/src/cubic_loader/qlik/ods_tables.py
index 316b6e8..5f039bd 100644
--- a/src/cubic_loader/qlik/ods_tables.py
+++ b/src/cubic_loader/qlik/ods_tables.py
@@ -24,6 +24,7 @@
     "EDW.FARE_REVENUE_REPORT_SCHEDULE",  # addendum support
     # FMIS
     "EDW.FNP_GENERAL_JRNL_ACCOUNT_ENTRY",
+    "EDW.PAYMENT_SUMMARY",
     # KPI
     "EDW.DEVICE_LAST_STATE",
     "EDW.DEVICE_EVENT",
diff --git a/src/cubic_loader/qlik/rds_utils.py b/src/cubic_loader/qlik/rds_utils.py
index bcfcfae..309b171 100644
--- a/src/cubic_loader/qlik/rds_utils.py
+++ b/src/cubic_loader/qlik/rds_utils.py
@@ -72,8 +72,16 @@ def create_tables_from_schema(schema: List[DFMSchemaFields], schema_and_table: s
     assert len(dfm_keys) > 0
 
     # Create FACT Table
-    fact_columns = dfm_columns + [f"PRIMARY KEY ({','.join(dfm_keys)})"]
-    ops.append(f"CREATE TABLE IF NOT EXISTS {schema_and_table} ({",".join(fact_columns)});")
+    # FACT table is created without a primary key
+    # Tables coming from CUBIC ODS system are Oracle based which allows NULL values in Primary Key columns
+    # Postgres does not allow NULL in Primary Key columns, instead a standard INDEX on the Key columns is created
+    ops.append(f"CREATE TABLE IF NOT EXISTS {schema_and_table} ({",".join(dfm_columns)});")
+
+    # FACT Table Index on Primary Key columns
+    ops.append(
+        f"CREATE INDEX IF NOT EXISTS {schema_and_table.replace('.','_')}_fact_pk_idx on {schema_and_table} "
+        f"({','.join(dfm_keys)});"
+    )
 
     # Create HISTORY Table
     # partitioned by header__timestamp
@@ -83,8 +91,7 @@ def create_tables_from_schema(schema: List[DFMSchemaFields], schema_and_table: s
         ("header__change_seq", "CHANGE_SEQ"),
     )
     header_cols: List[str] = [f"{col[0]} {qlik_type_to_pg(col[1], 0)}" for col in header_fields]
-    history_keys = ["header__timestamp"] + dfm_keys + ["header__change_oper", "header__change_seq"]
-    history_columns = header_cols + dfm_columns + [f"PRIMARY KEY ({','.join(history_keys)})"]
+    history_columns = header_cols + dfm_columns
     ops.append(
         (
             f"CREATE TABLE IF NOT EXISTS {schema_and_table}_history ({",".join(history_columns)}) "