BFD-4349: Prune older shared system phase 1 claims (#3178)

Author: JFisk42

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

@@ -0,0 +1,3 @@
+CREATE INDEX ON idr.claim_professional_ss(bfd_updated_ts) WHERE clm_type_cd BETWEEN 1000 and 1999;
+
+CREATE INDEX ON idr.claim_institutional_ss(bfd_updated_ts) WHERE clm_type_cd BETWEEN 1000 and 1999;

apps/bfd-pipeline-idr/constants.py

@@ -10,8 +10,18 @@
 CLAIM_RX_TABLE = "idr.claim_rx"
 CLAIM_PROFESSIONAL_NCH_TABLE = "idr.claim_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_SS_TABLE = "idr.claim_institutional_ss"
+CLAIM_INSTITUTIONAL_ITEM_SS_TABLE = "idr.claim_item_institutional_ss"
+
+# Phase 1 claims younger than 60 days are allowed
+PHASE_1_CUTOFF = 60
+PHASE_1_SS_MIN = 1000
+PHASE_1_SS_MAX = 1999
+FISS_CLM_SOURCE = "21000"
+MCS_CLM_SOURCE = "22000"
+VMS_CLM_SOURCE = "23000"
 
 IDR_PREFIX = "cms_vdm_view_mdcr_prd"
 IDR_BENE_HISTORY_TABLE = f"{IDR_PREFIX}.v2_mdcr_bene_hstry"

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

@@ -19,6 +19,7 @@
     DEFAULT_MAX_DATE,
     DEFAULT_MIN_DATE,
     EMPTY_PARTITION,
+    FISS_CLM_SOURCE,
     IDR_BENE_HISTORY_TABLE,
     IDR_CLAIM_ANSI_SIGNATURE_TABLE,
     IDR_CLAIM_DATE_SIGNATURE_TABLE,
@@ -29,11 +30,16 @@
     IDR_PRIOR_AUTH_TABLE,
     INSTITUTIONAL_NCH_PARTITIONS,
     INSTITUTIONAL_SS_PARTITIONS,
+    MCS_CLM_SOURCE,
     NON_CLAIM_PARTITION,
     PART_D_CLAIM_TYPE_CODES,
     PART_D_PARTITIONS,
+    PHASE_1_CUTOFF,
+    PHASE_1_SS_MAX,
+    PHASE_1_SS_MIN,
     PROFESSIONAL_NCH_PARTITIONS,
     PROFESSIONAL_SS_PARTITIONS,
+    VMS_CLM_SOURCE,
 )
 from load_partition import LoadPartition, LoadPartitionGroup, PartitionType
 from settings import (
@@ -42,6 +48,7 @@
     MIN_CLAIM_NCH_TRANSACTION_DATE,
     MIN_CLAIM_SS_TRANSACTION_DATE,
     MIN_PRIOR_AUTH_TRANSACTION_DATE,
+    PHASE_1_PRUNE_BATCH_LIMIT,
 )
 
 type DbType = str | float | int | bool | date | datetime
@@ -343,9 +350,6 @@ def base_claim_filter(partition: LoadPartition) -> str:
 EXPR = "expr"
 DERIVED = "derived"
 COLUMN_MAP = "column_map"
-FISS_CLM_SOURCE = "21000"
-MCS_CLM_SOURCE = "22000"
-VMS_CLM_SOURCE = "23000"
 
 
 ALIAS_CLM = "clm"
@@ -692,17 +696,16 @@ def claim_filter(start_time: datetime, partition: LoadPartition) -> str:
         latest_claim_ind = ""
 
     # PAC data older than 60 days should be filtered
-    pac_cutoff_date = start_time - timedelta(days=60)
+    pac_cutoff_date = start_time - timedelta(days=PHASE_1_CUTOFF)
     start_time_sql = pac_cutoff_date.strftime("'%Y-%m-%d %H:%M:%S'")
-    pac_phase_1_min = 1000
-    pac_phase_1_max = 1999
+
     # Note: checking clm_type_cd as the first branch of the OR here might be more efficient
     # Since it's more likely to return true
     pac_filter = (
         f"""
         AND
         (
-            {clm}.clm_type_cd NOT BETWEEN {pac_phase_1_min} AND {pac_phase_1_max}
+            {clm}.clm_type_cd NOT BETWEEN {PHASE_1_SS_MIN} AND {PHASE_1_SS_MAX}
             OR
             (
                 {clm}.clm_src_id IN (
@@ -825,3 +828,26 @@ def claim_related_conditions_cte(source: Source) -> str:
             AND clm_rlt_cond_cd != '~'
             GROUP BY clm_rlt_cond_sgntr_sk
     """
+
+
+def stale_phase_1_claims_query(
+    header_table: str,
+    item_table: str,
+    cutoff_date: datetime,
+) -> tuple[str, tuple[datetime, datetime]]:
+    return (
+        f""" 
+            SELECT clm.clm_uniq_id 
+            FROM {header_table} clm
+            WHERE clm.clm_type_cd BETWEEN {PHASE_1_SS_MIN} AND {PHASE_1_SS_MAX}
+            AND clm.clm_src_id IN ('{FISS_CLM_SOURCE}', '{MCS_CLM_SOURCE}', '{VMS_CLM_SOURCE}')
+            AND clm.bfd_updated_ts < %s
+            AND NOT EXISTS (
+                SELECT 1 FROM {item_table} item
+                WHERE clm.clm_uniq_id = item.clm_uniq_id
+                AND item.bfd_updated_ts >= %s
+            )
+            LIMIT {PHASE_1_PRUNE_BATCH_LIMIT}
+        """,
+        (cutoff_date, cutoff_date),
+    )

apps/bfd-pipeline-idr/pipeline_stages.py

@@ -35,7 +35,7 @@
 from model.idr_contract_pbp_number import IdrContractPbpNumber
 from model.idr_prior_auth import IdrPriorAuth
 from parallel_executor import ParallelStagesExecutor, Stage
-from pipeline_utils import extract_and_load, prune_bene_lis_cmbnd
+from pipeline_utils import extract_and_load, prune_bene_lis_cmbnd, prune_phase_1_ss_claims
 from settings import enable_prior_auth_ingestion
 
 type NodePartitionedModelInput = tuple[type[IdrBaseModel], LoadPartition | None]
@@ -47,6 +47,10 @@
     IdrClaimProfessionalSs,
     IdrClaimInstitutionalSs,
 ]
+CLAIM_SS_TABLES: list[type[IdrBaseModel]] = [
+    IdrClaimProfessionalSs,
+    IdrClaimInstitutionalSs,
+]
 CLAIM_AUX_TABLES: list[type[IdrBaseModel]] = [
     # RX/Part D is special because we combine claim + claim line
     IdrClaimRx,
@@ -147,6 +151,14 @@ def _stage5_do_phase_1_prune(self) -> Stage[bool]:
             self.load_mode,
         )
 
+        for model in self._filter_tables(CLAIM_SS_TABLES):
+            yield functools.partial(
+                prune_phase_1_ss_claims,
+                model,
+                self.load_mode,
+                self.start_time,
+            )
+
     def _filter_tables(self, tables: list[type[IdrBaseModel]]) -> list[type[IdrBaseModel]]:
         return [
             t

apps/bfd-pipeline-idr/pipeline_utils.py

@@ -8,13 +8,22 @@
 from snowflake.connector.network import ReauthenticationRequest, RetryRequest
 
 from batch_worker import LoadingBatchWorkerClient
-from constants import DEFAULT_MAX_DATE, DEFAULT_PARTITION
+from constants import (
+    CLAIM_INSTITUTIONAL_ITEM_SS_TABLE,
+    CLAIM_INSTITUTIONAL_SS_TABLE,
+    CLAIM_PROFESSIONAL_ITEM_SS_TABLE,
+    CLAIM_PROFESSIONAL_SS_TABLE,
+    DEFAULT_MAX_DATE,
+    DEFAULT_PARTITION,
+    PHASE_1_CUTOFF,
+)
 from extractor import PostgresExtractor, SnowflakeExtractor, Source
 from load_partition import LoadPartition
 from loader import LoadType, PostgresLoader, get_connection_string, should_track_load_progress
 from model.base_model import (
     LoadMode,
     T,
+    stale_phase_1_claims_query,
 )
 from model.idr_beneficiary_low_income_subsidy_cmbnd import IdrBeneficiaryLowIncomeSubsidyCmbnd
 from model.load_progress import LoadProgress
@@ -115,6 +124,45 @@ def extract_and_load(
             raise ex
 
 
+def prune_phase_1_ss_claims(
+    cls: type[T],
+    load_mode: LoadMode,
+    job_start: datetime,
+) -> bool:
+    shared_claim_tables = {
+        CLAIM_INSTITUTIONAL_SS_TABLE: CLAIM_INSTITUTIONAL_ITEM_SS_TABLE,
+        CLAIM_PROFESSIONAL_SS_TABLE: CLAIM_PROFESSIONAL_ITEM_SS_TABLE,
+    }
+
+    claim_table = cls.table()
+    item_table = shared_claim_tables.get(claim_table)
+    if item_table is None:
+        return True
+
+    prune_cutoff_date = job_start - timedelta(days=PHASE_1_CUTOFF)
+    logger.info("pruning phase 1 ss claims older than {}", prune_cutoff_date)
+
+    prune_query, params = stale_phase_1_claims_query(claim_table, item_table, prune_cutoff_date)
+
+    with psycopg.connect(get_connection_string(load_mode)) as conn:
+        for target_table in [item_table, claim_table]:
+            total_row_count = 0
+            while True:
+                with conn.transaction():
+                    res = conn.execute(
+                        f"""DELETE FROM {target_table} WHERE clm_uniq_id IN ({prune_query})""",  # type: ignore
+                        params,
+                    )
+
+                    total_row_count += res.rowcount
+                    logger.info("pruned {} rows from {}", res.rowcount, item_table)
+
+                    if res.rowcount == 0:
+                        logger.info("Total rows pruned from {}: {}", item_table, total_row_count)
+                        break
+    return True
+
+
 def prune_bene_lis_cmbnd(
     load_mode: LoadMode,
 ) -> bool:

apps/bfd-pipeline-idr/settings.py

@@ -100,6 +100,10 @@ def enable_prior_auth_ingestion() -> bool:
 """Number of minimum connections to hold in the pool concurrently per-batch for non-LOCAL loads.
 Defaults to 20."""
 
+PHASE_1_PRUNE_BATCH_LIMIT = int(getenv("PHASE_1_PRUNE_BATCH_LIMIT", "10_000"))
+"""The maximum batch size for pruning old claims (phase 1 claims from shared systems) on
+INCREMENTAL loads. Defaults to 10000."""
+
 BENEFICIARY_PRUNE_BATCH_LIMIT = int(getenv("IDR_BENEFICIARY_PRUNE_BATCH_LIMIT", "1000"))
 """Maximum rows to delete per prune statement for LIS combined beneficiary records."""
 

apps/bfd-pipeline-idr/test_pipeline.py

@@ -2,7 +2,7 @@
 import shutil
 import subprocess
 from collections.abc import Generator
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import cast
 from uuid import uuid4
@@ -198,11 +198,6 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
     rows = cur.fetchmany(1)
     assert rows[0]["clm_uniq_id"] == 113370100080
 
-    cur = conn.execute("select * from idr.claim_institutional_ss order by clm_uniq_id")
-    assert cur.rowcount == 21
-    rows = cur.fetchmany(1)
-    assert rows[0]["clm_uniq_id"] == 123359318723
-
     cur = conn.execute("select * from idr.claim_professional_nch order by clm_uniq_id")
     assert cur.rowcount == 51
     rows = cur.fetchmany(1)
@@ -223,11 +218,6 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
     rows = cur.fetchmany(1)
     assert rows[0]["clm_uniq_id"] == 113370100080
 
-    cur = conn.execute("select * from idr.claim_item_institutional_ss order by clm_uniq_id")
-    assert cur.rowcount == 327
-    rows = cur.fetchmany(1)
-    assert rows[0]["clm_uniq_id"] == 123359318723
-
     cur = conn.execute("select * from idr.claim_item_professional_nch order by clm_uniq_id")
     assert cur.rowcount == 442
     rows = cur.fetchmany(1)
@@ -240,6 +230,32 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
 
     conn.commit()
 
+    # Phase 1 SS (PAC) claims older than 60 days will be pruned on incremental loads
+    if load_type == LoadType.INITIAL:
+        cur = conn.execute("select * from idr.claim_institutional_ss order by clm_uniq_id")
+        assert cur.rowcount == 21
+        rows = cur.fetchmany(1)
+        assert rows[0]["clm_uniq_id"] == 123359318723
+
+        cur = conn.execute("select * from idr.claim_item_institutional_ss order by clm_uniq_id")
+        assert cur.rowcount == 327
+        rows = cur.fetchmany(1)
+        assert rows[0]["clm_uniq_id"] == 123359318723
+
+    else:
+        make_it_stale_ts = datetime.now(UTC) + timedelta(days=60)
+        _advance_time(make_it_stale_ts)
+        run(Source.POSTGRES, LoadMode.SYNTHETIC, LoadType.INCREMENTAL)
+        cur = conn.execute("select * from idr.claim_institutional_ss order by clm_uniq_id")
+        assert cur.rowcount == 9
+        rows = cur.fetchmany(1)
+        assert rows[0]["clm_uniq_id"] == 849348853948
+
+        cur = conn.execute("select * from idr.claim_item_institutional_ss order by clm_uniq_id")
+        assert cur.rowcount == 151
+        rows = cur.fetchmany(1)
+        assert rows[0]["clm_uniq_id"] == 849348853948
+
     # Test incremental loading logic involving 'source_load_events' if we're testing incremental
     # mode
     if load_type == LoadType.INCREMENTAL: