SyteLine Incremental Extract Query Generator for SSIS and ETL

CREATE TABLE dbo.etl_watermark (
    table_name sysname      NOT NULL,
    site_ref   nvarchar(20) NOT NULL,
    last_value datetime2(3) NOT NULL,
    CONSTRAINT PK_etl_watermark PRIMARY KEY (table_name, site_ref)
);

INSERT INTO dbo.etl_watermark (table_name, site_ref, last_value)
VALUES (N'coitem_mst', N'MAIN', '1900-01-01');

-- @LastValue comes from the watermark table, @RunStart is captured once at the start of the run
DECLARE @Site nvarchar(20) = N'MAIN';
DECLARE @Overlap int = 5;
DECLARE @LastValue datetime2(3) = ?;   -- parameter mapped by SSIS/ADF
DECLARE @RunStart datetime2(3) = ?;    -- parameter mapped by SSIS/ADF

SELECT
    co_num, co_line, co_release, item, qty_ordered, qty_shipped, stat, due_date,
    RowPointer,
    RecordDate,
    CreateDate
FROM dbo.coitem_mst
WHERE site_ref = @Site
  AND RecordDate > DATEADD(minute, -@Overlap, @LastValue)
  AND RecordDate <= @RunStart;

-- Separate batch: map the same @RunStart value captured for block 2 to the ? marker
UPDATE dbo.etl_watermark
SET last_value = ?
WHERE table_name = N'coitem_mst'
  AND site_ref = N'MAIN';

-- Extract on the source replica into dbo.land_coitem_keys:
-- SELECT RowPointer FROM dbo.coitem_mst WHERE site_ref = N'MAIN';

-- Then in the staging database:
DELETE t
FROM dbo.stg_coitem t
WHERE NOT EXISTS (
    SELECT 1
    FROM dbo.land_coitem_keys k
    WHERE k.RowPointer = t.RowPointer
);

MERGE dbo.stg_coitem AS t
USING dbo.land_coitem AS s
    ON t.RowPointer = s.RowPointer
WHEN MATCHED AND s.RecordDate > t.RecordDate THEN
    UPDATE SET t.co_num = s.co_num, t.co_line = s.co_line, t.co_release = s.co_release, t.item = s.item, t.qty_ordered = s.qty_ordered, t.qty_shipped = s.qty_shipped, t.stat = s.stat, t.due_date = s.due_date, t.RecordDate = s.RecordDate
WHEN NOT MATCHED BY TARGET THEN
    INSERT (co_num, co_line, co_release, item, qty_ordered, qty_shipped, stat, due_date, RowPointer, RecordDate)
    VALUES (s.co_num, s.co_line, s.co_release, s.item, s.qty_ordered, s.qty_shipped, s.stat, s.due_date, s.RowPointer, s.RecordDate);

TRUNCATE TABLE dbo.land_coitem;