pt/pipekit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make pm20100 incremental on MODIFDT

Browse Source 
My earlier reasoning for leaving this full — "open tables delete rows,
so delete-by-key strands them" — does not hold for PM20100. The table
accumulates rather than sheds: 46,841 of its 46,886 rows belong to a
document already in PM30200 (history), and 46,864 also appear in
PM30300. MODIFDT spans 2017-2026 and is populated on every row.

The changed set is self-referential rather than parent-driven: only 42
of 46,886 rows have a parent in PM20000, so a PM20000 join would sync
almost nothing. Instead, stage every apply row belonging to a voucher
that has any row modified since the watermark, and key the merge on
(vchrnmbr, doctype) so the delete replaces the voucher's whole apply
set. The natural grain (vchrnmbr, doctype, aptvchnm, aptodcty) has 42
duplicates and would drop rows if used as the key directly.

7-day lookback stages 60 rows (738 at 30 days) against 46,886 for a
full reload. 84s -> 5s. reconcile.py --super-quick reports IN SYNC.

The watermark guards MAX(modifdt) with LEAST(..., CURRENT_DATE) so a
future-dated row cannot advance it past real data.

Residual risk, same class as icstt: an upstream delete cannot
propagate. Low here given the table demonstrably retains rows whose
documents went historical years ago.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Paul Trowbridge 2026-09-04 00:34:43 -04:00
parent b855ab91d7
commit 855dd1a402
2 changed files with 106 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
config/modules/pm20100.json
View File

@ -4,8 +4,8 @@
"dest_connection": "usmidsap02", "dest_connection": "usmidsap02",
"dest_table": "gp.pm20100", "dest_table": "gp.pm20100",
"staging_table": "pipekit_staging.pm20100", "staging_table": "pipekit_staging.pm20100",
"merge_strategy": "full", "merge_strategy": "incremental",
"merge_key": null, "merge_key": "vchrnmbr, doctype",
"enabled": 1, "enabled": 1,
"dest_description": "Dynamics GP open payables apply detail", "dest_description": "Dynamics GP open payables apply detail",
"columns": [ "columns": [
@ -626,7 +626,14 @@
"description": null "description": null
} }
], ],
"watermarks": [], "watermarks": [
{
"name": "pm20100_ts",
"connection": "usmidsap02",
"resolver_sql": "SELECT COALESCE(LEAST(MAX(modifdt), CURRENT_DATE::timestamp) - INTERVAL '7 days', '1900-01-01') FROM gp.pm20100",
"default_value": "1900-01-01"
}
],
"hooks": [], "hooks": [],
"description": "Dynamics GP open payables apply detail" "description": "Dynamics GP open payables apply detail"
} }

185
config/modules/pm20100.sql
View File

@ -1,90 +1,97 @@
SELECT * FROM OPENQUERY(GPSERVER, '
SELECT SELECT
RTRIM([KEYSOURC]) AS keysourc, RTRIM(d.[KEYSOURC]) AS keysourc,
[DOCDATE] AS docdate, d.[DOCDATE] AS docdate,
[TIME1] AS time1, d.[TIME1] AS time1,
RTRIM([VENDORID]) AS vendorid, RTRIM(d.[VENDORID]) AS vendorid,
[POSTED] AS posted, d.[POSTED] AS posted,
RTRIM([BCHSOURC]) AS bchsourc, RTRIM(d.[BCHSOURC]) AS bchsourc,
RTRIM([USERID]) AS userid, RTRIM(d.[USERID]) AS userid,
RTRIM([APTVCHNM]) AS aptvchnm, RTRIM(d.[APTVCHNM]) AS aptvchnm,
[APTODCTY] AS aptodcty, d.[APTODCTY] AS aptodcty,
[APTODCDT] AS aptodcdt, d.[APTODCDT] AS aptodcdt,
RTRIM([APTODCNM]) AS aptodcnm, RTRIM(d.[APTODCNM]) AS aptodcnm,
[APPLDAMT] AS appldamt, d.[APPLDAMT] AS appldamt,
[DISTKNAM] AS distknam, d.[DISTKNAM] AS distknam,
[DISAVTKN] AS disavtkn, d.[DISAVTKN] AS disavtkn,
[WROFAMNT] AS wrofamnt, d.[WROFAMNT] AS wrofamnt,
RTRIM([CURNCYID]) AS curncyid, RTRIM(d.[CURNCYID]) AS curncyid,
[CURRNIDX] AS currnidx, d.[CURRNIDX] AS currnidx,
[ORAPPAMT] AS orappamt, d.[ORAPPAMT] AS orappamt,
[ORDISTKN] AS ordistkn, d.[ORDISTKN] AS ordistkn,
[ORDATKN] AS ordatkn, d.[ORDATKN] AS ordatkn,
[ORWROFAM] AS orwrofam, d.[ORWROFAM] AS orwrofam,
[Apply_To_Document_Amount] AS apply_to_document_amount, d.[Apply_To_Document_Amount] AS apply_to_document_amount,
[Apply_To_Orig_Doc_Amount] AS apply_to_orig_doc_amount, d.[Apply_To_Orig_Doc_Amount] AS apply_to_orig_doc_amount,
[Apply_To_Discount_Date] AS apply_to_discount_date, d.[Apply_To_Discount_Date] AS apply_to_discount_date,
[Apply_To_Due_Date] AS apply_to_due_date, d.[Apply_To_Due_Date] AS apply_to_due_date,
RTRIM([Apply_To_Description]) AS apply_to_description, RTRIM(d.[Apply_To_Description]) AS apply_to_description,
[Apply_To_IS_MC_Trx] AS apply_to_is_mc_trx, d.[Apply_To_IS_MC_Trx] AS apply_to_is_mc_trx,
[ApplyToGLPostDate] AS applytoglpostdate, d.[ApplyToGLPostDate] AS applytoglpostdate,
[APTOEXRATE] AS aptoexrate, d.[APTOEXRATE] AS aptoexrate,
[APTODENRATE] AS aptodenrate, d.[APTODENRATE] AS aptodenrate,
[APTORTCLCMETH] AS aptortclcmeth, d.[APTORTCLCMETH] AS aptortclcmeth,
[APTOMCTRXSTT] AS aptomctrxstt, d.[APTOMCTRXSTT] AS aptomctrxstt,
[ISMCTRX] AS ismctrx, d.[ISMCTRX] AS ismctrx,
RTRIM([VCHRNMBR]) AS vchrnmbr, RTRIM(d.[VCHRNMBR]) AS vchrnmbr,
[DOCTYPE] AS doctype, d.[DOCTYPE] AS doctype,
RTRIM([APFRDCNM]) AS apfrdcnm, RTRIM(d.[APFRDCNM]) AS apfrdcnm,
RTRIM([FROMCURR]) AS fromcurr, RTRIM(d.[FROMCURR]) AS fromcurr,
[Apply_From_Document_Amou] AS apply_from_document_amou, d.[Apply_From_Document_Amou] AS apply_from_document_amou,
[Apply_From_Orig_Doc_Amou] AS apply_from_orig_doc_amou, d.[Apply_From_Orig_Doc_Amou] AS apply_from_orig_doc_amou,
[APFRMAPLYAMT] AS apfrmaplyamt, d.[APFRMAPLYAMT] AS apfrmaplyamt,
[APFRMDISCTAKEN] AS apfrmdisctaken, d.[APFRMDISCTAKEN] AS apfrmdisctaken,
[APFRMDISCAVAIL] AS apfrmdiscavail, d.[APFRMDISCAVAIL] AS apfrmdiscavail,
[APFRMWROFAMT] AS apfrmwrofamt, d.[APFRMWROFAMT] AS apfrmwrofamt,
[ActualApplyToAmount] AS actualapplytoamount, d.[ActualApplyToAmount] AS actualapplytoamount,
[ActualDiscTakenAmount] AS actualdisctakenamount, d.[ActualDiscTakenAmount] AS actualdisctakenamount,
[ActualDiscAvailTaken] AS actualdiscavailtaken, d.[ActualDiscAvailTaken] AS actualdiscavailtaken,
[ActualWriteOffAmount] AS actualwriteoffamount, d.[ActualWriteOffAmount] AS actualwriteoffamount,
[APFRMEXRATE] AS apfrmexrate, d.[APFRMEXRATE] AS apfrmexrate,
[APFRMDENRATE] AS apfrmdenrate, d.[APFRMDENRATE] AS apfrmdenrate,
[APFRMRTCLCMETH] AS apfrmrtclcmeth, d.[APFRMRTCLCMETH] AS apfrmrtclcmeth,
[APFRMMCTRXSTT] AS apfrmmctrxstt, d.[APFRMMCTRXSTT] AS apfrmmctrxstt,
RTRIM([Apply_From_Description]) AS apply_from_description, RTRIM(d.[Apply_From_Description]) AS apply_from_description,
[Apply_From_IS_MC_Trx] AS apply_from_is_mc_trx, d.[Apply_From_IS_MC_Trx] AS apply_from_is_mc_trx,
[PPSAMDED] AS ppsamded, d.[PPSAMDED] AS ppsamded,
[GSTDSAMT] AS gstdsamt, d.[GSTDSAMT] AS gstdsamt,
RTRIM([TAXDTLID]) AS taxdtlid, RTRIM(d.[TAXDTLID]) AS taxdtlid,
[UPDTKNAM] AS updtknam, d.[UPDTKNAM] AS updtknam,
[UPGSTDAM] AS upgstdam, d.[UPGSTDAM] AS upgstdam,
[UNPPPSAD] AS unpppsad, d.[UNPPPSAD] AS unpppsad,
[UPWROFAM] AS upwrofam, d.[UPWROFAM] AS upwrofam,
[PMDSTMSG] AS pmdstmsg, d.[PMDSTMSG] AS pmdstmsg,
[MODIFDT] AS modifdt, d.[MODIFDT] AS modifdt,
RTRIM([MDFUSRID]) AS mdfusrid, RTRIM(d.[MDFUSRID]) AS mdfusrid,
[POSTEDDT] AS posteddt, d.[POSTEDDT] AS posteddt,
RTRIM([PTDUSRID]) AS ptdusrid, RTRIM(d.[PTDUSRID]) AS ptdusrid,
[TEN99AMNT] AS ten99amnt, d.[TEN99AMNT] AS ten99amnt,
[RLGANLOS] AS rlganlos, d.[RLGANLOS] AS rlganlos,
[APYFRMRNDAMT] AS apyfrmrndamt, d.[APYFRMRNDAMT] AS apyfrmrndamt,
[APYTORNDAMT] AS apytorndamt, d.[APYTORNDAMT] AS apytorndamt,
[APYTORNDDISC] AS apytornddisc, d.[APYTORNDDISC] AS apytornddisc,
[XCHGRATE] AS xchgrate, d.[XCHGRATE] AS xchgrate,
[DECPLCUR] AS decplcur, d.[DECPLCUR] AS decplcur,
[RATECALC] AS ratecalc, d.[RATECALC] AS ratecalc,
RTRIM([Payment_To_Print_On_Stub]) AS payment_to_print_on_stub, RTRIM(d.[Payment_To_Print_On_Stub]) AS payment_to_print_on_stub,
[OAPYFRMRNDAMT] AS oapyfrmrndamt, d.[OAPYFRMRNDAMT] AS oapyfrmrndamt,
[OAPYTORNDAMT] AS oapytorndamt, d.[OAPYTORNDAMT] AS oapytorndamt,
[OAPYTORNDDISC] AS oapytornddisc, d.[OAPYTORNDDISC] AS oapytornddisc,
[Settled_Gain_CreditCurrT] AS settled_gain_creditcurrt, d.[Settled_Gain_CreditCurrT] AS settled_gain_creditcurrt,
[Settled_Loss_CreditCurrT] AS settled_loss_creditcurrt, d.[Settled_Loss_CreditCurrT] AS settled_loss_creditcurrt,
[Settled_Gain_DebitCurrTr] AS settled_gain_debitcurrtr, d.[Settled_Gain_DebitCurrTr] AS settled_gain_debitcurrtr,
[Settled_Loss_DebitCurrTr] AS settled_loss_debitcurrtr, d.[Settled_Loss_DebitCurrTr] AS settled_loss_debitcurrtr,
[Settled_Gain_DebitDiscAv] AS settled_gain_debitdiscav, d.[Settled_Gain_DebitDiscAv] AS settled_gain_debitdiscav,
[Settled_Loss_DebitDiscAv] AS settled_loss_debitdiscav, d.[Settled_Loss_DebitDiscAv] AS settled_loss_debitdiscav,
RTRIM([VADCDTRO]) AS vadcdtro, RTRIM(d.[VADCDTRO]) AS vadcdtro,
[Credit1099Amount] AS credit1099amount, d.[Credit1099Amount] AS credit1099amount,
[DEFTEN99TYPE] AS deften99type, d.[DEFTEN99TYPE] AS deften99type,
[DEFTEN99BOXNUMBER] AS deften99boxnumber, d.[DEFTEN99BOXNUMBER] AS deften99boxnumber,
[DEX_ROW_ID] AS dex_row_id d.[DEX_ROW_ID] AS dex_row_id
FROM [GPSERVER].[CHG].[dbo].[PM20100] FROM CHG.dbo.PM20100 d
WHERE EXISTS (
SELECT 1 FROM CHG.dbo.PM20100 x
WHERE x.VCHRNMBR = d.VCHRNMBR AND x.DOCTYPE = d.DOCTYPE
AND x.[MODIFDT] > ''{pm20100_ts}''
)
')