PostgreSQL incremental sample

CREATE SCHEMA IF NOT EXISTS mantrix_source;

CREATE TABLE mantrix_source.large_dataset (
  id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  item_code text UNIQUE NOT NULL,
  quantity integer NOT NULL,
  price numeric(12, 2) NOT NULL,
  status text NOT NULL,
  metadata jsonb,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL
);

CREATE INDEX large_dataset_updated_at_idx
  ON mantrix_source.large_dataset (updated_at);

CREATE SCHEMA IF NOT EXISTS mantrix_destination;

CREATE TABLE mantrix_destination.large_dataset (
  id bigint PRIMARY KEY,
  item_code text UNIQUE NOT NULL,
  quantity integer NOT NULL,
  price numeric(12, 2) NOT NULL,
  status text NOT NULL,
  metadata jsonb,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL
);

INSERT INTO mantrix_source.large_dataset (
  item_code, quantity, price, status, metadata, created_at, updated_at
)
SELECT
  'ITEM-' || lpad(g::text, 6, '0'),
  (g % 100) + 1,
  (g * 1.25)::numeric(12, 2),
  CASE WHEN g % 7 = 0 THEN 'backorder' ELSE 'available' END,
  jsonb_build_object('batch', 'initial-8000', 'ordinal', g),
  '2026-07-01 00:00:00+00'::timestamptz + g * interval '1 second',
  '2026-07-01 00:00:00+00'::timestamptz + g * interval '1 second'
FROM generate_series(1, 8000) AS g;

SELECT
  id,
  item_code,
  quantity,
  price,
  status,
  metadata,
  created_at,
  updated_at
FROM {{ source('raw', 'mantrix_source__large_dataset') }}

INSERT INTO mantrix_source.large_dataset (
  item_code, quantity, price, status, metadata, created_at, updated_at
)
SELECT
  'ITEM-' || lpad(g::text, 6, '0'),
  (g % 100) + 1,
  (g * 1.25)::numeric(12, 2),
  'available',
  jsonb_build_object('batch', 'incremental-2000', 'ordinal', g),
  '2026-07-02 00:00:00+00'::timestamptz + (g - 8000) * interval '1 second',
  '2026-07-02 00:00:00+00'::timestamptz + (g - 8000) * interval '1 second'
FROM generate_series(8001, 10000) AS g;

UPDATE mantrix_source.large_dataset
SET
  price = price + 10,
  status = 'updated',
  metadata = metadata || '{"changed_in_delta":true}'::jsonb,
  updated_at = '2026-07-02 01:00:00+00'
WHERE id BETWEEN 1 AND 25;

SELECT count(*) AS total_rows
FROM mantrix_destination.large_dataset;

SELECT count(*) AS new_rows
FROM mantrix_destination.large_dataset
WHERE id > 8000;

SELECT count(*) AS updated_rows
FROM mantrix_destination.large_dataset
WHERE id BETWEEN 1 AND 25
  AND status = 'updated';