Testing changes involve populating CSV claims data which will be trimmed out before it is populated onto associated claims tables, as that is how current filtering works, and then the data is re seeded to test against the pruning logic.

Removal of datetime processing to uncouple the logic from how phase 1 was doing it,

renaming things for better conventions

Author: patrickgmayday

apps/bfd-db-migrator-ng/migrations/V1_38_0__add_non_latest_claim_prune_idx.sql

@@ -1,3 +1,7 @@
-CREATE INDEX ON idr.claim_professional_ss(clm_idr_ld_dt, clm_uniq_id) WHERE clm_ltst_clm_ind = 'N';
+CREATE INDEX ON idr.claim_professional_ss(clm_ltst_clm_ind) WHERE clm_ltst_clm_ind = 'N';
 
-CREATE INDEX ON idr.claim_institutional_ss(clm_idr_ld_dt, clm_uniq_id) WHERE clm_ltst_clm_ind = 'N';
+CREATE INDEX ON idr.claim_institutional_ss(clm_ltst_clm_ind) WHERE clm_ltst_clm_ind = 'N';
+
+CREATE INDEX ON idr.claim_professional_nch(clm_ltst_clm_ind) WHERE clm_ltst_clm_ind = 'N';
+
+CREATE INDEX ON idr.claim_institutional_nch(clm_ltst_clm_ind) WHERE clm_ltst_clm_ind = 'N';

apps/bfd-pipeline-idr/constants.py

@@ -9,9 +9,11 @@
 BENEFICIARY_TABLE = "idr.beneficiary"
 CLAIM_RX_TABLE = "idr.claim_rx"
 CLAIM_PROFESSIONAL_NCH_TABLE = "idr.claim_professional_nch"
+CLAIM_PROFESSIONAL_ITEM_NCH_TABLE = "idr.claim_item_professional_nch"
 CLAIM_PROFESSIONAL_SS_TABLE = "idr.claim_professional_ss"
 CLAIM_PROFESSIONAL_ITEM_SS_TABLE = "idr.claim_item_professional_ss"
 CLAIM_INSTITUTIONAL_NCH_TABLE = "idr.claim_institutional_nch"
+CLAIM_INSTITUTIONAL_ITEM_NCH_TABLE = "idr.claim_item_institutional_nch"
 CLAIM_INSTITUTIONAL_SS_TABLE = "idr.claim_institutional_ss"
 CLAIM_INSTITUTIONAL_ITEM_SS_TABLE = "idr.claim_item_institutional_ss"
 

apps/bfd-pipeline-idr/model/base_model.py

@@ -872,22 +872,11 @@ def stale_phase_1_claims_query(
     )
 
 
-def non_latest_non_part_d_claims_query(
-    claim_table: str,
-    cutoff_date: datetime,
-) -> tuple[str, tuple[datetime]]:
-    return (
-        f"""
-            WITH claims AS (
-                SELECT clm.clm_uniq_id
-                FROM {claim_table} clm
-                WHERE clm.clm_ltst_clm_ind = 'N'
-                AND clm.clm_idr_ld_dt < %s
-                ORDER BY clm.clm_idr_ld_dt, clm.clm_uniq_id
-            )
-            SELECT clm.clm_uniq_id
-            FROM claims clm
-            LIMIT {PHASE_1_PRUNE_BATCH_LIMIT}
-        """,
-        (cutoff_date,),
-    )
+def stale_non_part_d_claims_query(claim_table: str) -> str:
+    return f"""
+        SELECT clm.clm_uniq_id
+        FROM {claim_table} clm
+        WHERE clm.clm_ltst_clm_ind = 'N'
+        ORDER BY clm.clm_uniq_id
+        LIMIT {PHASE_1_PRUNE_BATCH_LIMIT}
+    """

apps/bfd-pipeline-idr/pipeline_stages.py

@@ -41,8 +41,8 @@
     prune_bene_lis_cmbnd,
     prune_bene_ma_part_d,
     prune_bene_ma_part_d_rx,
-    prune_non_latest_non_part_d_ss_claims,
     prune_phase_1_ss_claims,
+    prune_stale_non_part_d_claims,
 )
 from settings import enable_prior_auth_ingestion
 
@@ -176,11 +176,12 @@ def _stage5_prune_obsolete_rows(self) -> Stage[bool]:
                 self.load_mode,
                 self.start_time,
             )
+
+        for model in self._filter_tables(CLAIM_TABLES):
             yield functools.partial(
-                prune_non_latest_non_part_d_ss_claims,
+                prune_stale_non_part_d_claims,
                 model,
                 self.load_mode,
-                self.start_time,
             )
 
     def _filter_tables(self, tables: list[type[IdrBaseModel]]) -> list[type[IdrBaseModel]]:

apps/bfd-pipeline-idr/pipeline_utils.py

@@ -9,9 +9,13 @@
 
 from batch_worker import LoadingBatchWorkerClient
 from constants import (
+    CLAIM_INSTITUTIONAL_ITEM_NCH_TABLE,
     CLAIM_INSTITUTIONAL_ITEM_SS_TABLE,
+    CLAIM_INSTITUTIONAL_NCH_TABLE,
     CLAIM_INSTITUTIONAL_SS_TABLE,
+    CLAIM_PROFESSIONAL_ITEM_NCH_TABLE,
     CLAIM_PROFESSIONAL_ITEM_SS_TABLE,
+    CLAIM_PROFESSIONAL_NCH_TABLE,
     CLAIM_PROFESSIONAL_SS_TABLE,
     DEFAULT_MAX_DATE,
     DEFAULT_PARTITION,
@@ -23,7 +27,7 @@
 from model.base_model import (
     LoadMode,
     T,
-    non_latest_non_part_d_claims_query,
+    stale_non_part_d_claims_query,
     stale_phase_1_claims_query,
 )
 from model.idr_beneficiary_low_income_subsidy_cmbnd import IdrBeneficiaryLowIncomeSubsidyCmbnd
@@ -41,6 +45,15 @@
 }
 
 
+# Keep this mapping explicit so stale non-Part-D pruning remains independent of Phase 1 SS pruning.
+_NON_PART_D_CLAIM_ITEM_TABLES = {
+    CLAIM_INSTITUTIONAL_NCH_TABLE: CLAIM_INSTITUTIONAL_ITEM_NCH_TABLE,
+    CLAIM_INSTITUTIONAL_SS_TABLE: CLAIM_INSTITUTIONAL_ITEM_SS_TABLE,
+    CLAIM_PROFESSIONAL_NCH_TABLE: CLAIM_PROFESSIONAL_ITEM_NCH_TABLE,
+    CLAIM_PROFESSIONAL_SS_TABLE: CLAIM_PROFESSIONAL_ITEM_SS_TABLE,
+}
+
+
 def get_progress(
     load_mode: LoadMode,
     source: Source,
@@ -169,20 +182,18 @@ def prune_phase_1_ss_claims(
     return True
 
 
-def prune_non_latest_non_part_d_ss_claims(
+def prune_stale_non_part_d_claims(
     cls: type[T],
     load_mode: LoadMode,
-    job_start: datetime,
 ) -> bool:
     claim_table = cls.table()
-    item_table = _SHARED_SYSTEM_CLAIM_ITEM_TABLES.get(claim_table)
+    item_table = _NON_PART_D_CLAIM_ITEM_TABLES.get(claim_table)
     if item_table is None:
         return True
 
-    prune_cutoff_date = job_start - timedelta(days=PHASE_1_CUTOFF)
-    logger.info("pruning non-latest non-Part-D ss claims older than {}", prune_cutoff_date)
+    logger.info("pruning stale non-Part-D claims")
 
-    prune_query, params = non_latest_non_part_d_claims_query(claim_table, prune_cutoff_date)
+    prune_query = stale_non_part_d_claims_query(claim_table)
 
     total_row_counts = {
         item_table: 0,
@@ -195,8 +206,7 @@ def prune_non_latest_non_part_d_ss_claims(
             with conn.transaction():
                 for target_table in [item_table, claim_table]:
                     res = conn.execute(
-                        f"""DELETE FROM {target_table} WHERE clm_uniq_id IN ({prune_query})""",  # type: ignore
-                        params,
+                        f"""DELETE FROM {target_table} WHERE clm_uniq_id IN ({prune_query})"""  # type: ignore
                     )
 
                     total_row_counts[target_table] += res.rowcount

apps/bfd-pipeline-idr/test_pipeline.py

@@ -93,9 +93,69 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
         rows = cur.fetchmany(1)
         assert rows[0]["mbi_num"] == "1OX4Y88RV68"
 
-    # Seed a parent claim so the prune job has a row to delete.
-    # CSV covers item pruning because non-latest parents do not load.
+    # Seed stale non-Part-D parent claims so the prune job has rows to delete.
+    # CSVs cover item pruning because stale non-Part-D parents do not load.
     if load_type == LoadType.INCREMENTAL:
+        claim_table = sql.Identifier("idr", "claim_institutional_nch")
+        cur = conn.execute(
+            "select * from idr.claim_institutional_nch where clm_uniq_id = 113370100080"
+        )
+        assert cur.rowcount == 1
+        row = cur.fetchone()
+        assert row is not None
+
+        stale_institutional_nch_parent_claim = dict(row)
+        stale_institutional_nch_parent_claim["clm_uniq_id"] = 999999434801
+        stale_institutional_nch_parent_claim["clm_ltst_clm_ind"] = "N"
+        stale_institutional_nch_parent_claim["clm_type_cd"] = 40
+
+        columns = sql.SQL(", ").join(
+            sql.Identifier(k) for k in stale_institutional_nch_parent_claim
+        )
+        values = sql.SQL(", ").join(stale_institutional_nch_parent_claim.values())
+
+        conn.execute(
+            t"""
+            INSERT INTO {claim_table:i}
+            (
+                {columns:q}
+            )
+            VALUES
+            (
+                {values:q}
+            )
+            """
+        )
+
+        claim_table = sql.Identifier("idr", "claim_professional_nch")
+        cur = conn.execute(
+            "select * from idr.claim_professional_nch where clm_uniq_id = 119855147698"
+        )
+        assert cur.rowcount == 1
+        row = cur.fetchone()
+        assert row is not None
+
+        stale_professional_nch_parent_claim = dict(row)
+        stale_professional_nch_parent_claim["clm_uniq_id"] = 999999434802
+        stale_professional_nch_parent_claim["clm_ltst_clm_ind"] = "N"
+        stale_professional_nch_parent_claim["clm_type_cd"] = 81
+
+        columns = sql.SQL(", ").join(sql.Identifier(k) for k in stale_professional_nch_parent_claim)
+        values = sql.SQL(", ").join(stale_professional_nch_parent_claim.values())
+
+        conn.execute(
+            t"""
+            INSERT INTO {claim_table:i}
+            (
+                {columns:q}
+            )
+            VALUES
+            (
+                {values:q}
+            )
+            """
+        )
+
         claim_table = sql.Identifier("idr", "claim_institutional_ss")
         cur = conn.execute(
             "select * from idr.claim_institutional_ss where clm_uniq_id = 123359318723"
@@ -104,26 +164,51 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
         row = cur.fetchone()
         assert row is not None
 
-        non_latest_parent_claim = dict(row)
-        non_latest_parent_claim["clm_uniq_id"] = 999999434800
-        non_latest_parent_claim["clm_ltst_clm_ind"] = "N"
-        non_latest_parent_claim["clm_type_cd"] = 2081
-        # Make the seeded claim old enough to be picked up by the prune job.
-        non_latest_parent_claim["clm_idr_ld_dt"] = datetime(2019, 6, 13, tzinfo=UTC).date()
+        stale_institutional_ss_parent_claim = dict(row)
+        stale_institutional_ss_parent_claim["clm_uniq_id"] = 999999434800
+        stale_institutional_ss_parent_claim["clm_ltst_clm_ind"] = "N"
+        stale_institutional_ss_parent_claim["clm_type_cd"] = 2081
+
+        columns = sql.SQL(", ").join(sql.Identifier(k) for k in stale_institutional_ss_parent_claim)
+        values = sql.SQL(", ").join(stale_institutional_ss_parent_claim.values())
+
+        conn.execute(
+            t"""
+            INSERT INTO {claim_table:i}
+            (
+                {columns:q}
+            )
+            VALUES
+            (
+                {values:q}
+            )
+            """
+        )
+
+        claim_table = sql.Identifier("idr", "claim_professional_ss")
+        cur = conn.execute(
+            "select * from idr.claim_professional_ss where clm_uniq_id = 4991490559710"
+        )
+        assert cur.rowcount == 1
+        row = cur.fetchone()
+        assert row is not None
 
-        if "bfd_row_id" in non_latest_parent_claim:
-            non_latest_parent_claim["bfd_row_id"] = 999999434800
+        stale_professional_ss_parent_claim = dict(row)
+        stale_professional_ss_parent_claim["clm_uniq_id"] = 999999434803
+        stale_professional_ss_parent_claim["clm_ltst_clm_ind"] = "N"
+        stale_professional_ss_parent_claim["clm_type_cd"] = 2800
 
-        if "clm_dt_sgntr_sk" in non_latest_parent_claim:
-            non_latest_parent_claim["clm_dt_sgntr_sk"] = 999999434800
+        columns = sql.SQL(", ").join(sql.Identifier(k) for k in stale_professional_ss_parent_claim)
+        values = sql.SQL(", ").join(stale_professional_ss_parent_claim.values())
 
         conn.execute(
             t"""
-            INSERT INTO {claim_table:i} (
-                {sql.SQL(", ").join(sql.Identifier(k) for k in non_latest_parent_claim):q}
+            INSERT INTO {claim_table:i}
+            (
+                {columns:q}
             )
             VALUES (
-                {sql.SQL(", ").join(non_latest_parent_claim.values()):q}
+                {values:q}
             )
             """
         )
@@ -251,10 +336,19 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
     rows = cur.fetchmany(1)
     assert rows[0]["clm_uniq_id"] == 113370100080
 
-    # Non-latest non-Part-D SS parent claims do not remain in the final claim table
+    # Stale non-Part-D parent claims do not remain in the final claim tables
+    cur = conn.execute("select * from idr.claim_institutional_nch where clm_uniq_id = 999999434801")
+    assert cur.rowcount == 0
+
+    cur = conn.execute("select * from idr.claim_professional_nch where clm_uniq_id = 999999434802")
+    assert cur.rowcount == 0
+
     cur = conn.execute("select * from idr.claim_institutional_ss where clm_uniq_id = 999999434800")
     assert cur.rowcount == 0
 
+    cur = conn.execute("select * from idr.claim_professional_ss where clm_uniq_id = 999999434803")
+    assert cur.rowcount == 0
+
     cur = conn.execute("select * from idr.claim_professional_nch order by clm_uniq_id")
     assert cur.rowcount == 51
     rows = cur.fetchmany(1)
@@ -271,28 +365,64 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
     assert rows[0]["clm_uniq_id"] == 166776396279
 
     cur = conn.execute("select * from idr.claim_item_institutional_nch order by clm_uniq_id")
-    assert cur.rowcount == 795
+    if load_type == LoadType.INITIAL:
+        assert cur.rowcount == 796
+    elif load_type == LoadType.INCREMENTAL:
+        assert cur.rowcount == 795
     rows = cur.fetchmany(1)
     assert rows[0]["clm_uniq_id"] == 113370100080
 
-    # Items for non-latest non-Part-D SS claims are pruned on incremental loads
+    # Items for stale non-Part-D claims are pruned on incremental loads
+    cur = conn.execute(
+        "select * from idr.claim_item_institutional_nch where clm_uniq_id = 999999434801"
+    )
+    if load_type == LoadType.INITIAL:
+        assert cur.rowcount == 1
+    elif load_type == LoadType.INCREMENTAL:
+        assert cur.rowcount == 0
+
+    cur = conn.execute(
+        "select * from idr.claim_item_professional_nch where clm_uniq_id = 999999434802"
+    )
+    if load_type == LoadType.INITIAL:
+        assert cur.rowcount == 1
+    elif load_type == LoadType.INCREMENTAL:
+        assert cur.rowcount == 0
+
     cur = conn.execute(
         "select * from idr.claim_item_institutional_ss where clm_uniq_id = 999999434800"
     )
     if load_type == LoadType.INITIAL:
         assert cur.rowcount == 1
-    else:
+    elif load_type == LoadType.INCREMENTAL:
+        assert cur.rowcount == 0
+
+    cur = conn.execute(
+        "select * from idr.claim_item_professional_ss where clm_uniq_id = 999999434803"
+    )
+    if load_type == LoadType.INITIAL:
+        assert cur.rowcount == 1
+    elif load_type == LoadType.INCREMENTAL:
         assert cur.rowcount == 0
 
     cur = conn.execute("select * from idr.claim_item_professional_nch order by clm_uniq_id")
-    assert cur.rowcount == 442
+    if load_type == LoadType.INITIAL:
+        assert cur.rowcount == 443
+    elif load_type == LoadType.INCREMENTAL:
+        assert cur.rowcount == 442
     rows = cur.fetchmany(1)
     assert rows[0]["clm_uniq_id"] == 119855147698
 
     cur = conn.execute("select * from idr.claim_item_professional_ss order by clm_uniq_id")
+    if load_type == LoadType.INITIAL:
+        assert cur.rowcount == 2
+    elif load_type == LoadType.INCREMENTAL:
+        assert cur.rowcount == 1
+
+    cur = conn.execute(
+        "select * from idr.claim_item_professional_ss where clm_uniq_id = 4991490559710"
+    )
     assert cur.rowcount == 1
-    rows = cur.fetchmany(1)
-    assert rows[0]["clm_uniq_id"] == 4991490559710
 
     conn.commit()
 

apps/bfd-pipeline-idr/test_samples1/SYNTHETIC_CLM.csv

@@ -146,3 +146,6 @@ CLM_DT_SGNTR_SK,CLM_UNIQ_ID,CLM_RLT_COND_SGNTR_SK,CLM_TYPE_CD,CLM_SRC_ID,CLM_FRO
 261063166925,8244064276500,,2041,21000,2018-11-20,2018-02-11,07890767122076DPC,,1,2025-08-15,2020-03-26,9999-12-31,55419,313081892,1,1,,,,235338.02,,,,1,3,1,10311,Z,2023-09-30 04:22:49.883470,2025-05-13 12:31:54.373295,Y,Y,77523.95,1454.05,157814.07,20.62,,,,,,,,1093792350,1558719914,1720749690,1720749690,1023051596,001509,18.97,1003
 910809502733,4991490559710,,2800,21000,2024-06-27,2024-07-27,53497334068645KXT,,1,2025-08-15,2024-10-15,9999-12-31,96561,356275291,57,1,,,,465369.03,,,,8,4,F,11201,~,2024-11-03 06:13:33.783760,,Y,Y,180655.56,743.33,284713.47,5.47,,,,,,,,1093792350,1003488552,1942945159,1730548868,1437702123,001503,6.03,1003
 999999434800,0999999434800,472102076196,2081,21000,2019-01-28,2019-02-15,08962240069952RYH,,1,2025-08-15,2019-06-13,9999-12-31,39852,830951103,4,2,,,,939736.35,,,,9,5,5,12302,X,2019-08-20 18:19:44.806287,,Y,N,105073.42,1358.12,834662.93,24.84,,,,,,,,1104867175,1437702123,1972944437,1023051596,1558719914,001510,1.39,1003
+999999434801,0999999434801,673069517797,40,20000,2018-08-20,2018-09-17,49278204746841BZW,,1,2025-08-15,2022-10-03,9999-12-31,50315,800678894,RD,0,,,,502053.89,,,,5,6,B,02102,~,2024-09-08 18:51:16.069167,,Y,N,448471.81,924.67,53582.08,14.05,6,1447692959,,,,,24.43,1093792350,1942945159,1023051596,1003488552,1558719914,001502,16.72,7
+999999434802,0999999434802,,81,20000,2019-07-26,2019-08-25,07506929803761ZEC,,1,2025-08-15,2025-07-12,9999-12-31,44297,284934751,SV,3,,,,217013.36,,,,1,9,G,04212,E,2025-07-16 23:48:35.088250,,Y,N,152632.12,782.77,64381.24,19.72,,1730548868,1669572467.0,,803052.05,190912.0,22.1,,,,,,,,7
+999999434803,0999999434803,,2800,21000,2024-06-27,2024-07-27,53497334068645KXT,,1,2025-08-15,2024-10-15,9999-12-31,96561,356275291,57,1,,,,465369.03,,,,8,4,F,11201,~,2024-11-03 06:13:33.783760,,Y,N,180655.56,743.33,284713.47,5.47,,,,,,,,1093792350,1003488552,1942945159,1730548868,1437702123,001503,6.03,1003

apps/bfd-pipeline-idr/test_samples1/SYNTHETIC_CLM_INSTNL.csv

@@ -74,3 +74,4 @@ GEO_BENE_SK,CLM_DT_SGNTR_SK,CLM_TYPE_CD,CLM_NUM_SK,CLM_FI_ACTN_CD,CLM_ADMSN_TYPE
 90844,746386566763,1013,1,1,4,80,1,9,6,3,10,105.69,232.88,,,2419.57,1,0,0.74,15.82,8.59,22.9,19.86,18.73,20.47,,,2025-02-04 15:32:32.066356,,,,W,,,
 49683,261063166627,30,1,5,0,39,0,F,15,7,8,108.51,26.37,1867.93,1012.23,1019.41,2,5,0.92,24.13,11.65,9.88,12.52,23.96,20.73,21.68,,2025-08-14 13:34:11.226932,,,,16,,,
 39852,999999434800,2081,1,5,4,15,0,K,32,6,0,204.82,116.49,,,3374.91,0,1,0.92,1.11,21.29,6.31,14.73,4.2,12.0,,,,,,,,,,
+50315,999999434801,40,1,8,7,26,1,2,2,6,10,44.96,169.36,255.81,1792.19,2136.37,1,5,0.56,12.06,8.16,5.95,20.82,6.11,14.5,2.98,2,2023-06-21 12:30:48.824346,,13.72,,,,,

apps/bfd-pipeline-idr/test_samples1/SYNTHETIC_CLM_LINE.csv

@@ -1035,3 +1035,6 @@ CLM_UNIQ_ID,CLM_NUM_SK,CLM_TYPE_CD,GEO_BENE_SK,CLM_DT_SGNTR_SK,CLM_LINE_CVRD_PD_
 8244064276648,1,30,49683,261063166627,3.84,,520.44,,0.74,2025-02-25,2025-03-15,,,0.13,922.11,3.99,2.04,4.94,1.42,1,2025-02-25,,,,,,,99453,32,US,PC,0631,0.5,2025-08-14 11:57:36.751142,2025-08-14 19:14:23.874445,32,,U6
 4991490559710,1,2800,96561,910809502733,3.84,,520.44,,0.74,2025-02-25,2025-03-15,,,0.13,922.11,3.99,2.04,4.94,1.42,1,2025-02-25,,,,,,,99453,32,US,PC,0631,0.5,2025-08-14 11:57:36.751142,2025-08-14 19:14:23.874445,32,,U6
 0999999434800,1,2081,39852,999999434800,1.58,,939.29,00264180032,0.73,2019-01-28,2019-02-15,682.82,ML,0.94,1390.81,0.23,3.28,8.67,3.94,0,2019-01-28,,,,,,,99453,1P,23,U6,0223,2.25,,,U6,,PC
+0999999434801,1,40,50315,999999434801,1.88,,1246.4,00338004904,2.65,2018-08-20,2018-09-17,265.89,ML,4.14,851.0,3.64,1.14,5.16,3.98,1,2018-08-20,,,,,,,99453,23,U6,,0904,2.62,2024-05-27 18:48:44.414809,,,,
+0999999434802,1,81,44297,999999434802,0.12,,338.52,,2.21,2019-07-26,2019-08-25,,,0.03,526.04,2.11,0.5,3.8,4.08,0,2019-07-26,3.13,Y92311,98,,,,99453,22,U6,28,0760,4.54,2025-07-29 15:05:11.190614,,1P,1730548868,1P
+0999999434803,1,2800,96561,999999434803,3.84,,520.44,,0.74,2025-02-25,2025-03-15,,,0.13,922.11,3.99,2.04,4.94,1.42,1,2025-02-25,,,,,,,99453,32,US,PC,0631,0.5,2025-08-14 11:57:36.751142,2025-08-14 19:14:23.874445,32,,U6