Skip to content

Commit 36efe2b

Browse files
patrickgmaydaypatrickgmayday
committed
-Merge header and item pruning to one transaction
-Align code more closely for ss claims pruning for more consistent architecture -add indexes -small testing update --fix missing import when converting from wildcard imports
1 parent d0b55dd commit 36efe2b

6 files changed

Lines changed: 45 additions & 94 deletions

File tree

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
CREATE INDEX ON idr.claim_professional_ss(clm_idr_ld_dt, clm_uniq_id) WHERE clm_ltst_clm_ind = 'N';
2+
3+
CREATE INDEX ON idr.claim_institutional_ss(clm_idr_ld_dt, clm_uniq_id) WHERE clm_ltst_clm_ind = 'N';

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

Lines changed: 9 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -867,41 +867,21 @@ def stale_phase_1_claims_query(
867867
)
868868

869869

870-
def non_latest_non_part_d_claim_items_query(
871-
item_table: str,
872-
cutoff_date: datetime,
873-
) -> tuple[str, tuple[datetime]]:
874-
part_d_codes = ",".join(str(code) for code in PART_D_CLAIM_TYPE_CODES)
875-
876-
return (
877-
f"""
878-
SELECT item.clm_uniq_id, item.bfd_row_id
879-
FROM {item_table} item
880-
JOIN {IDR_CLAIM_TABLE} clm ON clm.clm_uniq_id = item.clm_uniq_id
881-
WHERE clm.clm_ltst_clm_ind = 'N'
882-
AND clm.clm_type_cd NOT IN ({part_d_codes})
883-
AND clm.clm_idr_ld_dt < %s
884-
ORDER BY item.clm_uniq_id, item.bfd_row_id
885-
LIMIT {PHASE_1_PRUNE_BATCH_LIMIT}
886-
""",
887-
(cutoff_date,),
888-
)
889-
890-
891-
def non_latest_non_part_d_parent_claims_query(
870+
def non_latest_non_part_d_claims_query(
892871
claim_table: str,
893872
cutoff_date: datetime,
894873
) -> tuple[str, tuple[datetime]]:
895-
part_d_codes = ",".join(str(code) for code in PART_D_CLAIM_TYPE_CODES)
896-
897874
return (
898875
f"""
876+
WITH claims AS (
877+
SELECT clm.clm_uniq_id
878+
FROM {claim_table} clm
879+
WHERE clm.clm_ltst_clm_ind = 'N'
880+
AND clm.clm_idr_ld_dt < %s
881+
ORDER BY clm.clm_idr_ld_dt, clm.clm_uniq_id
882+
)
899883
SELECT clm.clm_uniq_id
900-
FROM {claim_table} clm
901-
WHERE clm.clm_ltst_clm_ind = 'N'
902-
AND clm.clm_type_cd NOT IN ({part_d_codes})
903-
AND clm.clm_idr_ld_dt < %s
904-
ORDER BY clm.clm_uniq_id
884+
FROM claims clm
905885
LIMIT {PHASE_1_PRUNE_BATCH_LIMIT}
906886
""",
907887
(cutoff_date,),

‎apps/bfd-pipeline-idr/pipeline_stages.py‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,7 @@
4141
prune_bene_lis_cmbnd,
4242
prune_bene_ma_part_d,
4343
prune_bene_ma_part_d_rx,
44-
prune_non_latest_non_part_d_ss_claim_items,
45-
prune_non_latest_non_part_d_ss_parent_claims,
44+
prune_non_latest_non_part_d_ss_claims,
4645
prune_phase_1_ss_claims,
4746
)
4847
from settings import enable_prior_auth_ingestion
@@ -178,13 +177,7 @@ def _stage5_prune_obsolete_rows(self) -> Stage[bool]:
178177
self.start_time,
179178
)
180179
yield functools.partial(
181-
prune_non_latest_non_part_d_ss_claim_items,
182-
model,
183-
self.load_mode,
184-
self.start_time,
185-
)
186-
yield functools.partial(
187-
prune_non_latest_non_part_d_ss_parent_claims,
180+
prune_non_latest_non_part_d_ss_claims,
188181
model,
189182
self.load_mode,
190183
self.start_time,

‎apps/bfd-pipeline-idr/pipeline_utils.py‎

Lines changed: 26 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,7 @@
2323
from model.base_model import (
2424
LoadMode,
2525
T,
26-
non_latest_non_part_d_claim_items_query,
27-
non_latest_non_part_d_parent_claims_query,
26+
non_latest_non_part_d_claims_query,
2827
stale_phase_1_claims_query,
2928
)
3029
from model.idr_beneficiary_low_income_subsidy_cmbnd import IdrBeneficiaryLowIncomeSubsidyCmbnd
@@ -34,7 +33,6 @@
3433
from settings import (
3534
BENEFICIARY_PART_D_PRUNE_BATCH_LIMIT,
3635
BENEFICIARY_PRUNE_BATCH_LIMIT,
37-
PHASE_1_PRUNE_BATCH_LIMIT,
3836
)
3937

4038
_SHARED_SYSTEM_CLAIM_ITEM_TABLES = {
@@ -171,7 +169,7 @@ def prune_phase_1_ss_claims(
171169
return True
172170

173171

174-
def prune_non_latest_non_part_d_ss_claim_items(
172+
def prune_non_latest_non_part_d_ss_claims(
175173
cls: type[T],
176174
load_mode: LoadMode,
177175
job_start: datetime,
@@ -182,56 +180,39 @@ def prune_non_latest_non_part_d_ss_claim_items(
182180
return True
183181

184182
prune_cutoff_date = job_start - timedelta(days=PHASE_1_CUTOFF)
183+
logger.info("pruning non-latest non-Part-D ss claims older than {}", prune_cutoff_date)
185184

186-
logger.info("pruning non-latest non-Part-D ss claim items older than {}", prune_cutoff_date)
185+
prune_query, params = non_latest_non_part_d_claims_query(claim_table, prune_cutoff_date)
187186

188-
prune_query, params = non_latest_non_part_d_claim_items_query(item_table, prune_cutoff_date)
187+
total_row_counts = {
188+
item_table: 0,
189+
claim_table: 0,
190+
}
189191

190192
with psycopg.connect(get_connection_string(load_mode)) as conn:
191193
while True:
194+
claim_row_count = 0
192195
with conn.transaction():
193-
res = conn.execute(
194-
f"""
195-
DELETE FROM {item_table}
196-
WHERE (clm_uniq_id, bfd_row_id) IN ({prune_query})
197-
""", # type: ignore
198-
params,
199-
)
200-
logger.info("pruned {} rows from {}", res.rowcount, item_table)
201-
if res.rowcount < PHASE_1_PRUNE_BATCH_LIMIT:
202-
break
203-
204-
return True
205-
206-
207-
def prune_non_latest_non_part_d_ss_parent_claims(
208-
cls: type[T],
209-
load_mode: LoadMode,
210-
job_start: datetime,
211-
) -> bool:
212-
claim_table = cls.table()
213-
if claim_table not in _SHARED_SYSTEM_CLAIM_ITEM_TABLES:
214-
return True
215-
216-
prune_cutoff_date = job_start - timedelta(days=PHASE_1_CUTOFF)
196+
for target_table in [item_table, claim_table]:
197+
res = conn.execute(
198+
f"""DELETE FROM {target_table} WHERE clm_uniq_id IN ({prune_query})""", # type: ignore
199+
params,
200+
)
217201

218-
logger.info("pruning non-latest non-Part-D ss parent claims older than {}", prune_cutoff_date)
202+
total_row_counts[target_table] += res.rowcount
203+
logger.info("pruned {} rows from {}", res.rowcount, target_table)
219204

220-
prune_query, params = non_latest_non_part_d_parent_claims_query(claim_table, prune_cutoff_date)
205+
if target_table == claim_table:
206+
claim_row_count = res.rowcount
221207

222-
with psycopg.connect(get_connection_string(load_mode)) as conn:
223-
while True:
224-
with conn.transaction():
225-
res = conn.execute(
226-
f"""
227-
DELETE FROM {claim_table}
228-
WHERE clm_uniq_id IN ({prune_query})
229-
""", # type: ignore
230-
params,
231-
)
232-
logger.info("pruned {} rows from {}", res.rowcount, claim_table)
233-
if res.rowcount < PHASE_1_PRUNE_BATCH_LIMIT:
234-
break
208+
if claim_row_count == 0:
209+
for target_table in [item_table, claim_table]:
210+
logger.info(
211+
"Total rows pruned from {}: {}",
212+
target_table,
213+
total_row_counts[target_table],
214+
)
215+
break
235216

236217
return True
237218

‎apps/bfd-pipeline-idr/test_pipeline.py‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -105,17 +105,17 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
105105
assert row is not None
106106

107107
non_latest_parent_claim = dict(row)
108-
non_latest_parent_claim["clm_uniq_id"] = 999999434801
108+
non_latest_parent_claim["clm_uniq_id"] = 999999434800
109109
non_latest_parent_claim["clm_ltst_clm_ind"] = "N"
110110
non_latest_parent_claim["clm_type_cd"] = 2081
111111
# Make the seeded claim old enough to be picked up by the prune job.
112112
non_latest_parent_claim["clm_idr_ld_dt"] = datetime(2019, 6, 13, tzinfo=UTC).date()
113113

114114
if "bfd_row_id" in non_latest_parent_claim:
115-
non_latest_parent_claim["bfd_row_id"] = 999999434801
115+
non_latest_parent_claim["bfd_row_id"] = 999999434800
116116

117117
if "clm_dt_sgntr_sk" in non_latest_parent_claim:
118-
non_latest_parent_claim["clm_dt_sgntr_sk"] = 999999434801
118+
non_latest_parent_claim["clm_dt_sgntr_sk"] = 999999434800
119119

120120
conn.execute(
121121
t"""
@@ -251,17 +251,10 @@ def _do_test_pipeline(conn: Connection[DictRow], load_type: LoadType) -> None:
251251
rows = cur.fetchmany(1)
252252
assert rows[0]["clm_uniq_id"] == 113370100080
253253

254-
# Non-latest non-Part-D SS parent claims are filtered before final claim-table load
254+
# Non-latest non-Part-D SS parent claims do not remain in the final claim table
255255
cur = conn.execute("select * from idr.claim_institutional_ss where clm_uniq_id = 999999434800")
256256
assert cur.rowcount == 0
257257

258-
# Existing non-latest non-Part-D SS parent claims are pruned on incremental loads
259-
if load_type == LoadType.INCREMENTAL:
260-
cur = conn.execute(
261-
"select * from idr.claim_institutional_ss where clm_uniq_id = 999999434801"
262-
)
263-
assert cur.rowcount == 0
264-
265258
cur = conn.execute("select * from idr.claim_professional_nch order by clm_uniq_id")
266259
assert cur.rowcount == 51
267260
rows = cur.fetchmany(1)

‎apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/claim/ClaimAsyncService.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import gov.cms.bfd.server.ng.claim.model.ClaimBase;
1010
import gov.cms.bfd.server.ng.claim.model.ClaimSubtype;
1111
import gov.cms.bfd.server.ng.claim.model.ClaimTypeCode;
12+
import gov.cms.bfd.server.ng.claim.model.PriorAuthorization;
1213
import gov.cms.bfd.server.ng.claim.model.SystemType;
1314
import gov.cms.bfd.server.ng.input.ClaimSearchCriteria;
1415
import gov.cms.bfd.server.ng.log.QueryTelemetryUtil;

0 commit comments

Comments
 (0)