Merge remote-tracking branch 'origin/master' into BFD-4348-prune-older-claim-versions-excluding-part-d

Author: patrickgmayday

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/model/base_model.py

@@ -48,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
@@ -827,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,8 +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_phase_1_ss_claims
+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]
@@ -48,7 +47,7 @@
     IdrClaimProfessionalSs,
     IdrClaimInstitutionalSs,
 ]
-_CLAIM_SS_TABLES: list[type[IdrBaseModel]] = [
+CLAIM_SS_TABLES: list[type[IdrBaseModel]] = [
     IdrClaimProfessionalSs,
     IdrClaimInstitutionalSs,
 ]
@@ -152,11 +151,7 @@ def _stage5_do_phase_1_prune(self) -> Stage[bool]:
             self.load_mode,
         )
 
-    def _stage5_do_phase_1_prune(self) -> Stage[bool]:
-        if self.load_type == LoadType.INITIAL:
-            return
-
-        for model in self._filter_tables(_CLAIM_SS_TABLES):
+        for model in self._filter_tables(CLAIM_SS_TABLES):
             yield functools.partial(
                 prune_phase_1_ss_claims,
                 model,

apps/bfd-pipeline-idr/pipeline_utils.py

@@ -9,33 +9,28 @@
 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,
-    FISS_CLM_SOURCE,
     IDR_CLAIM_TABLE,
-    MCS_CLM_SOURCE,
     PART_D_CLAIM_TYPE_CODES,
     PHASE_1_CUTOFF,
-    PHASE_1_SS_MAX,
-    PHASE_1_SS_MIN,
-    VMS_CLM_SOURCE,
 )
 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
-from settings import BENEFICIARY_PRUNE_BATCH_LIMIT
-from settings import PRUNE_BATCH_MAX_SIZE
+from settings import BENEFICIARY_PRUNE_BATCH_LIMIT, PRUNE_BATCH_MAX_SIZE
 
 
 def get_progress(
@@ -138,7 +133,7 @@ def _prune_table_in_batches(
     delete_query: str,
     params: tuple[Any, ...] | None = None,
 ) -> None:
-    # Run a batched DELETE until no matching rows remain.
+    """Run a batched DELETE until no matching rows remain."""
     total_rows_pruned = 0
 
     while True:
@@ -174,50 +169,27 @@ def prune_phase_1_ss_claims(
     prune_cutoff_date = job_start - timedelta(days=PHASE_1_CUTOFF)
     part_d_codes = ",".join(str(code) for code in PART_D_CLAIM_TYPE_CODES)
 
-    phase_1_claim_filter = f"""
-        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.clm_idr_ld_dt < %s
-    """
-    non_latest_non_part_d_claim_filter = f"""
-        clm.clm_ltst_clm_ind = 'N'
-        AND clm.clm_type_cd NOT IN ({part_d_codes})
-        AND clm.clm_idr_ld_dt < %s
-    """
-
     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:
-        _prune_table_in_batches(
-            conn,
-            item_table,
-            f"""
-                    DELETE FROM {item_table}
-                    WHERE (clm_uniq_id, bfd_row_id) IN (
-                        SELECT item.clm_uniq_id, item.bfd_row_id
-                        FROM {item_table} item
-                        JOIN {claim_table} clm ON clm.clm_uniq_id = item.clm_uniq_id
-                        WHERE {phase_1_claim_filter}
-                        LIMIT {PRUNE_BATCH_MAX_SIZE}
-                    )
+        for target_table in [item_table, claim_table]:
+            _prune_table_in_batches(
+                conn,
+                target_table,
+                f"""
+                    DELETE FROM {target_table}
+                    WHERE clm_uniq_id IN ({prune_query})
                 """,
-            (prune_cutoff_date,),
-        )
+                params,
+            )
 
-        _prune_table_in_batches(
-            conn,
-            claim_table,
-            f"""
-                    DELETE FROM {claim_table}
-                    WHERE clm_uniq_id IN (
-                        SELECT clm.clm_uniq_id FROM {claim_table} clm
-                        WHERE {phase_1_claim_filter}
-                        LIMIT {PRUNE_BATCH_MAX_SIZE}
-                    )
-                """,
-            (prune_cutoff_date,),
-        )
+        non_latest_non_part_d_claim_filter = f"""
+            clm.clm_ltst_clm_ind = 'N'
+            AND clm.clm_type_cd NOT IN ({part_d_codes})
+            AND clm.clm_idr_ld_dt < %s
+        """
 
         logger.info("pruning non-latest non-Part-D ss claims older than {}", prune_cutoff_date)
 

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
@@ -267,6 +267,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 == 328
+        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:

insights/terraform/projects/bb2/services/analytics/modules/lambda/update_athena_metric_tables/sql_templates/template_generate_metrics_for_report_date.sql

@@ -1515,41 +1515,59 @@ SELECT
       )
   ) as auth_v3_user_makes_it_to_permission_screen_bene_count,
   (
-    select
-      count(*)
-    from
-      api_audit_events
-    WHERE
-      (
-        CONTAINS((SELECT enabled_metrics_list FROM report_params),
-          'auth_v1_v2_user_clicks_connect_bene_count')
-        AND (
-          try_cast(crosswalk_fhir_id as BIGINT) > 0
-          OR COALESCE(try_cast(crosswalk_fhir_id_v3 as BIGINT), 0) > 0
-        )
-        AND type = 'Authorization'
-        AND 
-        ( path LIKE '/v1/o/authorize%'
-          OR path LIKE '/v2/o/authorize%'
-        )
-      )
+    SELECT COUNT(*)
+    FROM (
+        SELECT
+            path,
+            crosswalk_fhir_id_v3,
+            crosswalk_fhir_id
+        FROM
+            api_audit_events
+        WHERE
+            (
+                CONTAINS((SELECT enabled_metrics_list FROM report_params),
+                    'auth_v1_v2_user_clicks_connect_bene_count')
+                AND (
+                    try_cast(crosswalk_fhir_id as BIGINT) > 0
+                    OR COALESCE(try_cast(crosswalk_fhir_id_v3 as BIGINT), 0) > 0
+                )
+                AND type = 'Authorization'
+                AND (
+                    path LIKE '/v1/o/authorize%'
+                    OR path LIKE '/v2/o/authorize%'
+                )
+            )
+        GROUP BY
+            path,
+            crosswalk_fhir_id_v3,
+            crosswalk_fhir_id
+    ) AS grouped_results
   ) as auth_v1_v2_user_clicks_connect_bene_count,
   (
-    select
-      count(*)
-    from
-      api_audit_events
-    WHERE
-      (
-        CONTAINS((SELECT enabled_metrics_list FROM report_params),
-          'auth_v3_user_clicks_connect_bene_count')
-        AND (
-          try_cast(crosswalk_fhir_id as BIGINT) > 0
-          OR COALESCE(try_cast(crosswalk_fhir_id_v3 as BIGINT), 0) > 0
-        )
-        AND type = 'Authorization'
-        AND path LIKE '/v3/o/authorize%'
-      )
+    SELECT COUNT(*)
+    FROM (
+        SELECT
+            path,
+            crosswalk_fhir_id_v3,
+            crosswalk_fhir_id
+        FROM
+            api_audit_events
+        WHERE
+            (
+                CONTAINS((SELECT enabled_metrics_list FROM report_params),
+                    'auth_v3_user_clicks_connect_bene_count')
+                AND (
+                    try_cast(crosswalk_fhir_id as BIGINT) > 0
+                    OR COALESCE(try_cast(crosswalk_fhir_id_v3 as BIGINT), 0) > 0
+                )
+                AND type = 'Authorization'
+                AND path LIKE '/v3/o/authorize%'
+            )
+        GROUP BY
+            path,
+            crosswalk_fhir_id_v3,
+            crosswalk_fhir_id
+    ) AS grouped_results
   ) as auth_v3_user_clicks_connect_bene_count,
   (
     select