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

@@ -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",

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)}) "