Make the three AP history modules incremental

The Accounts Payable group ran ~9.5 min, 83% of it in four full reloads. The cost is the GPSERVER linked-server hop, not row volume (pm30200 managed only ~590 rows/sec), so cutting rows crossing the link is the whole win. Predicates go inside OPENQUERY to push down to the remote — a 4-part name would drag the table across before filtering. pm30200 own DEX_ROW_TS pm30600 changed set: parent PM30200 pm30300 changed set: PM30200 union PM20000 All three key on (vchrnmbr, doctype), verified exactly unique in PM30200 at 108,803/108,803. Watermark resolves off gp.pm30200 with a 7-day lookback, so run order within the group must stay pm30200 first and the lookback must exceed the sync interval. The two-parent union on pm30300 is load-bearing: 38 rows have a parent only in the open table, and a PM30200-only join would strand them permanently. pm30600 needs no union (0 orphans, verified). pm30700 stays full — 213 of its rows have no parent in either table, so no changed-set join can reach them, and it only costs 13s. pm20100 and pm00400 stay full because open and index tables delete rows, which a delete-by-key incremental structurally cannot propagate (cf. icstt). 391s -> 15s. reconcile.py --super-quick reports IN SYNC for all three. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-04 00:21:59 -04:00 · 2026-09-04 00:21:59 -04:00 · b855ab91d7
commit b855ab91d7
parent e8a3acaebd
6 changed files with 142 additions and 99 deletions
--- a/config/modules/pm30200.json
+++ b/config/modules/pm30200.json
@ -4,8 +4,8 @@
  "dest_connection": "usmidsap02",
  "dest_table": "gp.pm30200",
  "staging_table": "pipekit_staging.pm30200",
-  "merge_strategy": "full",
-  "merge_key": null,
+  "merge_strategy": "incremental",
+  "merge_key": "vchrnmbr, doctype",
  "enabled": 1,
  "dest_description": "Dynamics GP payables history",
  "columns": [
@ -556,7 +556,14 @@
      "description": null
    }
  ],
-  "watermarks": [],
+  "watermarks": [
+    {
+      "name": "pm30200_ts",
+      "connection": "usmidsap02",
+      "resolver_sql": "SELECT COALESCE(MAX(dex_row_ts) - INTERVAL '7 days', '1900-01-01') FROM gp.pm30200",
+      "default_value": "1900-01-01"
+    }
+  ],
  "hooks": [],
  "description": "Dynamics GP payables history"
 }
--- a/config/modules/pm30200.sql
+++ b/config/modules/pm30200.sql
@ -1,3 +1,4 @@
+SELECT * FROM OPENQUERY(GPSERVER, '
 SELECT
  RTRIM([VCHRNMBR]) AS vchrnmbr,
  RTRIM([VENDORID]) AS vendorid,
@ -77,4 +78,6 @@ SELECT
  [Workflow_Status] AS workflow_status,
  [DEX_ROW_TS] AS dex_row_ts,
  [DEX_ROW_ID] AS dex_row_id
-FROM [GPSERVER].[CHG].[dbo].[PM30200]
+FROM CHG.dbo.PM30200
+WHERE [DEX_ROW_TS] > ''{pm30200_ts}''
+')
--- a/config/modules/pm30300.json
+++ b/config/modules/pm30300.json
@ -4,8 +4,8 @@
  "dest_connection": "usmidsap02",
  "dest_table": "gp.pm30300",
  "staging_table": "pipekit_staging.pm30300",
-  "merge_strategy": "full",
-  "merge_key": null,
+  "merge_strategy": "incremental",
+  "merge_key": "vchrnmbr, doctype",
  "enabled": 1,
  "dest_description": "Dynamics GP payables apply history",
  "columns": [
@ -458,7 +458,14 @@
      "description": null
    }
  ],
-  "watermarks": [],
+  "watermarks": [
+    {
+      "name": "pm30300_ts",
+      "connection": "usmidsap02",
+      "resolver_sql": "SELECT COALESCE(MAX(dex_row_ts) - INTERVAL '7 days', '1900-01-01') FROM gp.pm30200",
+      "default_value": "1900-01-01"
+    }
+  ],
  "hooks": [],
  "description": "Dynamics GP payables apply history"
 }
--- a/config/modules/pm30300.sql
+++ b/config/modules/pm30300.sql
@ -1,66 +1,78 @@
+SELECT * FROM OPENQUERY(GPSERVER, '
 SELECT
-  RTRIM([VENDORID]) AS vendorid,
-  [DOCDATE] AS docdate,
-  [DATE1] AS date1,
-  [GLPOSTDT] AS glpostdt,
-  [TIME1] AS time1,
-  RTRIM([VCHRNMBR]) AS vchrnmbr,
-  [DOCTYPE] AS doctype,
-  RTRIM([APFRDCNM]) AS apfrdcnm,
-  [ApplyFromGLPostDate] AS applyfromglpostdate,
-  RTRIM([FROMCURR]) AS fromcurr,
-  [APFRMAPLYAMT] AS apfrmaplyamt,
-  [APFRMDISCTAKEN] AS apfrmdisctaken,
-  [APFRMDISCAVAIL] AS apfrmdiscavail,
-  [APFRMWROFAMT] AS apfrmwrofamt,
-  [ActualApplyToAmount] AS actualapplytoamount,
-  [ActualDiscTakenAmount] AS actualdisctakenamount,
-  [ActualDiscAvailTaken] AS actualdiscavailtaken,
-  [ActualWriteOffAmount] AS actualwriteoffamount,
-  [APFRMEXRATE] AS apfrmexrate,
-  [APFRMDENRATE] AS apfrmdenrate,
-  [APFRMRTCLCMETH] AS apfrmrtclcmeth,
-  [APFRMMCTRXSTT] AS apfrmmctrxstt,
-  RTRIM([APTVCHNM]) AS aptvchnm,
-  [APTODCTY] AS aptodcty,
-  RTRIM([APTODCNM]) AS aptodcnm,
-  [APTODCDT] AS aptodcdt,
-  [ApplyToGLPostDate] AS applytoglpostdate,
-  RTRIM([CURNCYID]) AS curncyid,
-  [CURRNIDX] AS currnidx,
-  [APPLDAMT] AS appldamt,
-  [DISTKNAM] AS distknam,
-  [DISAVTKN] AS disavtkn,
-  [WROFAMNT] AS wrofamnt,
-  [ORAPPAMT] AS orappamt,
-  [ORDISTKN] AS ordistkn,
-  [ORDATKN] AS ordatkn,
-  [ORWROFAM] AS orwrofam,
-  [APTOEXRATE] AS aptoexrate,
-  [APTODENRATE] AS aptodenrate,
-  [APTORTCLCMETH] AS aptortclcmeth,
-  [APTOMCTRXSTT] AS aptomctrxstt,
-  [PPSAMDED] AS ppsamded,
-  [GSTDSAMT] AS gstdsamt,
-  RTRIM([TAXDTLID]) AS taxdtlid,
-  [POSTED] AS posted,
-  [TEN99AMNT] AS ten99amnt,
-  [RLGANLOS] AS rlganlos,
-  [APYFRMRNDAMT] AS apyfrmrndamt,
-  [APYTORNDAMT] AS apytorndamt,
-  [APYTORNDDISC] AS apytornddisc,
-  [OAPYFRMRNDAMT] AS oapyfrmrndamt,
-  [OAPYTORNDAMT] AS oapytorndamt,
-  [OAPYTORNDDISC] AS oapytornddisc,
-  [Settled_Gain_CreditCurrT] AS settled_gain_creditcurrt,
-  [Settled_Loss_CreditCurrT] AS settled_loss_creditcurrt,
-  [Settled_Gain_DebitCurrTr] AS settled_gain_debitcurrtr,
-  [Settled_Loss_DebitCurrTr] AS settled_loss_debitcurrtr,
-  [Settled_Gain_DebitDiscAv] AS settled_gain_debitdiscav,
-  [Settled_Loss_DebitDiscAv] AS settled_loss_debitdiscav,
-  [Revaluation_Status] AS revaluation_status,
-  [Credit1099Amount] AS credit1099amount,
-  [DEFTEN99TYPE] AS deften99type,
-  [DEFTEN99BOXNUMBER] AS deften99boxnumber,
-  [DEX_ROW_ID] AS dex_row_id
-FROM [GPSERVER].[CHG].[dbo].[PM30300]
+  RTRIM(d.[VENDORID]) AS vendorid,
+  d.[DOCDATE] AS docdate,
+  d.[DATE1] AS date1,
+  d.[GLPOSTDT] AS glpostdt,
+  d.[TIME1] AS time1,
+  RTRIM(d.[VCHRNMBR]) AS vchrnmbr,
+  d.[DOCTYPE] AS doctype,
+  RTRIM(d.[APFRDCNM]) AS apfrdcnm,
+  d.[ApplyFromGLPostDate] AS applyfromglpostdate,
+  RTRIM(d.[FROMCURR]) AS fromcurr,
+  d.[APFRMAPLYAMT] AS apfrmaplyamt,
+  d.[APFRMDISCTAKEN] AS apfrmdisctaken,
+  d.[APFRMDISCAVAIL] AS apfrmdiscavail,
+  d.[APFRMWROFAMT] AS apfrmwrofamt,
+  d.[ActualApplyToAmount] AS actualapplytoamount,
+  d.[ActualDiscTakenAmount] AS actualdisctakenamount,
+  d.[ActualDiscAvailTaken] AS actualdiscavailtaken,
+  d.[ActualWriteOffAmount] AS actualwriteoffamount,
+  d.[APFRMEXRATE] AS apfrmexrate,
+  d.[APFRMDENRATE] AS apfrmdenrate,
+  d.[APFRMRTCLCMETH] AS apfrmrtclcmeth,
+  d.[APFRMMCTRXSTT] AS apfrmmctrxstt,
+  RTRIM(d.[APTVCHNM]) AS aptvchnm,
+  d.[APTODCTY] AS aptodcty,
+  RTRIM(d.[APTODCNM]) AS aptodcnm,
+  d.[APTODCDT] AS aptodcdt,
+  d.[ApplyToGLPostDate] AS applytoglpostdate,
+  RTRIM(d.[CURNCYID]) AS curncyid,
+  d.[CURRNIDX] AS currnidx,
+  d.[APPLDAMT] AS appldamt,
+  d.[DISTKNAM] AS distknam,
+  d.[DISAVTKN] AS disavtkn,
+  d.[WROFAMNT] AS wrofamnt,
+  d.[ORAPPAMT] AS orappamt,
+  d.[ORDISTKN] AS ordistkn,
+  d.[ORDATKN] AS ordatkn,
+  d.[ORWROFAM] AS orwrofam,
+  d.[APTOEXRATE] AS aptoexrate,
+  d.[APTODENRATE] AS aptodenrate,
+  d.[APTORTCLCMETH] AS aptortclcmeth,
+  d.[APTOMCTRXSTT] AS aptomctrxstt,
+  d.[PPSAMDED] AS ppsamded,
+  d.[GSTDSAMT] AS gstdsamt,
+  RTRIM(d.[TAXDTLID]) AS taxdtlid,
+  d.[POSTED] AS posted,
+  d.[TEN99AMNT] AS ten99amnt,
+  d.[RLGANLOS] AS rlganlos,
+  d.[APYFRMRNDAMT] AS apyfrmrndamt,
+  d.[APYTORNDAMT] AS apytorndamt,
+  d.[APYTORNDDISC] AS apytornddisc,
+  d.[OAPYFRMRNDAMT] AS oapyfrmrndamt,
+  d.[OAPYTORNDAMT] AS oapytorndamt,
+  d.[OAPYTORNDDISC] AS oapytornddisc,
+  d.[Settled_Gain_CreditCurrT] AS settled_gain_creditcurrt,
+  d.[Settled_Loss_CreditCurrT] AS settled_loss_creditcurrt,
+  d.[Settled_Gain_DebitCurrTr] AS settled_gain_debitcurrtr,
+  d.[Settled_Loss_DebitCurrTr] AS settled_loss_debitcurrtr,
+  d.[Settled_Gain_DebitDiscAv] AS settled_gain_debitdiscav,
+  d.[Settled_Loss_DebitDiscAv] AS settled_loss_debitdiscav,
+  d.[Revaluation_Status] AS revaluation_status,
+  d.[Credit1099Amount] AS credit1099amount,
+  d.[DEFTEN99TYPE] AS deften99type,
+  d.[DEFTEN99BOXNUMBER] AS deften99boxnumber,
+  d.[DEX_ROW_ID] AS dex_row_id
+FROM CHG.dbo.PM30300 d
+WHERE EXISTS (
+  SELECT 1 FROM CHG.dbo.PM30200 p
+  WHERE p.VCHRNMBR = d.VCHRNMBR AND p.DOCTYPE = d.DOCTYPE
+    AND p.[DEX_ROW_TS] > ''{pm30300_ts}''
+)
+OR EXISTS (
+  SELECT 1 FROM CHG.dbo.PM20000 o
+  WHERE o.VCHRNMBR = d.VCHRNMBR AND o.DOCTYPE = d.DOCTYPE
+    AND o.[DEX_ROW_TS] > ''{pm30300_ts}''
+)
+')
--- a/config/modules/pm30600.json
+++ b/config/modules/pm30600.json
@ -4,8 +4,8 @@
  "dest_connection": "usmidsap02",
  "dest_table": "gp.pm30600",
  "staging_table": "pipekit_staging.pm30600",
-  "merge_strategy": "full",
-  "merge_key": null,
+  "merge_strategy": "incremental",
+  "merge_key": "vchrnmbr, doctype",
  "enabled": 1,
  "dest_description": "Dynamics GP payables GL distribution history",
  "columns": [
@ -171,7 +171,14 @@
      "description": null
    }
  ],
-  "watermarks": [],
+  "watermarks": [
+    {
+      "name": "pm30600_ts",
+      "connection": "usmidsap02",
+      "resolver_sql": "SELECT COALESCE(MAX(dex_row_ts) - INTERVAL '7 days', '1900-01-01') FROM gp.pm30200",
+      "default_value": "1900-01-01"
+    }
+  ],
  "hooks": [],
  "description": "Dynamics GP payables GL distribution history"
 }
--- a/config/modules/pm30600.sql
+++ b/config/modules/pm30600.sql
@ -1,25 +1,32 @@
+SELECT * FROM OPENQUERY(GPSERVER, '
 SELECT
-  [DOCTYPE] AS doctype,
-  RTRIM([VCHRNMBR]) AS vchrnmbr,
-  [DSTSQNUM] AS dstsqnum,
-  [CNTRLTYP] AS cntrltyp,
-  [CRDTAMNT] AS crdtamnt,
-  [DEBITAMT] AS debitamt,
-  [DSTINDX] AS dstindx,
-  [DISTTYPE] AS disttype,
-  [CHANGED] AS changed,
-  RTRIM([USERID]) AS userid,
-  [PSTGSTUS] AS pstgstus,
-  RTRIM([VENDORID]) AS vendorid,
-  RTRIM([TRXSORCE]) AS trxsorce,
-  [PSTGDATE] AS pstgdate,
-  RTRIM([CURNCYID]) AS curncyid,
-  [CURRNIDX] AS currnidx,
-  [ORCRDAMT] AS orcrdamt,
-  [ORDBTAMT] AS ordbtamt,
-  RTRIM([APTVCHNM]) AS aptvchnm,
-  [APTODCTY] AS aptodcty,
-  [SPCLDIST] AS spcldist,
-  RTRIM([DistRef]) AS distref,
-  [DEX_ROW_ID] AS dex_row_id
-FROM [GPSERVER].[CHG].[dbo].[PM30600]
+  d.[DOCTYPE] AS doctype,
+  RTRIM(d.[VCHRNMBR]) AS vchrnmbr,
+  d.[DSTSQNUM] AS dstsqnum,
+  d.[CNTRLTYP] AS cntrltyp,
+  d.[CRDTAMNT] AS crdtamnt,
+  d.[DEBITAMT] AS debitamt,
+  d.[DSTINDX] AS dstindx,
+  d.[DISTTYPE] AS disttype,
+  d.[CHANGED] AS changed,
+  RTRIM(d.[USERID]) AS userid,
+  d.[PSTGSTUS] AS pstgstus,
+  RTRIM(d.[VENDORID]) AS vendorid,
+  RTRIM(d.[TRXSORCE]) AS trxsorce,
+  d.[PSTGDATE] AS pstgdate,
+  RTRIM(d.[CURNCYID]) AS curncyid,
+  d.[CURRNIDX] AS currnidx,
+  d.[ORCRDAMT] AS orcrdamt,
+  d.[ORDBTAMT] AS ordbtamt,
+  RTRIM(d.[APTVCHNM]) AS aptvchnm,
+  d.[APTODCTY] AS aptodcty,
+  d.[SPCLDIST] AS spcldist,
+  RTRIM(d.[DistRef]) AS distref,
+  d.[DEX_ROW_ID] AS dex_row_id
+FROM CHG.dbo.PM30600 d
+WHERE EXISTS (
+  SELECT 1 FROM CHG.dbo.PM30200 p
+  WHERE p.VCHRNMBR = d.VCHRNMBR AND p.DOCTYPE = d.DOCTYPE
+    AND p.[DEX_ROW_TS] > ''{pm30600_ts}''
+)
+')