BFD-4205: Refactor xref and kill credit handling (#2782)

Author: aschey-forpeople

apps/bfd-model-idr/generator_util.py

@@ -127,6 +127,9 @@ def generate_bene_xref(self, new_bene_sk, old_bene_sk):
         src_rec_ctre_ts = self.fake.date_time_between_dates(
             efctv_ts, datetime.datetime.now() - datetime.timedelta(days=1)
         )
+        src_rec_updt_ts = self.fake.date_time_between_dates(
+            src_rec_ctre_ts, datetime.datetime.now() - datetime.timedelta(days=1)
+        )
         insrt_ts = self.fake.date_time_between_dates(
             efctv_ts, datetime.datetime.now() - datetime.timedelta(days=1)
         )
@@ -140,6 +143,7 @@ def generate_bene_xref(self, new_bene_sk, old_bene_sk):
             "BENE_HICN_NUM": bene_hicn_num,
             "BENE_KILL_CRED_CD": str(kill_cred_cd),
             "SRC_REC_CRTE_TS": str(src_rec_ctre_ts),
+            "SRC_REC_UPDT_TS": str(src_rec_updt_ts),
             "IDR_TRANS_EFCTV_TS": str(efctv_ts),
             "IDR_INSRT_TS": str(insrt_ts),
             "IDR_UPDT_TS": str(updt_ts),
@@ -456,6 +460,7 @@ def save_output_files(self):
                 [
                     "BENE_SK",
                     "BENE_XREF_EFCTV_SK",
+                    "BENE_XREF_SK",
                     "BENE_MBI_ID",
                     "BENE_LAST_NAME",
                     "BENE_1ST_NAME",

apps/bfd-model-idr/patient_generator.py

@@ -146,6 +146,7 @@
 
         patient["BENE_SK"] = str(pt_bene_sk)
         patient["BENE_XREF_EFCTV_SK"] = str(pt_bene_sk)
+        patient["BENE_XREF_SK"] = patient["BENE_XREF_EFCTV_SK"]
         generator.used_bene_sk.append(pt_bene_sk)
 
         if pd.notna(row.get("BENE_MBI_ID")):
@@ -186,6 +187,7 @@
         pt_bene_sk = generator.gen_bene_sk()
         patient["BENE_SK"] = str(pt_bene_sk)
         patient["BENE_XREF_EFCTV_SK"] = str(pt_bene_sk)
+        patient["BENE_XREF_SK"] = patient["BENE_XREF_EFCTV_SK"]
         generator.used_bene_sk.append(pt_bene_sk)
 
         num_mbis = random.choices([1, 2, 3, 4], weights=[0.8, 0.14, 0.05, 0.01])[0]

apps/bfd-pipeline/bfd-pipeline-idr/bfd.sql

@@ -2,10 +2,12 @@ DROP SCHEMA IF EXISTS idr CASCADE;
 
 CREATE SCHEMA idr;
 CREATE TABLE idr.beneficiary(
+    -- columns from V2_MDCR_BENE_HSTRY
     bene_sk BIGINT NOT NULL, 
     bene_xref_efctv_sk BIGINT NOT NULL, 
     bene_xref_efctv_sk_computed BIGINT NOT NULL GENERATED ALWAYS 
-        AS (CASE WHEN bene_xref_efctv_sk = 0 THEN bene_sk ELSE bene_xref_efctv_sk END) STORED,
+        -- if kill credit is set to 1, the merge is invalid, and we should act like it's not present
+        AS (CASE WHEN bene_xref_efctv_sk = 0 OR bene_kill_cred_cd = '1' THEN bene_sk ELSE bene_xref_efctv_sk END) STORED,
     bene_mbi_id VARCHAR(11) NOT NULL,
     bene_1st_name VARCHAR(30) NOT NULL,
     bene_midl_name VARCHAR(15) NOT NULL,
@@ -28,8 +30,13 @@ CREATE TABLE idr.beneficiary(
     idr_ltst_trans_flg VARCHAR(1) NOT NULL,
     idr_trans_efctv_ts TIMESTAMPTZ NOT NULL,
     idr_trans_obslt_ts TIMESTAMPTZ NOT NULL,
-    idr_insrt_ts TIMESTAMPTZ NOT NULL,
-    idr_updt_ts TIMESTAMPTZ NOT NULL,
+    idr_insrt_ts_bene TIMESTAMPTZ NOT NULL,
+    idr_updt_ts_bene TIMESTAMPTZ NOT NULL,
+    -- columns from V2_MDCR_BENE_XREF
+    bene_kill_cred_cd VARCHAR(1) NOT NULL,
+    src_rec_updt_ts TIMESTAMPTZ NOT NULL,
+    idr_insrt_ts_xref TIMESTAMPTZ NOT NULL,
+    idr_updt_ts_xref TIMESTAMPTZ NOT NULL,
     bfd_created_ts TIMESTAMPTZ NOT NULL,
     bfd_updated_ts TIMESTAMPTZ NOT NULL,
     PRIMARY KEY(bene_sk, idr_trans_efctv_ts)
@@ -129,21 +136,6 @@ CREATE TABLE idr.beneficiary_election_period_usage (
     PRIMARY KEY(bene_sk, cntrct_pbp_sk, bene_enrlmt_efctv_dt)
 );
 
-CREATE TABLE idr.beneficiary_xref (
-    bene_sk BIGINT NOT NULL,
-    bene_xref_sk BIGINT NOT NULL,
-    bene_hicn_num VARCHAR(11) NOT NULL,
-    bene_kill_cred_cd VARCHAR(1) NOT NULL,
-    idr_insrt_ts TIMESTAMPTZ NOT NULL,
-    idr_updt_ts TIMESTAMPTZ NOT NULL,
-    src_rec_crte_ts TIMESTAMPTZ NOT NULL,
-    idr_trans_efctv_ts TIMESTAMPTZ NOT NULL,
-    idr_trans_obslt_ts TIMESTAMPTZ NOT NULL,
-    bfd_created_ts TIMESTAMPTZ NOT NULL,
-    bfd_updated_ts TIMESTAMPTZ NOT NULL,
-    PRIMARY KEY(bene_sk, bene_hicn_num, src_rec_crte_ts)
-);
-
 CREATE TABLE idr.contract_pbp_number (
     cntrct_pbp_sk BIGINT NOT NULL PRIMARY KEY,
     cntrct_drug_plan_ind_cd VARCHAR(1) NOT NULL,
@@ -447,6 +439,17 @@ CREATE VIEW idr.beneficiary_entitlement_reason_current AS
 SELECT * FROM idr.beneficiary_entitlement_reason
 WHERE idr_ltst_trans_flg = 'Y' AND bene_rng_bgn_dt <= NOW() - INTERVAL '12 hours' AND bene_rng_end_dt >= NOW() - INTERVAL '12 hours';
 
+CREATE VIEW idr.beneficiary_identity AS
+SELECT DISTINCT 
+    bene.bene_sk, 
+    bene.bene_xref_efctv_sk_computed, 
+    bene.bene_mbi_id,
+    bene_mbi.bene_mbi_efctv_dt, 
+    bene_mbi.bene_mbi_obslt_dt
+FROM idr.beneficiary bene
+LEFT JOIN idr.beneficiary_mbi_id bene_mbi 
+    ON bene.bene_mbi_id = bene_mbi.bene_mbi_id;
+
 CREATE OR REPLACE FUNCTION idr.refresh_overshare_mbis()
     RETURNS VOID AS $$
     DECLARE comment_sql TEXT;

apps/bfd-pipeline/bfd-pipeline-idr/mock-idr.sql

@@ -3,7 +3,8 @@ DROP SCHEMA IF EXISTS cms_vdm_view_mdcr_prd CASCADE;
 CREATE SCHEMA cms_vdm_view_mdcr_prd;
 
 CREATE TABLE cms_vdm_view_mdcr_prd.v2_mdcr_bene_hstry (
-     bene_sk BIGINT NOT NULL, 
+    bene_sk BIGINT NOT NULL, 
+    bene_xref_sk BIGINT NOT NULL,
     bene_xref_efctv_sk BIGINT NOT NULL, 
     bene_mbi_id VARCHAR(11),
     bene_1st_name VARCHAR(30),
@@ -122,6 +123,7 @@ CREATE TABLE cms_vdm_view_mdcr_prd.v2_mdcr_bene_xref (
     idr_insrt_ts TIMESTAMPTZ NOT NULL,
     idr_updt_ts TIMESTAMPTZ,
     src_rec_crte_ts TIMESTAMPTZ NOT NULL,
+    src_rec_updt_ts TIMESTAMPTZ NOT NULL,
     PRIMARY KEY(bene_sk, bene_hicn_num, src_rec_crte_ts)
 );
 

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

@@ -59,6 +59,17 @@ def transform_null_int(value: int | None) -> float:
 DERIVED = "derived"
 COLUMN_MAP = "column_map"
 
+ALIAS_CLM = "clm"
+ALIAS_DCMTN = "dcmtn"
+ALIAS_SGNTR = "sgntr"
+ALIAS_LINE = "line"
+ALIAS_PROCEDURE = "prod"
+ALIAS_INSTNL = "instnl"
+ALIAS_PRFNL = "prfnl"
+ALIAS_VAL = "val"
+ALIAS_HSTRY = "hstry"
+ALIAS_XREF = "xref"
+
 
 class IdrBaseModel(BaseModel):
     @staticmethod
@@ -160,7 +171,8 @@ def insert_keys(cls) -> list[str]:
 
 
 class IdrBeneficiary(IdrBaseModel):
-    bene_sk: Annotated[int, {PRIMARY_KEY: True}]
+    # columns from V2_MDCR_BENE_HSTRY
+    bene_sk: Annotated[int, {PRIMARY_KEY: True, ALIAS: ALIAS_HSTRY}]
     bene_xref_efctv_sk: int
     bene_mbi_id: Annotated[str, BeforeValidator(transform_null_string)]
     bene_1st_name: str
@@ -184,9 +196,26 @@ class IdrBeneficiary(IdrBaseModel):
     idr_ltst_trans_flg: Annotated[str, BeforeValidator(transform_null_string)]
     idr_trans_efctv_ts: Annotated[datetime, {PRIMARY_KEY: True}]
     idr_trans_obslt_ts: datetime
-    idr_insrt_ts: Annotated[datetime, {BATCH_TIMESTAMP: True}]
-    idr_updt_ts: Annotated[
-        datetime, {UPDATE_TIMESTAMP: True}, BeforeValidator(transform_null_date_to_min)
+    idr_insrt_ts_bene: Annotated[
+        datetime, {BATCH_TIMESTAMP: True, ALIAS: ALIAS_HSTRY, COLUMN_MAP: "idr_insrt_ts"}
+    ]
+    idr_updt_ts_bene: Annotated[
+        datetime,
+        {UPDATE_TIMESTAMP: True, ALIAS: ALIAS_HSTRY, COLUMN_MAP: "idr_updt_ts"},
+        BeforeValidator(transform_null_date_to_min),
+    ]
+    # columns from V2_MDCR_BENE_XREF
+    bene_kill_cred_cd: Annotated[str, BeforeValidator(transform_default_string)]
+    src_rec_updt_ts: Annotated[datetime, BeforeValidator(transform_null_date_to_min)]
+    idr_insrt_ts_xref: Annotated[
+        datetime,
+        {BATCH_TIMESTAMP: True, ALIAS: ALIAS_XREF, COLUMN_MAP: "idr_insrt_ts"},
+        BeforeValidator(transform_null_date_to_min),
+    ]
+    idr_updt_ts_xref: Annotated[
+        datetime,
+        {UPDATE_TIMESTAMP: True, ALIAS: ALIAS_XREF, COLUMN_MAP: "idr_updt_ts"},
+        BeforeValidator(transform_null_date_to_min),
     ]
 
     @staticmethod
@@ -199,11 +228,47 @@ def computed_keys() -> list[str]:
 
     @staticmethod
     def _current_fetch_query(start_time: datetime) -> str:  # noqa: ARG004
-        return """
-            SELECT {COLUMNS}
-            FROM cms_vdm_view_mdcr_prd.v2_mdcr_bene_hstry
-            {WHERE_CLAUSE}
-            {ORDER_BY}
+        hstry = ALIAS_HSTRY
+        xref = ALIAS_XREF
+        # There can be multiple xref records for the same bene_sk/bene_ref_sk combo
+        # so we need to find the most recent one based on src_rec_updt_ts.
+
+        # Unlike idr_updt_ts, src_rec_updt_ts will be set to the created timestamp
+        # if no update has been applied. Therefore, we can just check the updated timestamp
+        # without caring about the created timestamp.
+
+        # There can also be duplicate values with the same idr_insrt_ts, so we have to rely on
+        # src_rec_insrt_ts/src_rec_updt_ts for this.
+        return f"""
+            WITH ordered_xref AS (
+                SELECT bene_sk, bene_xref_sk, ROW_NUMBER() OVER (
+                    PARTITION BY bene_sk, bene_xref_sk 
+                    ORDER BY src_rec_updt_ts DESC
+                ) AS row_order
+                FROM cms_vdm_view_mdcr_prd.v2_mdcr_bene_xref
+            ), 
+            current_xref AS (
+                SELECT 
+                    ox.bene_sk, 
+                    ox.bene_xref_sk, 
+                    bx.bene_kill_cred_cd, 
+                    bx.src_rec_updt_ts,
+                    bx.idr_insrt_ts,
+                    bx.idr_updt_ts
+                FROM ordered_xref ox
+                JOIN cms_vdm_view_mdcr_prd.v2_mdcr_bene_xref bx
+                    ON bx.bene_sk = ox.bene_sk AND bx.bene_xref_sk = ox.bene_xref_sk
+                WHERE ox.row_order = 1
+            )
+            SELECT {{COLUMNS}}
+            FROM cms_vdm_view_mdcr_prd.v2_mdcr_bene_hstry {hstry}
+            -- NOTE: the join condition is intentionally inverted here
+            -- In the xref table, the bene_sk and bene_xref_sk fields are mirrored
+            LEFT JOIN current_xref {xref} 
+                ON {xref}.bene_sk = {hstry}.bene_xref_sk 
+                AND {xref}.bene_xref_sk = {hstry}.bene_sk
+            {{WHERE_CLAUSE}}
+            {{ORDER_BY}}
         """
 
 
@@ -344,33 +409,6 @@ def _current_fetch_query(start_time: datetime) -> str:  # noqa: ARG004
         """
 
 
-class IdrBeneficiaryXref(IdrBaseModel):
-    bene_hicn_num: Annotated[str, {PRIMARY_KEY: True}]
-    bene_sk: Annotated[int, {PRIMARY_KEY: True}]
-    bene_xref_sk: int
-    bene_kill_cred_cd: Annotated[str, BeforeValidator(transform_default_string)]
-    idr_trans_efctv_ts: datetime
-    idr_trans_obslt_ts: datetime
-    idr_insrt_ts: datetime
-    idr_updt_ts: Annotated[
-        datetime, {UPDATE_TIMESTAMP: True}, BeforeValidator(transform_null_date_to_min)
-    ]
-    src_rec_crte_ts: Annotated[datetime, {PRIMARY_KEY: True, BATCH_TIMESTAMP: True}]
-
-    @staticmethod
-    def table() -> str:
-        return "idr.beneficiary_xref"
-
-    @staticmethod
-    def _current_fetch_query(start_time: datetime) -> str:  # noqa: ARG004
-        return """
-            SELECT {COLUMNS}
-            FROM cms_vdm_view_mdcr_prd.v2_mdcr_bene_xref
-            {WHERE_CLAUSE}
-            {ORDER_BY}
-        """
-
-
 class IdrElectionPeriodUsage(IdrBaseModel):
     bene_sk: Annotated[int, {PRIMARY_KEY: True}]
     cntrct_pbp_sk: Annotated[int, {PRIMARY_KEY: True}]
@@ -422,16 +460,6 @@ def _current_fetch_query(start_time: datetime) -> str:  # noqa: ARG004
         """
 
 
-ALIAS_CLM = "clm"
-ALIAS_DCMTN = "dcmtn"
-ALIAS_SGNTR = "sgntr"
-ALIAS_LINE = "line"
-ALIAS_PROCEDURE = "prod"
-ALIAS_INSTNL = "instnl"
-ALIAS_PRFNL = "prfnl"
-ALIAS_VAL = "val"
-
-
 def claim_type_clause(start_time: datetime) -> str:  # noqa: ARG001
     latest_claims_env = "IDR_LATEST_CLAIMS"
     if latest_claims_env in os.environ and os.environ[latest_claims_env] in ("1", "true"):

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

@@ -17,7 +17,6 @@
     IdrBeneficiaryMbiId,
     IdrBeneficiaryStatus,
     IdrBeneficiaryThirdParty,
-    IdrBeneficiaryXref,
     IdrClaim,
     IdrClaimAnsiSignature,
     IdrClaimDateSignature,
@@ -142,7 +141,6 @@ def run_pipeline(data_extractor: Extractor, connection_string: str) -> None:
         IdrBeneficiaryThirdParty,
         IdrBeneficiaryEntitlement,
         IdrBeneficiaryEntitlementReason,
-        IdrBeneficiaryXref,
         # Ignore for now, we'll likely source these elsewhere when we load contract data
         # IdrContractPbpNumber,
         # IdrElectionPeriodUsage,

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

@@ -75,10 +75,12 @@ def test_pipeline(self, psql_url: str) -> None:
         assert rows[0]["bene_mbi_id"] == "1S000000000"
         assert rows[1]["bene_mbi_id"] == "5B88XK5JN88"
 
-        cur = conn.execute("select * from idr.beneficiary_xref")
+        cur = conn.execute(
+            "select * from idr.beneficiary where bene_kill_cred_cd != '' order by bene_sk"
+        )
         assert cur.rowcount == 6
         rows = cur.fetchmany(1)
-        assert rows[0]["bene_sk"] == 454323619
+        assert rows[0]["bene_sk"] == 174441863
 
         cur = conn.execute("select * from idr.beneficiary_third_party order by bene_sk")
         assert cur.rowcount == 4