> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mantrixflow.com/llms.txt
> Use this file to discover all available pages before exploring further.

# PostgreSQL incremental sample

> Reproduce an 8,000-row initial load followed by 2,000 inserts and 25 updates.

This sample validates checkpointed Incremental sync and Upsert delivery with a
realistic data volume.

## 1. Create source and destination tables

Run the source DDL in the source database:

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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);
```

Run the destination DDL in the destination database:

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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
);
```

## 2. Seed 8,000 source rows

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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;
```

## 3. Configure the pipeline

1. Create and test PostgreSQL source and destination connections.
2. Create a pipeline with the source connection and open its workspace.
3. In **Source**, discover and enable `mantrix_source.large_dataset`.
4. Set its sync mode to **Incremental**, cursor to `updated_at`, preview, and
   save stream settings.
5. In **Destinations**, add and save the PostgreSQL destination.
6. In **Transformations**, select that destination and create a transformation
   using the source stream.

Use this model:

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
SELECT
  id,
  item_code,
  quantity,
  price,
  status,
  metadata,
  created_at,
  updated_at
FROM {{ source('raw', 'mantrix_source__large_dataset') }}
```

Save the draft, validate, preview, and publish it. Return to the destination
editor, set the output table to `mantrix_destination.large_dataset`, set the
Upsert key to `id`, and save. Validate from **Overview**, click **Run all**, and
inspect **Runs**. The first run should deliver 8,000 rows.

## 4. Apply the incremental delta

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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;
```

Click **Run all** again. The Incremental run should process 2,025 changed rows.

## 5. Verify

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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';
```

Expected results:

* `total_rows`: `10000`
* `new_rows`: `2000`
* `updated_rows`: `25`