ERP Incremental Extract Watermark SQL Generator

IF OBJECT_ID('dbo.etl_watermark') IS NULL
BEGIN
  CREATE TABLE dbo.etl_watermark (
    source_table  sysname      NOT NULL PRIMARY KEY,
    last_mod      datetime2(3) NOT NULL,
    last_run_utc  datetime2(3) NOT NULL DEFAULT SYSUTCDATETIME()
  );
END

IF NOT EXISTS (SELECT 1 FROM dbo.etl_watermark WHERE source_table = 'item_transactions')
  INSERT INTO dbo.etl_watermark (source_table, last_mod)
  VALUES ('item_transactions', '1900-01-01');

DECLARE @from datetime2(3);
DECLARE @to   datetime2(3);

SELECT @from = DATEADD(MINUTE, -5, last_mod)
FROM dbo.etl_watermark
WHERE source_table = 'item_transactions';

SELECT TOP (50000) trans_id, item, whse, qty, trans_type, RecordDate
INTO #batch
FROM dbo.item_transactions
WHERE RecordDate > @from
ORDER BY RecordDate, trans_id;

SELECT @to = MAX(RecordDate) FROM #batch;

-- Hand #batch to your loader here, then upsert on trans_id at the destination.
SELECT * FROM #batch;

UPDATE dbo.etl_watermark
SET last_mod = @to,
    last_run_utc = SYSUTCDATETIME()
WHERE source_table = 'item_transactions'
  AND @to IS NOT NULL
  AND @to > last_mod;

DROP TABLE #batch;

SELECT COUNT(*) AS source_rows, MAX(RecordDate) AS source_max_mod
FROM dbo.item_transactions
WHERE RecordDate > '1900-01-01';

-- Compare with the row count and max RecordDate at the destination.