pg_stream

pg_stream is a PostgreSQL 18 extension that turns ordinary SQL views into self-maintaining stream tables — no external processes, no sidecars, no bespoke refresh pipelines. Just CREATE EXTENSION pg_stream and your views stay fresh.

-- Declare a stream table — a view that maintains itself
SELECT pgstream.create_stream_table(
    'active_orders',
    'SELECT * FROM orders WHERE status = ''active''',
    '30s'
);

-- Insert a row — the stream table updates automatically on the next refresh
INSERT INTO orders (id, status) VALUES (42, 'active');
SELECT count(*) FROM active_orders;  -- 1

The problem with materialized views

PostgreSQL's materialized views are powerful but frustrating. REFRESH MATERIALIZED VIEW re-runs the entire query from scratch, even if only one row changed in a million-row table. Your choices are: burn CPU on full recomputation, or accept stale data. Most teams end up building bespoke refresh pipelines just to keep summary tables current.

What pg_stream does differently

pg_stream captures changes to your source tables and — on each refresh cycle — derives a delta query that processes only the changed rows and merges the result into the materialized table. One insert into a million-row source table? pg_stream touches exactly one row's worth of computation.

The approach is grounded in the DBSP differential dataflow framework (Budiu et al., 2022). Delta queries are derived automatically from your SQL's operator tree: joins produce the classic bilinear expansion, aggregates maintain auxiliary counters, and linear operators like filters pass deltas through unchanged.

Key capabilities

Demand-driven scheduling

With the default CALCULATED schedule mode, you only set an explicit refresh interval on the stream tables your application actually queries. The system propagates that cadence upward through the dependency graph: each upstream stream table inherits the tightest schedule among its downstream dependents. You declare freshness requirements where they matter — at the consumer — and the entire pipeline adjusts without manual coordination.

Hybrid change capture

pg_stream bootstraps with lightweight row-level triggers — no configuration needed, works out of the box. Once the first refresh succeeds and wal_level = logical is available, the system automatically transitions to WAL-based logical replication for lower write-side overhead. The transition is seamless: trigger → transitioning → WAL-only. If anything goes wrong, it falls back to triggers.

Explore this documentation

• Getting Started — build a three-layer DAG from scratch in minutes
• SQL Reference — every function and option
• Architecture — how the engine works internally
• Configuration — GUC variables and tuning

Source & releases

Written in Rust using pgrx. Targets PostgreSQL 18. Apache 2.0 licensed.

• Repository: github.com/grove/pg-stream
• Install instructions: Installation
• Changelog: Changelog
• Roadmap: Roadmap

Getting Started with pg_stream

What is pg_stream?

pg_stream adds stream tables to PostgreSQL — tables that are defined by a SQL query and kept automatically up to date as the underlying data changes. Think of them as materialized views that refresh themselves, but smarter: instead of re-running the entire query on every refresh, pg_stream uses Incremental View Maintenance (IVM) to process only the rows that changed.

Traditional materialized views force a choice: either re-run the full query (expensive) or accept stale data. pg_stream eliminates this trade-off. When you insert a single row into a million-row table, pg_stream computes the effect of that one row on the query result — it doesn't touch the other 999,999.

How data flows

The key concept is that data flows downstream automatically — from your base tables through any chain of stream tables, without you writing a single line of orchestration code:

  You write to base tables
         │
         ▼
  ┌─────────────┐   triggers (or WAL)   ┌─────────────────────┐
  │ Base Tables │ ─────────────────────▶ │   Change Buffers    │
  │ (you write) │                        │ (pgstream_changes.*) │
  └─────────────┘                        └──────────┬──────────┘
                                                     │
                                           delta query (ΔQ) on refresh
                                                     │
                                                     ▼
  ┌──────────────────────────────────────────────────────────────┐
  │  Stream Table A  ◀── depends on base tables                  │
  └──────────────────────────┬───────────────────────────────────┘
                             │  change captured, buffer written
                             ▼
  ┌──────────────────────────────────────────────────────────────┐
  │  Stream Table B  ◀── depends on Stream Table A               │
  └──────────────────────────────────────────────────────────────┘

One write to a base table can ripple through an entire DAG of stream tables — each layer refreshed in the correct topological order, each doing only the work proportional to what actually changed.

1. You write to your base tables normally — INSERT, UPDATE, DELETE
2. Lightweight AFTER row-level triggers capture each change into a buffer, atomically in the same transaction. No polling, no logical replication slots required by default.
3. On each refresh cycle, pg_stream derives a delta query (ΔQ) that reads only the buffered changes since the last refresh frontier
4. The delta is merged into the stream table — only the affected rows are written
5. If other stream tables depend on this one, they are scheduled next (topological order)
6. Optionally: once wal_level = logical is available and the first refresh succeeds, pg_stream automatically transitions from triggers to WAL-based CDC (~2–15 μs write overhead vs triggers). The transition is seamless and transparent.

This tutorial walks through a concrete org-chart example so you can see this flow end to end, including a chain of stream tables that propagates changes automatically.

What you'll build

An employee org-chart system with two stream tables:

• department_tree — a recursive CTE that flattens a department hierarchy into paths like Company > Engineering > Backend
• department_stats — a join + aggregation over department_tree (a stream table!) that computes headcount and salary budget, with the full path included
• department_report — a further aggregation that rolls up stats to top-level departments

The chain departments → department_tree → department_stats → department_report demonstrates automatic downstream propagation: modify a department name in the base table and all three stream tables update automatically, in the right order, without any manual orchestration.

By the end you will have:

• Seen how stream tables are created, queried, and refreshed
• Watched a single UPDATE in a base table cascade through three layers of stream tables automatically
• Understood the two IVM strategies and when each applies

Prerequisites

• PostgreSQL 18.x with pg_stream installed (see INSTALL.md)
• shared_preload_libraries = 'pg_stream' in postgresql.conf
• psql or any SQL client

Step 1: Create the Base Tables

These are ordinary PostgreSQL tables — pg_stream doesn't require any special column types, annotations, or schema conventions. The only requirement is that tables have a primary key (pg_stream uses it internally to track which rows changed).

-- Department hierarchy (self-referencing tree)
CREATE TABLE departments (
    id         SERIAL PRIMARY KEY,
    name       TEXT NOT NULL,
    parent_id  INT REFERENCES departments(id)
);

-- Employees belong to a department
CREATE TABLE employees (
    id            SERIAL PRIMARY KEY,
    name          TEXT NOT NULL,
    department_id INT NOT NULL REFERENCES departments(id),
    salary        NUMERIC(10,2) NOT NULL
);

Now insert some data — a three-level department tree and a handful of employees:

-- Top-level
INSERT INTO departments (id, name, parent_id) VALUES
    (1, 'Company',     NULL);

-- Second level
INSERT INTO departments (id, name, parent_id) VALUES
    (2, 'Engineering', 1),
    (3, 'Sales',       1),
    (4, 'Operations',  1);

-- Third level (under Engineering)
INSERT INTO departments (id, name, parent_id) VALUES
    (5, 'Backend',     2),
    (6, 'Frontend',    2),
    (7, 'Platform',    2);

-- Employees
INSERT INTO employees (name, department_id, salary) VALUES
    ('Alice',   5, 120000),   -- Backend
    ('Bob',     5, 115000),   -- Backend
    ('Charlie', 6, 110000),   -- Frontend
    ('Diana',   7, 130000),   -- Platform
    ('Eve',     3, 95000),    -- Sales
    ('Frank',   3, 90000),    -- Sales
    ('Grace',   4, 100000);   -- Operations

At this point these are plain tables with no triggers, no change tracking, nothing special. The department tree looks like this:

Company (1)
├── Engineering (2)
│   ├── Backend (5)     — Alice, Bob
│   ├── Frontend (6)    — Charlie
│   └── Platform (7)    — Diana
├── Sales (3)           — Eve, Frank
└── Operations (4)      — Grace

Step 2: Create the First Stream Table — Recursive Hierarchy

Our first stream table flattens the department tree. For every department, it computes the full path from the root and the depth level. This uses WITH RECURSIVE — a SQL construct that can't be differentiated with simple algebraic rules (the recursion depends on itself), but pg_stream handles it using incremental strategies (semi-naive evaluation for inserts, Delete-and-Rederive for mixed changes) that we'll explain later.

SELECT pgstream.create_stream_table(
    'department_tree',
    $$
    WITH RECURSIVE tree AS (
        -- Base case: root departments (no parent)
        SELECT id, name, parent_id, name AS path, 0 AS depth
        FROM departments
        WHERE parent_id IS NULL

        UNION ALL

        -- Recursive step: children join back to the tree
        SELECT d.id, d.name, d.parent_id,
               tree.path || ' > ' || d.name AS path,
               tree.depth + 1
        FROM departments d
        JOIN tree ON d.parent_id = tree.id
    )
    SELECT id, name, parent_id, path, depth FROM tree
    $$,
    '30s',
    'DIFFERENTIAL'
);

What just happened?

That single function call did a lot of work atomically (all in one transaction):

1. Parsed the defining query into an operator tree — identifying the recursive CTE, the scan on departments, the join, the union
2. Created a storage table called department_tree in the public schema — a real PostgreSQL heap table with columns matching the SELECT output, plus internal columns __pgs_row_id (a hash used to track individual rows)
3. Installed CDC triggers on the departments table — lightweight AFTER INSERT OR UPDATE OR DELETE row-level triggers that will capture every future change
4. Created a change buffer table in the pgstream_changes schema — this is where the triggers write captured changes
5. Ran an initial full refresh — executed the recursive query against the current data and populated the storage table
6. Registered the stream table in pg_stream's catalog with a 30-second refresh schedule

Query it immediately — it's already populated:

SELECT * FROM department_tree ORDER BY path;

Expected output:

 id |    name     | parent_id |            path             | depth
----+-------------+-----------+-----------------------------+-------
  1 | Company     |           | Company                     |     0
  2 | Engineering |         1 | Company > Engineering       |     1
  5 | Backend     |         2 | Company > Engineering > Backend  | 2
  6 | Frontend    |         2 | Company > Engineering > Frontend | 2
  7 | Platform    |         2 | Company > Engineering > Platform | 2
  4 | Operations  |         1 | Company > Operations        |     1
  3 | Sales       |         1 | Company > Sales             |     1

This is a real PostgreSQL table — you can create indexes on it, join it in other queries, reference it in views, or even use it as a source for other stream tables. pg_stream keeps it in sync automatically.

Key insight: The recursive query that computes paths and depths would normally need to be re-run manually (or via REFRESH MATERIALIZED VIEW). With pg_stream, it stays fresh — any change to the departments table is automatically reflected within the schedule bound (30 seconds here).

Step 3: Chain Stream Tables — Build the Downstream Layers

Now create department_stats. The twist: instead of joining directly against departments, it joins against department_tree — the stream table we just created. This creates a chain: changes to departments update department_tree, whose changes then trigger department_stats to update.

This demonstrates how pg_stream builds a DAG — a directed acyclic graph of stream tables — and automatically schedules refreshes in topological order.

SELECT pgstream.create_stream_table(
    'department_stats',
    $$
    SELECT
        t.id          AS department_id,
        t.name        AS department_name,
        t.path        AS full_path,
        t.depth,
        COUNT(e.id)                    AS headcount,
        COALESCE(SUM(e.salary), 0)     AS total_salary,
        COALESCE(AVG(e.salary), 0)     AS avg_salary
    FROM department_tree t
    LEFT JOIN employees e ON e.department_id = t.id
    GROUP BY t.id, t.name, t.path, t.depth
    $$,
    'CALCULATED',       -- inherit schedule from downstream; see explanation below
    'DIFFERENTIAL'
);

What just happened — and why this one is different?

Like before, pg_stream parsed the query, created a storage table, and set up CDC. But department_stats depends on department_tree, not a base table — so no new triggers were installed. Instead, pg_stream registered department_tree as an upstream dependency in the DAG.

The schedule is 'CALCULATED', which means: "don't give this table its own schedule — inherit the tightest schedule of any downstream table that queries it". Since no other stream table has been created yet, PostgreSQL will prompt a full refresh on demand for now.

The query has no recursive CTE, so pg_stream uses algebraic differentiation:

1. Decomposed into operators: Scan(department_tree) → LEFT JOIN → Scan(employees) → Aggregate(GROUP BY + COUNT/SUM/AVG) → Project
2. Derived a differentiation rule for each:
   • Δ(Scan) = read only change buffer rows (not the full table)
   • Δ(LEFT JOIN) = join change rows from one side against the full other side
   • Δ(Aggregate) = for COUNT/SUM/AVG, add or subtract per group — no rescan needed
3. Composed these into a single delta query (ΔQ) that never touches unchanged rows

When one employee is inserted, the refresh reads one change buffer row, joins to find the department, and adjusts only that group's count and sum.

Query it:

SELECT department_name, full_path, headcount, total_salary
FROM department_stats
ORDER BY full_path;

Expected output:

 department_name |                 full_path                  | headcount | total_salary
-----------------+--------------------------------------------+-----------+--------------
 Backend         | Company > Engineering > Backend            |         2 |    235000.00
 Frontend        | Company > Engineering > Frontend           |         1 |    110000.00
 Platform        | Company > Engineering > Platform           |         1 |    130000.00
 Engineering     | Company > Engineering                      |         0 |         0.00
 Operations      | Company > Operations                       |         1 |    100000.00
 Sales           | Company > Sales                            |         2 |    185000.00
 Company         | Company                                    |         0 |         0.00

Notice that the full_path column comes from department_tree — this data already went through one layer of incremental maintenance before landing here.

Add a third layer: department_report

Now add a rollup that aggregates department_stats by top-level group (depth = 1):

SELECT pgstream.create_stream_table(
    'department_report',
    $$
    SELECT
        split_part(full_path, ' > ', 2) AS division,
        SUM(headcount)                  AS total_headcount,
        SUM(total_salary)               AS total_payroll
    FROM department_stats
    WHERE depth >= 1
    GROUP BY 1
    $$,
    '30s',            -- this is the only explicit schedule; CALCULATED tables above inherit it
    'DIFFERENTIAL'
);

The DAG is now:

departments (base)  employees (base)
      │                   │
      ▼                   │
department_tree ──────────┤
   (DIFF, CALCULATED)     │
      │                   ▼
      └──────▶ department_stats
                 (DIFF, CALCULATED)
                      │
                      ▼
               department_report
                  (DIFF, 30s)  ◀── only explicit schedule

department_report drives the whole pipeline. Because it has a 30-second schedule, pg_stream automatically propagates that cadence upstream: department_stats and department_tree will also be refreshed within 30 seconds of a base table change, in topological order, with no manual configuration.

Query the report:

SELECT * FROM department_report ORDER BY division;

  division   | total_headcount | total_payroll
-------------+-----------------+---------------
 Engineering |               4 |    475000.00
 Operations  |               1 |    100000.00
 Sales       |               2 |    185000.00

Step 4: Watch a Change Cascade Through All Three Layers

This is the heart of pg_stream. We'll make four changes to the base tables and watch changes propagate automatically through the three-layer DAG — each layer doing only the minimum work.

The data flow pipeline (three layers)

  Your SQL statement
       │
       ▼
  CDC trigger fires (same transaction)
  Change buffer receives one row
       │
       ▼
  Background scheduler fires (or manual refresh_stream_table)
       │
       ├──▶ [Layer 1] Refresh department_tree
       │         delta query reads change buffer
       │         MERGE touches only affected rows in department_tree
       │         department_tree's own change buffer is updated
       │
       ├──▶ [Layer 2] Refresh department_stats
       │         delta query reads department_tree's change buffer
       │         MERGE touches only affected department groups
       │
       └──▶ [Layer 3] Refresh department_report
                 delta query reads department_stats' change buffer
                 MERGE touches only affected division rows
                 All change buffers cleaned up ✓

All three layers run in a single scheduled pass, in topological order.

4a: A single INSERT ripples through all three layers

INSERT INTO employees (name, department_id, salary) VALUES
    ('Heidi', 6, 105000);  -- New Frontend engineer

What happened immediately (in your transaction): The AFTER INSERT trigger on employees fired and wrote one row to pgstream_changes.changes_<employees_oid>. The row contains the new values, action type I, and the LSN at the time of insert. Your transaction committed normally — no blocking.

The stream tables don't know about Heidi yet. The change is in the buffer, waiting for the next refresh.

Refresh the whole pipeline in one call (or wait for the 30-second schedule):

-- refresh_stream_table cascades to dependent tables automatically
SELECT pgstream.refresh_stream_table('department_report');

What happened across the three layers:

Check the result:

SELECT department_name, headcount, total_salary FROM department_stats
WHERE department_name = 'Frontend';

 department_name | headcount | total_salary
-----------------+-----------+--------------
 Frontend        |         2 |    215000.00

The 6 other groups in department_stats were not touched at all.

Contrast with a standard materialized view: REFRESH MATERIALIZED VIEW would re-scan all 8 employees, re-join with all 7 departments, re-aggregate, and update all 7 rows. With pg_stream, the work was proportional to the 1 changed row — across all three layers.

4b: A department change cascades through the whole DAG

Now change the departments table — the root of the entire chain:

INSERT INTO departments (id, name, parent_id) VALUES
    (8, 'DevOps', 2);  -- New team under Engineering

What happened: The CDC trigger on departments fired. The change buffer for departments has one new row. None of the stream tables know about it yet.

Refresh the whole pipeline:

SELECT pgstream.refresh_stream_table('department_report');

What happened across all three layers:

How the recursive CTE refresh works — unlike department_stats, recursive CTEs can't be algebraically differentiated (the recursion references itself). pg_stream uses incremental fixpoint strategies:

• INSERT → semi-naive evaluation: differentiate the base case, propagate the delta through the recursive term, stopping when no new rows are produced. Only new rows inserted.
• DELETE or UPDATE → Delete-and-Rederive (DRed): remove rows derived from deleted facts, re-derive rows that may have alternative derivation paths, handle cascades cleanly.

SELECT id, name, depth, path FROM department_tree WHERE name = 'DevOps';

 id |  name  | depth |           path                |
----+--------+-------+-------------------------------+
  8 | DevOps |         2 | Company > Engineering > DevOps |    2

The recursive CTE automatically expanded to include the new department at the correct depth and path. One inserted row in departments produced one new row in the stream table.

4c: UPDATE — A single rename that cascades everywhere

Rename "Engineering" to "R&D":

UPDATE departments SET name = 'R&D' WHERE id = 2;

What happened in the change buffer: The CDC trigger captured the old row (name='Engineering') and the new row (name='R&D'). Both old and new values are stored so the delta can compute what to remove and what to add.

Refresh:

SELECT pgstream.refresh_stream_table('department_report');

What happened:

Query to verify the cascade:

SELECT name, path FROM department_tree WHERE path LIKE '%R&D%' ORDER BY depth;

  name   |              path
---------+----------------------------------
 R&D     | Company > R&D
 Backend  | Company > R&D > Backend
 DevOps  | Company > R&D > DevOps
 Frontend | Company > R&D > Frontend
 Platform | Company > R&D > Platform

One UPDATE to a department name flowed through all three layers automatically — updating 5 + 5 + 2 rows across the chain.

4d: DELETE — Remove an employee

DELETE FROM employees WHERE name = 'Bob';

What happened: The AFTER DELETE trigger on employees fired, writing a change buffer row with action type D and Bob's old values (department_id=5, salary=115000). The delta query will use these old values to compute the correct aggregate adjustment — it knows to subtract 115000 from Backend's salary sum and decrement the count.

SELECT pgstream.refresh_stream_table('department_stats');
SELECT * FROM department_stats WHERE department_name = 'Backend';

 department_id | department_name | headcount | total_salary | avg_salary
---------------+-----------------+-----------+--------------+------------
             5 | Backend         |         1 |    120000.00 |  120000.00

Headcount dropped from 2 → 1 and the salary aggregates updated. Again, only the Backend group was touched — the other 6 department rows were untouched.

Step 5: Automatic Scheduling — Let the DAG Drive Itself

In the examples above, we called refresh_stream_table() manually. In production you never need to do this. pg_stream runs a background scheduler that automatically refreshes stale tables — in topological order.

How schedules propagate

We gave department_report a '30s' schedule and the two upstream tables a 'CALCULATED' schedule. This is the recommended pattern:

 department_tree    (CALCULATED → inherits 30s from downstream)
       │
 department_stats   (CALCULATED → inherits 30s from downstream)
       │
 department_report  (30s — the only explicit schedule)

CALCULATED means: compute the tightest schedule across all downstream dependents. You declare freshness requirements at the tables your application queries — the system figures out how often each upstream table needs to refresh.

What the scheduler does every second

1. Queries the catalog for stream tables past their freshness bound
2. Sorts them topologically (upstream first) — department_tree refreshes before department_stats, which refreshes before department_report
3. Runs each refresh (respecting pg_stream.max_concurrent_refreshes)
4. Updates the last-refresh frontier

Monitoring

-- Current status of all stream tables
SELECT table_name, schedule, last_refresh_at, stale, refresh_mode
FROM pgstream.pgs_status();

    table_name      | schedule  | last_refresh_at         | stale | refresh_mode
--------------------+-----------+-------------------------+-------+--------------
 department_tree    | 30s       | 2026-02-26 10:30:00.123 | f     | DIFFERENTIAL
 department_stats   | 30s       | 2026-02-26 10:30:00.456 | f     | DIFFERENTIAL
 department_report  | 30s       | 2026-02-26 10:30:00.789 | f     | DIFFERENTIAL

-- Detailed performance stats
SELECT table_name, refresh_count, avg_refresh_ms, rows_affected_last
FROM pgstream.pg_stat_stream_tables;

Optional: WAL-based CDC

By default pg_stream uses triggers. If wal_level = logical is configured, set:

ALTER SYSTEM SET pg_stream.cdc_mode = 'auto';
SELECT pg_reload_conf();

pg_stream will automatically transition each stream table from trigger-based to WAL-based capture after the first successful refresh — reducing per-write overhead from ~50–200 μs to ~2–15 μs. The transition is transparent; your queries and the refresh schedule are unaffected.

Step 6: Understanding the Two IVM Strategies

You've now seen both strategies pg_stream uses for incremental view maintenance. Understanding when each applies helps you write efficient stream table queries.

Algebraic Differentiation (used by department_stats)

For queries composed of scans, filters, joins, and algebraic aggregates (COUNT, SUM, AVG), pg_stream can derive the IVM delta mathematically. The rules come from the theory of DBSP (Database Stream Processing):

The total cost is proportional to the number of changes, not the table size. For a million-row table with 10 changes, the delta query touches ~10 rows.

Incremental Strategies for Recursive CTEs (used by department_tree)

For recursive CTEs, pg_stream can't derive an algebraic delta because the recursion references itself. Instead it uses two complementary strategies, chosen automatically based on what changed:

Semi-naive evaluation (for INSERT-only changes):

1. Differentiate the base case — find the new seed rows
2. Propagate the delta through the recursive term, iterating until no new rows are produced
3. The result is only the new rows created by the change — not the whole tree

Delete-and-Rederive (DRed) (for DELETE or UPDATE):

1. Remove all rows derived from the old fact
2. Re-derive rows that had the old fact as one of their derivation paths (they may still be reachable via other paths)
3. Insert the newly derived rows under the new fact

Both strategies are more efficient than full recomputation — they work on the affected portion of the result set, not the entire recursive query. The MERGE only modifies rows that actually changed.

When to use which?

You don't choose — pg_stream detects the strategy automatically based on the query structure:

Step 7: Clean Up

When you're done experimenting, drop the stream tables. Drop dependents before their sources:

SELECT pgstream.drop_stream_table('department_report');
SELECT pgstream.drop_stream_table('department_stats');
SELECT pgstream.drop_stream_table('department_tree');

DROP TABLE employees;
DROP TABLE departments;

drop_stream_table atomically removes in a single transaction:

• The storage table (e.g., public.department_stats)
• CDC triggers on source tables (removed only if no other stream table references the same source)
• Change buffer tables in pgstream_changes
• Catalog entries in pgstream.pgs_stream_tables

Summary: What You Learned

The key takeaway: you write to base tables — pg_stream does the rest. Data flows downstream automatically, each layer doing the minimum work proportional to what changed, in dependency order.

What's Next?

• SQL_REFERENCE.md — Full API reference for all functions, views, and configuration
• ARCHITECTURE.md — Deep dive into the system architecture and data flow
• DVM_OPERATORS.md — How each SQL operator is differentiated for incremental maintenance
• CONFIGURATION.md — GUC variables for tuning schedule, concurrency, and cleanup behavior
• What Happens on INSERT — Detailed trace of a single INSERT through the entire pipeline
• What Happens on UPDATE — How UPDATEs are split into D+I, group key changes, and net-effect computation
• What Happens on DELETE — Reference counting, group deletion, and INSERT+DELETE cancellation
• What Happens on TRUNCATE — Why TRUNCATE bypasses triggers and how to recover

SQL Reference

Complete reference for all SQL functions, views, and catalog tables provided by pgstream.

Functions

pgstream.create_stream_table

Create a new stream table.

pgstream.create_stream_table(
    name          text,
    query         text,
    schedule text      DEFAULT '1m',
    refresh_mode  text      DEFAULT 'DIFFERENTIAL',
    initialize    bool      DEFAULT true
) → void

Parameters:

Duration format:

Cron expression format:

schedule also accepts standard cron expressions for time-based scheduling. The scheduler refreshes the stream table when the cron schedule fires, rather than checking staleness.

Note: Cron-scheduled stream tables do not participate in CALCULATED schedule resolution. The stale column in monitoring views returns NULL for cron-scheduled tables.

Example:

-- Duration-based: refresh when data is staler than 2 minutes
SELECT pgstream.create_stream_table(
    'order_totals',
    'SELECT region, SUM(amount) AS total FROM orders GROUP BY region',
    '2m',
    'DIFFERENTIAL'
);

-- Cron-based: refresh every hour
SELECT pgstream.create_stream_table(
    'hourly_summary',
    'SELECT date_trunc(''hour'', ts), COUNT(*) FROM events GROUP BY 1',
    '@hourly',
    'FULL'
);

-- Cron-based: refresh at 6 AM on weekdays
SELECT pgstream.create_stream_table(
    'daily_report',
    'SELECT region, SUM(revenue) AS total FROM sales GROUP BY region',
    '0 6 * * 1-5',
    'FULL'
);

Aggregate Examples:

All supported aggregate functions work in both FULL and DIFFERENTIAL modes:

-- Algebraic aggregates (fully differential — no rescan needed)
SELECT pgstream.create_stream_table(
    'sales_summary',
    'SELECT region, COUNT(*) AS cnt, SUM(amount) AS total, AVG(amount) AS avg_amount
     FROM orders GROUP BY region',
    '1m',
    'DIFFERENTIAL'
);

-- Semi-algebraic aggregates (MIN/MAX)
SELECT pgstream.create_stream_table(
    'salary_ranges',
    'SELECT department, MIN(salary) AS min_sal, MAX(salary) AS max_sal
     FROM employees GROUP BY department',
    '2m',
    'DIFFERENTIAL'
);

-- Group-rescan aggregates (BOOL_AND/OR, STRING_AGG, ARRAY_AGG, JSON_AGG, JSONB_AGG,
--                          BIT_AND, BIT_OR, BIT_XOR, JSON_OBJECT_AGG, JSONB_OBJECT_AGG,
--                          STDDEV, STDDEV_POP, STDDEV_SAMP, VARIANCE, VAR_POP, VAR_SAMP,
--                          MODE, PERCENTILE_CONT, PERCENTILE_DISC,
--                          CORR, COVAR_POP, COVAR_SAMP, REGR_AVGX, REGR_AVGY,
--                          REGR_COUNT, REGR_INTERCEPT, REGR_R2, REGR_SLOPE,
--                          REGR_SXX, REGR_SXY, REGR_SYY, ANY_VALUE)
SELECT pgstream.create_stream_table(
    'team_members',
    'SELECT department,
            STRING_AGG(name, '', '' ORDER BY name) AS members,
            ARRAY_AGG(employee_id) AS member_ids,
            BOOL_AND(active) AS all_active,
            JSON_AGG(name) AS members_json
     FROM employees
     GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- Bitwise aggregates
SELECT pgstream.create_stream_table(
    'permission_summary',
    'SELECT department,
            BIT_OR(permissions) AS combined_perms,
            BIT_AND(permissions) AS common_perms,
            BIT_XOR(flags) AS xor_flags
     FROM employees
     GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- JSON object aggregates
SELECT pgstream.create_stream_table(
    'config_map',
    'SELECT department,
            JSON_OBJECT_AGG(setting_name, setting_value) AS settings,
            JSONB_OBJECT_AGG(key, value) AS metadata
     FROM config
     GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- Statistical aggregates
SELECT pgstream.create_stream_table(
    'salary_stats',
    'SELECT department,
            STDDEV_POP(salary) AS sd_pop,
            STDDEV_SAMP(salary) AS sd_samp,
            VAR_POP(salary) AS var_pop,
            VAR_SAMP(salary) AS var_samp
     FROM employees
     GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- Ordered-set aggregates (MODE, PERCENTILE_CONT, PERCENTILE_DISC)
SELECT pgstream.create_stream_table(
    'salary_percentiles',
    'SELECT department,
            MODE() WITHIN GROUP (ORDER BY grade) AS most_common_grade,
            PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY salary) AS median_salary,
            PERCENTILE_DISC(0.9) WITHIN GROUP (ORDER BY salary) AS p90_salary
     FROM employees
     GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- Regression / correlation aggregates (CORR, COVAR_*, REGR_*)
SELECT pgstream.create_stream_table(
    'regression_stats',
    'SELECT department,
            CORR(salary, experience) AS sal_exp_corr,
            COVAR_POP(salary, experience) AS covar_pop,
            COVAR_SAMP(salary, experience) AS covar_samp,
            REGR_SLOPE(salary, experience) AS slope,
            REGR_INTERCEPT(salary, experience) AS intercept,
            REGR_R2(salary, experience) AS r_squared,
            REGR_COUNT(salary, experience) AS regr_n
     FROM employees
     GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- ANY_VALUE aggregate (PostgreSQL 16+)
SELECT pgstream.create_stream_table(
    'dept_sample',
    'SELECT department, ANY_VALUE(office_location) AS sample_office
     FROM employees GROUP BY department',
    '1m',
    'DIFFERENTIAL'
);

-- FILTER clause on aggregates
SELECT pgstream.create_stream_table(
    'order_metrics',
    'SELECT region,
            COUNT(*) AS total,
            COUNT(*) FILTER (WHERE status = ''active'') AS active_count,
            SUM(amount) FILTER (WHERE status = ''shipped'') AS shipped_total
     FROM orders
     GROUP BY region',
    '1m',
    'DIFFERENTIAL'
);

CTE Examples:

Non-recursive CTEs are fully supported in both FULL and DIFFERENTIAL modes:

-- Simple CTE
SELECT pgstream.create_stream_table(
    'active_order_totals',
    'WITH active_users AS (
        SELECT id, name FROM users WHERE active = true
    )
    SELECT a.id, a.name, SUM(o.amount) AS total
    FROM active_users a
    JOIN orders o ON o.user_id = a.id
    GROUP BY a.id, a.name',
    '1m',
    'DIFFERENTIAL'
);

-- Chained CTEs (CTE referencing another CTE)
SELECT pgstream.create_stream_table(
    'top_regions',
    'WITH regional AS (
        SELECT region, SUM(amount) AS total FROM orders GROUP BY region
    ),
    ranked AS (
        SELECT region, total FROM regional WHERE total > 1000
    )
    SELECT * FROM ranked',
    '2m',
    'DIFFERENTIAL'
);

-- Multi-reference CTE (referenced twice in FROM — shared delta optimization)
SELECT pgstream.create_stream_table(
    'self_compare',
    'WITH totals AS (
        SELECT user_id, SUM(amount) AS total FROM orders GROUP BY user_id
    )
    SELECT t1.user_id, t1.total, t2.total AS next_total
    FROM totals t1
    JOIN totals t2 ON t1.user_id = t2.user_id + 1',
    '1m',
    'DIFFERENTIAL'
);

Recursive CTEs work with both FULL and DIFFERENTIAL modes:

-- Recursive CTE (hierarchy traversal)
SELECT pgstream.create_stream_table(
    'category_tree',
    'WITH RECURSIVE cat_tree AS (
        SELECT id, name, parent_id, 0 AS depth
        FROM categories WHERE parent_id IS NULL
        UNION ALL
        SELECT c.id, c.name, c.parent_id, ct.depth + 1
        FROM categories c
        JOIN cat_tree ct ON c.parent_id = ct.id
    )
    SELECT * FROM cat_tree',
    '5m',
    'FULL'  -- FULL mode: standard re-execution
);

-- Recursive CTE with DIFFERENTIAL mode (incremental semi-naive / DRed)
SELECT pgstream.create_stream_table(
    'org_chart',
    'WITH RECURSIVE reports AS (
        SELECT id, name, manager_id FROM employees WHERE manager_id IS NULL
        UNION ALL
        SELECT e.id, e.name, e.manager_id
        FROM employees e JOIN reports r ON e.manager_id = r.id
    )
    SELECT * FROM reports',
    '2m',
    'DIFFERENTIAL'  -- Uses semi-naive, DRed, or recomputation (auto-selected)
);

Set Operation Examples:

INTERSECT, INTERSECT ALL, EXCEPT, EXCEPT ALL, UNION, and UNION ALL are supported:

-- INTERSECT: customers who placed orders in BOTH regions
SELECT pgstream.create_stream_table(
    'bi_region_customers',
    'SELECT customer_id FROM orders_east
     INTERSECT
     SELECT customer_id FROM orders_west',
    '2m',
    'DIFFERENTIAL'
);

-- INTERSECT ALL: preserves duplicates (bag semantics)
SELECT pgstream.create_stream_table(
    'common_items',
    'SELECT item_name FROM warehouse_a
     INTERSECT ALL
     SELECT item_name FROM warehouse_b',
    '1m',
    'DIFFERENTIAL'
);

-- EXCEPT: orders not yet shipped
SELECT pgstream.create_stream_table(
    'unshipped_orders',
    'SELECT order_id FROM orders
     EXCEPT
     SELECT order_id FROM shipments',
    '1m',
    'DIFFERENTIAL'
);

-- EXCEPT ALL: preserves duplicate counts (bag subtraction)
SELECT pgstream.create_stream_table(
    'excess_inventory',
    'SELECT sku FROM stock_received
     EXCEPT ALL
     SELECT sku FROM stock_shipped',
    '5m',
    'DIFFERENTIAL'
);

-- UNION: deduplicated merge of two sources
SELECT pgstream.create_stream_table(
    'all_contacts',
    'SELECT email FROM customers
     UNION
     SELECT email FROM newsletter_subscribers',
    '5m',
    'DIFFERENTIAL'
);

LATERAL Set-Returning Function Examples:

Set-returning functions (SRFs) in the FROM clause are supported in both FULL and DIFFERENTIAL modes. Common SRFs include jsonb_array_elements, jsonb_each, jsonb_each_text, and unnest:

-- Flatten JSONB arrays into rows
SELECT pgstream.create_stream_table(
    'flat_children',
    'SELECT p.id, child.value AS val
     FROM parent_data p,
     jsonb_array_elements(p.data->''children'') AS child',
    '1m',
    'DIFFERENTIAL'
);

-- Expand JSONB key-value pairs (multi-column SRF)
SELECT pgstream.create_stream_table(
    'flat_properties',
    'SELECT d.id, kv.key, kv.value
     FROM documents d,
     jsonb_each(d.metadata) AS kv',
    '2m',
    'DIFFERENTIAL'
);

-- Unnest arrays
SELECT pgstream.create_stream_table(
    'flat_tags',
    'SELECT t.id, tag.tag
     FROM tagged_items t,
     unnest(t.tags) AS tag(tag)',
    '1m',
    'DIFFERENTIAL'
);

-- SRF with WHERE filter
SELECT pgstream.create_stream_table(
    'high_value_items',
    'SELECT p.id, (e.value)::int AS amount
     FROM products p,
     jsonb_array_elements(p.prices) AS e
     WHERE (e.value)::int > 100',
    '5m',
    'DIFFERENTIAL'
);

-- SRF combined with aggregation
SELECT pgstream.create_stream_table(
    'element_counts',
    'SELECT a.id, count(*) AS cnt
     FROM arrays a,
     jsonb_array_elements(a.data) AS e
     GROUP BY a.id',
    '1m',
    'FULL'
);

LATERAL Subquery Examples:

LATERAL subqueries in the FROM clause are supported in both FULL and DIFFERENTIAL modes. Use them for top-N per group, correlated aggregation, and conditional expansion:

-- Top-N per group: latest item per order
SELECT pgstream.create_stream_table(
    'latest_items',
    'SELECT o.id, o.customer, latest.amount
     FROM orders o,
     LATERAL (
         SELECT li.amount
         FROM line_items li
         WHERE li.order_id = o.id
         ORDER BY li.created_at DESC
         LIMIT 1
     ) AS latest',
    '1m',
    'DIFFERENTIAL'
);

-- Correlated aggregate
SELECT pgstream.create_stream_table(
    'dept_summaries',
    'SELECT d.id, d.name, stats.total, stats.cnt
     FROM departments d,
     LATERAL (
         SELECT SUM(e.salary) AS total, COUNT(*) AS cnt
         FROM employees e
         WHERE e.dept_id = d.id
     ) AS stats',
    '1m',
    'DIFFERENTIAL'
);

-- LEFT JOIN LATERAL: preserve outer rows with NULLs when subquery returns no rows
SELECT pgstream.create_stream_table(
    'dept_stats_all',
    'SELECT d.id, d.name, stats.total
     FROM departments d
     LEFT JOIN LATERAL (
         SELECT SUM(e.salary) AS total
         FROM employees e
         WHERE e.dept_id = d.id
     ) AS stats ON true',
    '1m',
    'DIFFERENTIAL'
);

WHERE Subquery Examples:

Subqueries in the WHERE clause are automatically transformed into semi-join, anti-join, or scalar subquery operators in the DVM operator tree:

-- EXISTS subquery: customers who have placed orders
SELECT pgstream.create_stream_table(
    'active_customers',
    'SELECT c.id, c.name
     FROM customers c
     WHERE EXISTS (SELECT 1 FROM orders o WHERE o.customer_id = c.id)',
    '1m',
    'DIFFERENTIAL'
);

-- NOT EXISTS: customers with no orders
SELECT pgstream.create_stream_table(
    'inactive_customers',
    'SELECT c.id, c.name
     FROM customers c
     WHERE NOT EXISTS (SELECT 1 FROM orders o WHERE o.customer_id = c.id)',
    '1m',
    'DIFFERENTIAL'
);

-- IN subquery: products that have been ordered
SELECT pgstream.create_stream_table(
    'ordered_products',
    'SELECT p.id, p.name
     FROM products p
     WHERE p.id IN (SELECT product_id FROM order_items)',
    '1m',
    'DIFFERENTIAL'
);

-- NOT IN subquery: products never ordered
SELECT pgstream.create_stream_table(
    'unordered_products',
    'SELECT p.id, p.name
     FROM products p
     WHERE p.id NOT IN (SELECT product_id FROM order_items)',
    '1m',
    'DIFFERENTIAL'
);

-- Scalar subquery in SELECT list
SELECT pgstream.create_stream_table(
    'products_with_max_price',
    'SELECT p.id, p.name, (SELECT max(price) FROM products) AS max_price
     FROM products p',
    '1m',
    'DIFFERENTIAL'
);

Notes:

• The defining query is parsed into an operator tree and validated for DVM support.
• Views as sources — views referenced in the defining query are automatically inlined as subqueries (auto-rewrite pass #0). CDC triggers are created on the underlying base tables. Nested views (view → view → table) are fully expanded. The user's original query is preserved in original_query for reinit and introspection. Materialized views are rejected in DIFFERENTIAL mode (use FULL mode or the underlying query directly). Foreign tables are also rejected in DIFFERENTIAL mode.
• CDC triggers and change buffer tables are created automatically for each source table.
• The ST is registered in the dependency DAG; cycles are rejected.
• Non-recursive CTEs are inlined as subqueries during parsing (Tier 1). Multi-reference CTEs share delta computation (Tier 2).
• Recursive CTEs in DIFFERENTIAL mode use three strategies, auto-selected per refresh: semi-naive evaluation for INSERT-only changes, Delete-and-Rederive (DRed) for mixed changes, and recomputation fallback when CTE columns don't match ST storage columns.
• LATERAL SRFs in DIFFERENTIAL mode use row-scoped recomputation: when a source row changes, only the SRF expansions for that row are re-evaluated.
• LATERAL subqueries in DIFFERENTIAL mode also use row-scoped recomputation: when an outer row changes, the correlated subquery is re-executed only for that row.
• WHERE subqueries (EXISTS, IN, scalar) are parsed into dedicated semi-join, anti-join, and scalar subquery operators with specialized delta computation.
• ALL (subquery) is the only subquery form that is currently rejected.
• ORDER BY is accepted but silently discarded — row order in the storage table is undefined (consistent with PostgreSQL's CREATE MATERIALIZED VIEW behavior). Apply ORDER BY when querying the stream table.
• LIMIT / OFFSET are rejected — stream tables materialize the full result set. Apply LIMIT when querying the stream table.

pgstream.alter_stream_table

Alter properties of an existing stream table.

pgstream.alter_stream_table(
    name          text,
    schedule text      DEFAULT NULL,
    refresh_mode  text      DEFAULT NULL,
    status        text      DEFAULT NULL
) → void

Parameters:

Examples:

-- Change schedule
SELECT pgstream.alter_stream_table('order_totals', schedule => '5m');

-- Switch to full refresh mode
SELECT pgstream.alter_stream_table('order_totals', refresh_mode => 'FULL');

-- Suspend a stream table
SELECT pgstream.alter_stream_table('order_totals', status => 'SUSPENDED');

-- Resume a suspended stream table
SELECT pgstream.alter_stream_table('order_totals', status => 'ACTIVE');

pgstream.drop_stream_table

Drop a stream table, removing the storage table and all catalog entries.

pgstream.drop_stream_table(name text) → void

Parameters:

Example:

SELECT pgstream.drop_stream_table('order_totals');

Notes:

• Drops the underlying storage table with CASCADE.
• Removes all catalog entries (metadata, dependencies, refresh history).
• Cleans up CDC triggers and change buffer tables for source tables that are no longer tracked by any ST.

pgstream.refresh_stream_table

Manually trigger a synchronous refresh of a stream table.

pgstream.refresh_stream_table(name text) → void

Parameters:

Example:

SELECT pgstream.refresh_stream_table('order_totals');

Notes:

• Blocked if the ST is SUSPENDED — use pgstream.alter_stream_table(... status => 'ACTIVE') first.
• Uses an advisory lock to prevent concurrent refreshes of the same ST.
• For DIFFERENTIAL mode, generates and applies a delta query. For FULL mode, truncates and reloads.
• Records the refresh in pgstream.pgs_refresh_history.

pgstream.pgs_status

Get the status of all stream tables.

pgstream.pgs_status() → SETOF record(
    name                text,
    status              text,
    refresh_mode        text,
    is_populated        bool,
    consecutive_errors  int,
    schedule            text,
    data_timestamp      timestamptz,
    staleness           interval
)

Example:

SELECT * FROM pgstream.pgs_status();

pgstream.st_refresh_stats

Return per-ST refresh statistics aggregated from the refresh history.

pgstream.st_refresh_stats() → SETOF record(
    pgs_name                text,
    pgs_schema              text,
    status                 text,
    refresh_mode           text,
    is_populated           bool,
    total_refreshes        bigint,
    successful_refreshes   bigint,
    failed_refreshes       bigint,
    total_rows_inserted    bigint,
    total_rows_deleted     bigint,
    avg_duration_ms        float8,
    last_refresh_action    text,
    last_refresh_status    text,
    last_refresh_at        timestamptz,
    staleness_secs       float8,
    stale           bool
)

Example:

SELECT pgs_name, status, total_refreshes, avg_duration_ms, stale
FROM pgstream.st_refresh_stats();

pgstream.get_refresh_history

Return refresh history for a specific stream table.

pgstream.get_refresh_history(
    name      text,
    max_rows  int  DEFAULT 20
) → SETOF record(
    refresh_id       bigint,
    data_timestamp   timestamptz,
    start_time       timestamptz,
    end_time         timestamptz,
    action           text,
    status           text,
    rows_inserted    bigint,
    rows_deleted     bigint,
    duration_ms      float8,
    error_message    text
)

Example:

SELECT action, status, rows_inserted, duration_ms
FROM pgstream.get_refresh_history('order_totals', 5);

pgstream.get_staleness

Get the current staleness in seconds for a specific stream table.

pgstream.get_staleness(name text) → float8

Returns NULL if the ST has never been refreshed.

Example:

SELECT pgstream.get_staleness('order_totals');
-- Returns: 12.345  (seconds since last refresh)

pgstream.slot_health

Check replication slot health for all tracked CDC slots.

pgstream.slot_health() → SETOF record(
    slot_name          text,
    source_relid       bigint,
    active             bool,
    retained_wal_bytes bigint,
    wal_status         text
)

Example:

SELECT * FROM pgstream.slot_health();

pgstream.check_cdc_health

Check CDC health for all tracked source tables. Returns per-source health status including the current CDC mode, replication slot details, estimated lag, and any alerts.

pgstream.check_cdc_health() → SETOF record(
    source_relid   bigint,
    source_table   text,
    cdc_mode       text,
    slot_name      text,
    lag_bytes      bigint,
    confirmed_lsn  text,
    alert          text
)

Columns:

Example:

SELECT * FROM pgstream.check_cdc_health();

pgstream.explain_st

Explain the DVM plan for a stream table's defining query.

pgstream.explain_st(name text) → SETOF record(
    property  text,
    value     text
)

Example:

SELECT * FROM pgstream.explain_st('order_totals');

pgstream.pg_stream_hash

Compute a 64-bit xxHash row ID from a text value.

pgstream.pg_stream_hash(input text) → bigint

Marked IMMUTABLE, PARALLEL SAFE.

Example:

SELECT pgstream.pg_stream_hash('some_key');
-- Returns: 1234567890123456789

pgstream.pg_stream_hash_multi

Compute a row ID by hashing multiple text values (composite keys).

pgstream.pg_stream_hash_multi(inputs text[]) → bigint

Marked IMMUTABLE, PARALLEL SAFE. Uses \x1E (record separator) between values and \x00NULL\x00 for NULL entries.

Example:

SELECT pgstream.pg_stream_hash_multi(ARRAY['key1', 'key2']);

Expression Support

pgstream's DVM parser supports a wide range of SQL expressions in defining queries. All expressions work in both FULL and DIFFERENTIAL modes.

Conditional Expressions

Comparison Operators

Boolean Tests

SQL Value Functions

Array and Row Expressions

Subquery Expressions

Subqueries are supported in the WHERE clause and SELECT list. They are parsed into dedicated DVM operators with specialized delta computation for incremental maintenance.

Notes:

• EXISTS and IN (subquery) in the WHERE clause are transformed into semi-join operators. NOT EXISTS and NOT IN (subquery) become anti-join operators.
• Multiple subqueries in the same WHERE clause are supported when combined with AND. Subqueries combined with OR are also supported — they are automatically rewritten into UNION of separate filtered queries.
• Scalar subqueries in the SELECT list are supported as long as they return exactly one row and one column.
• ALL (subquery) is supported — it is automatically rewritten to an anti-join via NOT EXISTS with the negated condition.

Auto-Rewrite Pipeline

pg_stream transparently rewrites certain SQL constructs before parsing. These rewrites are applied automatically and require no user action:

HAVING Clause

HAVING is fully supported. The filter predicate is applied on top of the aggregate delta computation — groups that pass the HAVING condition are included in the stream table.

SELECT pgstream.create_stream_table('big_departments',
  'SELECT department, COUNT(*) AS cnt FROM employees GROUP BY department HAVING COUNT(*) > 10',
  '1m', 'DIFFERENTIAL');

Tables Without Primary Keys (Keyless Tables)

Tables without a primary key can be used as sources. pg_stream generates a content-based row identity by hashing all column values using pg_stream_hash_multi(). This allows DIFFERENTIAL mode to work, though at the cost of being unable to distinguish truly duplicate rows (rows with identical values in all columns).

-- No primary key — pg_stream uses content hashing for row identity
CREATE TABLE events (ts TIMESTAMPTZ, payload JSONB);
SELECT pgstream.create_stream_table('event_summary',
  'SELECT payload->>''type'' AS event_type, COUNT(*) FROM events GROUP BY 1',
  '1m', 'DIFFERENTIAL');

Volatile Function Detection

pg_stream checks all functions and operators in the defining query against pg_proc.provolatile:

• VOLATILE functions (e.g., random(), now(), gen_random_uuid()) are rejected in DIFFERENTIAL mode because they produce different results on each evaluation, breaking delta correctness.
• VOLATILE operators — custom operators backed by volatile functions are also detected. The check resolves the operator’s implementation function via pg_operator.oprcode and checks its volatility in pg_proc.
• STABLE functions (e.g., current_setting()) produce a warning — they are consistent within a single transaction but may differ between refreshes.
• IMMUTABLE functions are always safe and produce no warnings.

FULL mode accepts all volatility classes since it re-evaluates the entire query each time.

COLLATE Expressions

COLLATE clauses on expressions are supported:

SELECT pgstream.create_stream_table('sorted_names',
  'SELECT name COLLATE "C" AS c_name FROM users',
  '1m', 'DIFFERENTIAL');

IS JSON Predicate (PostgreSQL 16+)

The IS JSON predicate validates whether a value is valid JSON. All variants are supported:

-- Filter rows with valid JSON
SELECT pgstream.create_stream_table('valid_json_events',
  'SELECT id, payload FROM events WHERE payload::text IS JSON',
  '1m', 'DIFFERENTIAL');

-- Type-specific checks
SELECT pgstream.create_stream_table('json_objects_only',
  'SELECT id, data IS JSON OBJECT AS is_obj,
          data IS JSON ARRAY AS is_arr,
          data IS JSON SCALAR AS is_scalar
   FROM json_data',
  '1m', 'FULL');

Supported variants: IS JSON, IS JSON OBJECT, IS JSON ARRAY, IS JSON SCALAR, IS NOT JSON (all forms), WITH UNIQUE KEYS.

SQL/JSON Constructors (PostgreSQL 16+)

SQL-standard JSON constructor functions are supported in both FULL and DIFFERENTIAL modes:

-- JSON_OBJECT: construct a JSON object from key-value pairs
SELECT pgstream.create_stream_table('user_json',
  'SELECT id, JSON_OBJECT(''name'' : name, ''age'' : age) AS data FROM users',
  '1m', 'DIFFERENTIAL');

-- JSON_ARRAY: construct a JSON array from values
SELECT pgstream.create_stream_table('value_arrays',
  'SELECT id, JSON_ARRAY(a, b, c) AS arr FROM measurements',
  '1m', 'FULL');

-- JSON(): parse a text value as JSON
-- JSON_SCALAR(): wrap a scalar value as JSON
-- JSON_SERIALIZE(): serialize a JSON value to text

Note: JSON_ARRAYAGG() and JSON_OBJECTAGG() are SQL-standard aggregate functions fully recognized by the DVM engine. In DIFFERENTIAL mode, they use the group-rescan strategy (affected groups are re-aggregated from source data). The full deparsed SQL is preserved to handle the special key: value, ABSENT ON NULL, ORDER BY, and RETURNING clause syntax.

JSON_TABLE (PostgreSQL 17+)

JSON_TABLE() generates a relational table from JSON data. It is supported in the FROM clause in both FULL and DIFFERENTIAL modes. Internally, it is modeled as a LateralFunction.

-- Extract structured data from a JSON column
SELECT pgstream.create_stream_table('user_phones',
  $$SELECT u.id, j.phone_type, j.phone_number
    FROM users u,
         JSON_TABLE(u.contact_info, '$.phones[*]'
           COLUMNS (
             phone_type TEXT PATH '$.type',
             phone_number TEXT PATH '$.number'
           )
         ) AS j$$,
  '1m', 'DIFFERENTIAL');

Supported column types:

• Regular columns — name TYPE PATH '$.path' (with optional ON ERROR/ON EMPTY behaviors)
• EXISTS columns — name TYPE EXISTS PATH '$.path'
• Formatted columns — name TYPE FORMAT JSON PATH '$.path'
• Nested columns — NESTED PATH '$.path' COLUMNS (...)

The PASSING clause is also supported for passing named variables to path expressions.

Unsupported Expression Types

The following are rejected with clear error messages rather than producing broken SQL:

Restrictions & Interoperability

Stream tables are standard PostgreSQL heap tables stored in the pgstream schema with an additional __pgs_row_id BIGINT PRIMARY KEY column managed by the refresh engine. This section describes what you can and cannot do with them.

Referencing Other Stream Tables

Stream tables can reference other stream tables in their defining query. This creates a dependency edge in the internal DAG, and the scheduler refreshes upstream tables before downstream ones. Cycles are detected and rejected at creation time.

-- ST1 reads from a base table
SELECT pgstream.create_stream_table('order_totals',
  'SELECT customer_id, SUM(amount) AS total FROM orders GROUP BY customer_id',
  '1m', 'DIFFERENTIAL');

-- ST2 reads from ST1
SELECT pgstream.create_stream_table('big_customers',
  'SELECT customer_id, total FROM pgstream.order_totals WHERE total > 1000',
  '1m', 'DIFFERENTIAL');

Views as Sources in Defining Queries

PostgreSQL views can be used as source tables in a stream table's defining query. Views are automatically inlined — replaced with their underlying SELECT definition as subqueries — so CDC triggers land on the actual base tables.

CREATE VIEW active_orders AS
  SELECT * FROM orders WHERE status = 'active';

-- This works in DIFFERENTIAL mode:
SELECT pgstream.create_stream_table('order_summary',
  'SELECT customer_id, COUNT(*) FROM active_orders GROUP BY customer_id',
  '1m', 'DIFFERENTIAL');
-- Internally, 'active_orders' is replaced with:
--   (SELECT ... FROM orders WHERE status = 'active') AS active_orders

Nested views (view → view → table) are fully expanded via a fixpoint loop. Column-renaming views (CREATE VIEW v(a, b) AS ...) work correctly — pg_get_viewdef() produces the proper column aliases.

When a view is inlined, the user's original SQL is stored in the original_query catalog column for reinit and introspection. The defining_query column contains the expanded (post-inlining) form.

DDL hooks: CREATE OR REPLACE VIEW on a view that was inlined into a stream table marks that ST for reinit. DROP VIEW sets affected STs to ERROR status.

Materialized views are rejected in DIFFERENTIAL mode — their stale-snapshot semantics prevent CDC triggers from tracking changes. Use the underlying query directly, or switch to FULL mode. In FULL mode, materialized views are allowed (no CDC needed).

Foreign tables are rejected in DIFFERENTIAL mode — row-level triggers cannot be created on foreign tables. Use FULL mode instead.

Partitioned Tables as Sources

Partitioned tables are fully supported as source tables in both FULL and DIFFERENTIAL modes. CDC triggers are installed on the partitioned parent table, and PostgreSQL 13+ ensures the trigger fires for all DML routed to child partitions. The change buffer uses the parent table's OID (pgstream_changes.changes_<parent_oid>).

CREATE TABLE orders (
    id INT, region TEXT, amount NUMERIC
) PARTITION BY LIST (region);
CREATE TABLE orders_us PARTITION OF orders FOR VALUES IN ('US');
CREATE TABLE orders_eu PARTITION OF orders FOR VALUES IN ('EU');

-- Works — inserts into any partition are captured:
SELECT pgstream.create_stream_table('order_summary',
  'SELECT region, SUM(amount) FROM orders GROUP BY region',
  '1m', 'DIFFERENTIAL');

Note: pg_stream targets PostgreSQL 18. On PostgreSQL 12 or earlier (not supported), parent triggers do not fire for partition-routed rows, which would cause silent data loss.

Logical Replication Targets

Tables that receive data via logical replication require special consideration. Changes arriving via replication do not fire normal row-level triggers, which means CDC triggers will miss those changes.

pg_stream emits a WARNING at stream table creation time if any source table is detected as a logical replication target (via pg_subscription_rel).

Workarounds:

• Use cdc_mode = 'wal' for WAL-based CDC that captures all changes regardless of origin.
• Use FULL refresh mode, which recomputes entirely from the current table state.
• Set a frequent refresh schedule with FULL mode to limit staleness.

Views on Stream Tables

PostgreSQL views can reference stream tables. The view reflects the data as of the most recent refresh.

CREATE VIEW top_customers AS
SELECT customer_id, total
FROM pgstream.order_totals
WHERE total > 500
ORDER BY total DESC;

Materialized Views on Stream Tables

Materialized views can reference stream tables, though this is typically redundant (both are physical snapshots of a query). The materialized view requires its own REFRESH MATERIALIZED VIEW — it does not auto-refresh when the stream table refreshes.

Logical Replication of Stream Tables

Stream tables can be published for logical replication like any ordinary table:

-- On publisher
CREATE PUBLICATION my_pub FOR TABLE pgstream.order_totals;

-- On subscriber
CREATE SUBSCRIPTION my_sub
  CONNECTION 'host=... dbname=...'
  PUBLICATION my_pub;

Caveats:

• The __pgs_row_id column is replicated (it is the primary key), which is an internal implementation detail.
• The subscriber receives materialized data, not the defining query. Refreshes on the publisher propagate as normal DML via logical replication.
• Do not install pg_stream on the subscriber and attempt to refresh the replicated table — it will have no CDC triggers or catalog entries.
• The internal change buffer tables (pgstream_changes.changes_<oid>) and catalog tables are not published by default; subscribers only receive the final output.

Known Delta Computation Limitations

The following edge cases produce incorrect delta results in DIFFERENTIAL mode under specific data mutation patterns. They have no effect on FULL mode.

JOIN Key Column Change + Simultaneous Right-Side Delete

When a row's join key column is updated in the same refresh cycle as the joined-side row is deleted, the delta query may fail to emit the required DELETE from the stream table:

-- Stream table joining orders with customers
SELECT pgstream.create_stream_table('order_details',
  'SELECT o.id, c.name FROM orders o JOIN customers c ON o.cust_id = c.id',
  '1m', 'DIFFERENTIAL');

-- Scenario that exposes the limitation:
-- In the same transaction (or same refresh interval):
UPDATE orders SET cust_id = 5 WHERE cust_id = 3;  -- key change
DELETE FROM customers WHERE id = 3;               -- old join partner deleted
-- The delta for the now-stale (orders.cust_id=3, customers.id=3) join result
-- may not be emitted as a DELETE, leaving a stale row in the stream table
-- until the next full refresh cycle.

Root cause: The JOIN delta query reads current_right (customers) after all changes are applied. When customer 3 is deleted before the delta runs, the DELETE half of the join cannot find its join partner and is silently dropped.

Mitigations:

• Adaptive FULL fallback (default): when the scheduler detects a high change volume, it switches to a full recompute, which will correct any stale rows. The threshold is configurable via pg_stream.adaptive_full_threshold.
• Avoid co-locating key-changing UPDATEs and DELETEs in the same refresh interval. Stagger changes across multiple refresh cycles.
• FULL mode for stream tables where join key changes and right-side deletes are expected to co-occur frequently.

CUBE/ROLLUP Expansion Limit

CUBE(a, b, c...n) on N columns generates $2^N$ grouping set branches (a UNION ALL of N queries). pg_stream rejects CUBE/ROLLUP that would produce more than 64 branches to prevent runaway memory usage during query generation. Use explicit GROUPING SETS(...) instead:

-- Rejected: CUBE(a, b, c, d, e, f, g) would generate 128 branches
-- Use instead:
SELECT pgstream.create_stream_table('multi_dim',
  'SELECT a, b, c, SUM(v) FROM t
   GROUP BY GROUPING SETS ((a, b, c), (a, b), (a), ())',
  '5m', 'DIFFERENTIAL');

What Is NOT Allowed

Tip: The __pgs_row_id column is visible but should be ignored by consuming queries — it is an implementation detail used for delta MERGE operations.

Views

pgstream.stream_tables_info

Status overview with computed staleness information.

SELECT * FROM pgstream.stream_tables_info;

Columns include all pgstream.pgs_stream_tables columns plus:

pgstream.pg_stat_stream_tables

Comprehensive monitoring view combining catalog metadata with aggregate refresh statistics.

SELECT * FROM pgstream.pg_stat_stream_tables;

Key columns:

Catalog Tables

pgstream.pgs_stream_tables

Core metadata for each stream table.

pgstream.pgs_dependencies

DAG edges — records which source tables each ST depends on, including CDC mode metadata.

pgstream.pgs_refresh_history

Audit log of all refresh operations.

pgstream.pgs_change_tracking

CDC slot tracking per source table.

Configuration

Complete reference for all pg_stream GUC (Grand Unified Configuration) variables.

Overview

pg_stream exposes sixteen configuration variables in the pg_stream namespace. All can be set in postgresql.conf or at runtime via SET / ALTER SYSTEM.

Required postgresql.conf settings:

shared_preload_libraries = 'pg_stream'

The extension must be loaded via shared_preload_libraries because it registers GUC variables and a background worker at startup.

Note: wal_level = logical and max_replication_slots are not required by default. The default CDC mode (trigger) uses lightweight row-level triggers. If you set pg_stream.cdc_mode = 'auto' or 'wal', then wal_level = logical is needed for WAL-based capture (see pg_stream.cdc_mode).

GUC Variables

pg_stream.enabled

Enable or disable the pg_stream extension.

When set to false, the background scheduler stops processing refreshes. Existing stream tables remain in the catalog but are not refreshed. Manual pgstream.refresh_stream_table() calls still work.

-- Disable automatic refreshes
SET pg_stream.enabled = false;

-- Re-enable
SET pg_stream.enabled = true;

pg_stream.scheduler_interval_ms

How often the background scheduler checks for stream tables that need refreshing.

Tuning Guidance:

• Low-latency workloads (sub-second schedule): Set to 100–500.
• Standard workloads (minutes of schedule): Default 1000 is appropriate.
• Low-overhead workloads (many STs with long schedules): Increase to 5000–10000 to reduce scheduler overhead.

The scheduler interval does not determine refresh frequency — it determines how often the scheduler checks whether any ST's staleness exceeds its schedule (or whether a cron expression has fired). The actual refresh frequency is governed by schedule (duration or cron) and canonical period alignment.

SET pg_stream.scheduler_interval_ms = 500;

pg_stream.min_schedule_seconds

Minimum allowed schedule value (in seconds) when creating or altering a stream table with a duration-based schedule. This limit does not apply to cron expressions.

This acts as a safety guardrail to prevent users from setting impractically small schedules that would cause excessive refresh overhead.

Tuning Guidance:

• Development/testing: Set to 1 for fast iteration.
• Production: Keep at 60 or higher to prevent excessive WAL consumption and CPU usage.

-- Allow 10-second schedules (for testing)
SET pg_stream.min_schedule_seconds = 10;

pg_stream.max_consecutive_errors

Maximum consecutive refresh failures before a stream table is moved to ERROR status.

When a ST's consecutive_errors reaches this threshold:

1. The ST status changes to ERROR.
2. Automatic refreshes stop for this ST.
3. Manual intervention is required: SELECT pgstream.alter_stream_table('...', status => 'ACTIVE').

Tuning Guidance:

• Strict (production): 3 — fail fast to surface issues.
• Lenient (development): 10–20 — tolerate transient errors.

SET pg_stream.max_consecutive_errors = 5;

pg_stream.change_buffer_schema

Schema where CDC change buffer tables are created.

Change buffer tables are named <schema>.changes_<oid> where <oid> is the source table's OID. Placing them in a dedicated schema keeps them out of the public namespace.

Tuning Guidance:

• Generally leave at the default. Change only if pgstream_changes conflicts with an existing schema in your database.

SET pg_stream.change_buffer_schema = 'my_change_buffers';

pg_stream.max_concurrent_refreshes

Maximum number of stream tables that can be refreshed simultaneously.

Controls concurrency in the scheduler. Each refresh acquires an advisory lock, and the scheduler skips STs that exceed this limit.

Tuning Guidance:

• Small databases (few STs): 1–4 is sufficient.
• Large deployments (50+ STs): Increase to 8–16 if the server has spare CPU and I/O capacity.
• Resource-constrained: Set to 1 for fully sequential refresh processing.

The optimal setting depends on:

• Number of CPU cores available
• I/O throughput (SSD vs HDD)
• Complexity of the defining queries
• Amount of concurrent OLTP workload

SET pg_stream.max_concurrent_refreshes = 8;

pg_stream.differential_max_change_ratio

Maximum change-to-table ratio before DIFFERENTIAL refresh falls back to FULL refresh.

When the number of pending change buffer rows exceeds this fraction of the source table's estimated row count, the refresh engine switches from DIFFERENTIAL (which uses JSONB parsing and window functions) to FULL refresh. At high change rates FULL refresh is cheaper because it avoids the per-row JSONB overhead.

Special Values:

• 0.0: Disable adaptive fallback — always use DIFFERENTIAL.
• 1.0: Always fall back to FULL (effectively forces FULL mode).

Tuning Guidance:

• OLTP with low change rates (< 5%): Default 0.15 is appropriate.
• Batch-load workloads (bulk inserts): Lower to 0.05–0.10 so large batches trigger FULL refresh sooner.
• Latency-sensitive (want deterministic refresh time): Set to 0.0 to always use DIFFERENTIAL.

-- Lower threshold for batch-heavy workloads
SET pg_stream.differential_max_change_ratio = 0.10;

-- Disable adaptive fallback
SET pg_stream.differential_max_change_ratio = 0.0;

pg_stream.cleanup_use_truncate

Use TRUNCATE instead of per-row DELETE for change buffer cleanup when the entire buffer is consumed by a refresh.

After a differential refresh consumes all rows from the change buffer, the engine must clean up the buffer table. TRUNCATE is O(1) regardless of row count, versus DELETE which must update indexes row-by-row. This saves 3–5 ms per refresh at 10%+ change rates.

Trade-off: TRUNCATE acquires an AccessExclusiveLock on the change buffer table. If concurrent DML on the source table is actively inserting into the same change buffer via triggers, this lock can cause brief contention.

Tuning Guidance:

• Most workloads: Leave at true — the performance benefit outweighs the brief lock.
• High-concurrency OLTP with continuous writes during refresh: Set to false if you observe lock-wait timeouts on the change buffer.

-- Use per-row DELETE for change buffer cleanup
SET pg_stream.cleanup_use_truncate = false;

pg_stream.merge_planner_hints

Inject SET LOCAL planner hints before MERGE execution during differential refresh.

When enabled, the refresh executor estimates the delta size and applies optimizer hints within the transaction:

• Delta ≥ 100 rows: SET LOCAL enable_nestloop = off — forces hash joins instead of nested-loop joins.
• Delta ≥ 10,000 rows: additionally SET LOCAL work_mem = '<N>MB' (see pg_stream.merge_work_mem_mb).

This reduces P95 latency spikes caused by PostgreSQL choosing nested-loop plans for medium/large delta sizes.

Tuning Guidance:

• Most workloads: Leave at true — the hints improve tail latency without affecting small deltas.
• Custom plan overrides: Set to false if you manage planner settings yourself or if the hints conflict with your pg_hint_plan configuration.

-- Disable planner hints
SET pg_stream.merge_planner_hints = false;

pg_stream.merge_work_mem_mb

work_mem value (in MB) applied via SET LOCAL when the delta exceeds 10,000 rows and planner hints are enabled.

A higher value lets PostgreSQL use larger in-memory hash tables for the MERGE join, avoiding disk-spilling sort/merge strategies on large deltas. This setting is only applied when both merge_planner_hints = true and the delta exceeds 10,000 rows.

Tuning Guidance:

• Servers with ample RAM (32+ GB): Increase to 128–256 for faster large-delta refreshes.
• Memory-constrained: Lower to 16–32 or disable planner hints entirely.
• Very large deltas (100K+ rows): Consider 256–512 if refresh latency matters.

SET pg_stream.merge_work_mem_mb = 128;

pg_stream.merge_strategy

Strategy for applying delta changes to the stream table during differential refresh.

Values:

• auto (default): Use MERGE for small deltas; switch to DELETE + INSERT when the delta exceeds 25% of the stream table row count.
• merge: Always use a single MERGE statement. Best for correctness and simplicity.
• delete_insert: Always use DELETE + INSERT. May be faster for very large deltas but has known limitations with aggregate/DISTINCT queries.

Tuning Guidance:

• Most workloads: Leave at 'auto' or 'merge'.
• Batch-heavy ETL: Try 'delete_insert' if MERGE performance degrades on large deltas, but test thoroughly with your specific queries.

-- Always use MERGE
SET pg_stream.merge_strategy = 'merge';

-- Always use DELETE + INSERT
SET pg_stream.merge_strategy = 'delete_insert';

pg_stream.use_prepared_statements

Use SQL PREPARE / EXECUTE for MERGE statements during differential refresh.

When enabled, the refresh executor issues PREPARE __pgs_merge_{id} on the first cache-hit cycle, then uses EXECUTE on subsequent cycles. After approximately 5 executions, PostgreSQL switches from a custom plan to a generic plan, saving 1–2 ms of parse/plan overhead per refresh.

Tuning Guidance:

• Most workloads: Leave at true — the cumulative parse/plan savings are significant for frequently-refreshed stream tables.
• Highly skewed data: Set to false if prepared-statement parameter sniffing produces poor plans (e.g., highly skewed LSN distributions causing bad join estimates).

-- Disable prepared statements
SET pg_stream.use_prepared_statements = false;

pg_stream.user_triggers

Control how user-defined triggers on stream tables are handled during refresh.

When a stream table has user-defined row-level triggers, the refresh engine can decompose the MERGE into explicit DELETE + UPDATE + INSERT statements so triggers fire with correct TG_OP, OLD, and NEW values.

Values:

• auto (default): Automatically detect user triggers on the stream table. If present, use the explicit DML path; otherwise use MERGE.
• on: Always use the explicit DML path, even without user triggers. Useful for testing.
• off: Always use MERGE. User triggers are suppressed during refresh. This is the escape hatch if explicit DML causes issues.

Notes:

• Row-level triggers do not fire during FULL refresh regardless of this setting. FULL refresh uses DISABLE TRIGGER USER / ENABLE TRIGGER USER to suppress them.
• The explicit DML path adds ~25–60% overhead compared to MERGE for affected stream tables.
• Stream tables without user triggers have zero overhead when using auto (only a fast pg_trigger check).

-- Auto-detect (default)
SET pg_stream.user_triggers = 'auto';

-- Always use explicit DML (for testing)
SET pg_stream.user_triggers = 'on';

-- Suppress triggers, use MERGE
SET pg_stream.user_triggers = 'off';

pg_stream.block_source_ddl

When enabled, column-affecting DDL (e.g., ALTER TABLE ... DROP COLUMN, ALTER TABLE ... ALTER COLUMN ... TYPE) on source tables tracked by stream tables is blocked with an ERROR instead of silently marking stream tables for reinitialization.

This is useful in production environments where you want to prevent accidental schema changes that would trigger expensive full recomputation of downstream stream tables.

Default: false
Context: Superuser

-- Block column-affecting DDL on tracked source tables
SET pg_stream.block_source_ddl = true;

-- Allow DDL (stream tables will be marked for reinit instead)
SET pg_stream.block_source_ddl = false;

Note: Only column-affecting changes are blocked. Benign DDL (adding indexes, comments, constraints) is always allowed regardless of this setting.

pg_stream.cdc_mode

CDC (Change Data Capture) mechanism selection.

Default: 'trigger'

-- Always use triggers (default, zero-config)
SET pg_stream.cdc_mode = 'trigger';

-- Enable automatic trigger → WAL transition
SET pg_stream.cdc_mode = 'auto';

-- Require WAL-based CDC (error if wal_level != logical)
SET pg_stream.cdc_mode = 'wal';

pg_stream.wal_transition_timeout

Maximum time (seconds) to wait for the WAL decoder to catch up during the transition from trigger-based to WAL-based CDC. If the decoder has not caught up within this timeout, the system falls back to triggers.

Default: 300 (5 minutes)
Range: 10 – 3600

SET pg_stream.wal_transition_timeout = 300;

Complete postgresql.conf Example

# Required
shared_preload_libraries = 'pg_stream'

# Optional tuning
pg_stream.enabled = true
pg_stream.scheduler_interval_ms = 1000
pg_stream.min_schedule_seconds = 60
pg_stream.max_consecutive_errors = 3
pg_stream.change_buffer_schema = 'pgstream_changes'
pg_stream.max_concurrent_refreshes = 4
pg_stream.differential_max_change_ratio = 0.15
pg_stream.cleanup_use_truncate = true
pg_stream.merge_planner_hints = true
pg_stream.merge_work_mem_mb = 64
pg_stream.merge_strategy = 'auto'
pg_stream.use_prepared_statements = true
pg_stream.user_triggers = 'auto'
pg_stream.block_source_ddl = false
pg_stream.cdc_mode = 'trigger'
pg_stream.wal_transition_timeout = 300

Runtime Configuration

All GUC variables can be changed at runtime by a superuser:

-- View current settings
SHOW pg_stream.enabled;
SHOW pg_stream.scheduler_interval_ms;

-- Change for current session
SET pg_stream.max_concurrent_refreshes = 8;

-- Change persistently (requires reload)
ALTER SYSTEM SET pg_stream.scheduler_interval_ms = 500;
SELECT pg_reload_conf();

Further Reading

• INSTALL.md — Installation and initial configuration
• ARCHITECTURE.md — System architecture overview
• SQL_REFERENCE.md — Complete function reference

Installation Guide

Prerequisites

Building from source additionally requires Rust 1.82+ and pgrx 0.17.x. Pre-built release artifacts only need a running PostgreSQL 18.x instance.

Installing from a Pre-built Release

1. Download the release archive

Download the archive for your platform from the GitHub Releases page:

Optionally verify the checksum against SHA256SUMS.txt from the same release:

sha256sum -c SHA256SUMS.txt

2. Extract and install

Linux / macOS:

tar xzf pg_stream-0.1.1-pg18-linux-amd64.tar.gz
cd pg_stream-0.1.1-pg18-linux-amd64

sudo cp lib/*.so  "$(pg_config --pkglibdir)/"
sudo cp extension/*.control extension/*.sql "$(pg_config --sharedir)/extension/"

Windows (PowerShell):

Expand-Archive pg_stream-0.1.1-pg18-windows-amd64.zip -DestinationPath .
cd pg_stream-0.1.1-pg18-windows-amd64

Copy-Item lib\*.dll  "$(pg_config --pkglibdir)\"
Copy-Item extension\* "$(pg_config --sharedir)\extension\"

3. Using with CloudNativePG (Kubernetes)

pg_stream is distributed as an OCI extension image for use with CloudNativePG Image Volume Extensions.

Requirements: Kubernetes 1.33+, CNPG 1.28+, PostgreSQL 18.

# Pull the extension image
docker pull ghcr.io/grove/pg_stream-ext:0.1.1

See cnpg/cluster-example.yaml and cnpg/database-example.yaml for complete Cluster and Database deployment examples.

4. Local Docker development (without Kubernetes)

For local development without Kubernetes, install the extension files manually into a standard PostgreSQL container from a release archive:

# Extract extension files from the release archive
tar xzf pg_stream-0.1.1-pg18-linux-amd64.tar.gz
cd pg_stream-0.1.1-pg18-linux-amd64

# Run PostgreSQL with the extension mounted
docker run --rm \
  -v $PWD/lib/pg_stream.so:/usr/lib/postgresql/18/lib/pg_stream.so:ro \
  -v $PWD/extension/:/tmp/ext/:ro \
  -e POSTGRES_PASSWORD=postgres \
  postgres:18.1 \
  sh -c 'cp /tmp/ext/* /usr/share/postgresql/18/extension/ && \
         exec postgres -c shared_preload_libraries=pg_stream'

Building from Source

1. Install Rust

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

2. Install pgrx

cargo install --locked cargo-pgrx --version 0.17.0
cargo pgrx init --pg18 $(pg_config --bindir)/pg_config

3. Build the Extension

# Development build (faster compilation)
cargo pgrx install --pg-config $(pg_config --bindir)/pg_config

# Release build (optimized, for production)
cargo pgrx install --release --pg-config $(pg_config --bindir)/pg_config

# Package for deployment (creates installable artifacts)
cargo pgrx package --pg-config $(pg_config --bindir)/pg_config

PostgreSQL Configuration

Add the following to postgresql.conf before starting PostgreSQL:

# Required — loads the extension shared library at server start
shared_preload_libraries = 'pg_stream'

# Recommended — must accommodate scheduler + refresh workers
max_worker_processes = 8

Note: wal_level = logical and max_replication_slots are not required. The extension uses lightweight row-level triggers for CDC, not logical replication.

Restart PostgreSQL after modifying these settings:

pg_ctl restart -D /path/to/data
# or
systemctl restart postgresql

Extension Installation

Connect to the target database and run:

CREATE EXTENSION pg_stream;

This creates:

• The pgstream schema with catalog tables and SQL functions
• The pgstream_changes schema for change buffer tables
• Event triggers for DDL tracking
• The pgstream.pg_stat_stream_tables monitoring view

Verification

After installation, verify everything is working:

-- Check the extension version
SELECT extname, extversion FROM pg_extension WHERE extname = 'pg_stream';

-- Or get a full status overview (includes version, scheduler state, stream table count)
SELECT * FROM pgstream.pgs_status();

Inspecting the installation

-- Check the installed version
SELECT extversion FROM pg_extension WHERE extname = 'pg_stream';

-- Check which schemas were created
SELECT schema_name
FROM information_schema.schemata
WHERE schema_name IN ('pgstream', 'pgstream_changes');

-- Check all registered GUC variables
SHOW pg_stream.enabled;
SHOW pg_stream.scheduler_interval_ms;
SHOW pg_stream.max_concurrent_refreshes;

-- Check the scheduler background worker is running
SELECT * FROM pgstream.pgs_status();

-- List all stream tables
SELECT pgs_schema, pgs_name, status, refresh_mode, is_populated
FROM pgstream.pgs_stream_tables;

-- Check that the shared library loaded correctly
SELECT * FROM pg_extension WHERE extname = 'pg_stream';

-- Verify the catalog tables exist
SELECT tablename
FROM pg_tables
WHERE schemaname = 'pgstream'
ORDER BY tablename;

Quick functional test

CREATE TABLE test_source (id INT PRIMARY KEY, val TEXT);
INSERT INTO test_source VALUES (1, 'hello');

SELECT pgstream.create_stream_table(
    'test_st',
    'SELECT id, val FROM test_source',
    '1m',
    'FULL'
);

SELECT * FROM test_st;
-- Should return: 1 | hello

-- Clean up
SELECT pgstream.drop_stream_table('test_st');
DROP TABLE test_source;

Uninstallation

-- Drop all stream tables first
SELECT pgstream.drop_stream_table(pgs_schema || '.' || pgs_name)
FROM pgstream.pgs_stream_tables;

-- Drop the extension
DROP EXTENSION pg_stream CASCADE;

Remove pg_stream from shared_preload_libraries in postgresql.conf and restart PostgreSQL.

Troubleshooting

Unit tests crash on macOS 26+ (symbol not found in flat namespace)

macOS 26 (Tahoe) changed dyld to eagerly resolve all flat-namespace symbols at binary load time. pgrx extensions reference PostgreSQL server-internal symbols (e.g. CacheMemoryContext, SPI_connect) via the -Wl,-undefined,dynamic_lookup linker flag. These symbols are normally provided by the postgres executable when the extension is loaded as a shared library — but for cargo test --lib there is no postgres process, so the test binary aborts immediately:

dyld[66617]: symbol not found in flat namespace '_CacheMemoryContext'

This affects local development only — integration tests, E2E tests, and the extension itself running inside PostgreSQL are unaffected.

The fix is built into the just test-unit recipe. It automatically:

1. Compiles a tiny C stub library (scripts/pg_stub.c → target/libpg_stub.dylib) that provides NULL/no-op definitions for the ~28 PostgreSQL symbols.
2. Compiles the test binary with --no-run.
3. Runs the binary with DYLD_INSERT_LIBRARIES pointing to the stub.

The stub is only built on macOS 26+. On Linux or older macOS, just test-unit runs cargo test --lib directly with no changes.

Note: The stub symbols are never called — unit tests exercise pure Rust logic only. If a test accidentally calls a PostgreSQL function it will crash with a NULL dereference (the desired fail-fast behavior).

If you run unit tests without just (e.g. directly via cargo test --lib), you can use the wrapper script instead:

./scripts/run_unit_tests.sh pg18

# With test name filter:
./scripts/run_unit_tests.sh pg18 -- test_parse_basic

Extension fails to load

Ensure shared_preload_libraries = 'pg_stream' is set and PostgreSQL has been restarted (not just reloaded). The extension requires shared memory initialization at startup.

Background worker not starting

Check that max_worker_processes is high enough to accommodate the scheduler worker plus any refresh workers. The default of 8 is usually sufficient.

Check logs for details

The extension logs at various levels. Enable debug logging for more detail:

SET client_min_messages TO debug1;

Architecture

This document describes the internal architecture of pg_stream — a PostgreSQL 18 extension that implements stream tables with differential view maintenance. For a high-level description of what pg_stream does and why, read ESSENCE.md. For release milestones and future plans, see ROADMAP.md.

High-Level Overview

┌─────────────────────────────────────────────────────────────────┐
│                     PostgreSQL 18 Backend                       │
│                                                                 │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌─────────────┐   │
│  │  Source  │   │  Source  │   │  Storage │   │  Storage    │   │
│  │  Table A │   │  Table B │   │  Table X │   │  Table Y    │   │
│  └────┬─────┘   └────┬─────┘   └────▲─────┘   └────▲────────┘   │
│       │              │              │              │            │
│  ═════╪══════════════╪══════════════╪══════════════╪════════    │
│       │              │              │              │            │
│  ┌────▼──────────────▼────┐   ┌────┴──────────────┴────┐        │
│  │  Hybrid CDC Layer      │   │  Delta Application     │        │
│  │  Triggers ──or── WAL   │   │  (INSERT/DELETE diffs) │        │
│  └────────────┬───────────┘   └────────────▲───────────┘        │
│               │                            │                    │
│  ┌────────────▼───────────┐   ┌────────────┴───────────┐        │
│  │   Change Buffer        │   │   DVM Engine           │        │
│  │   (pgstream_changes.*) │   │   (Operator Tree)      │        │
│  └────────────┬───────────┘   └────────────▲───────────┘        │
│               │                            │                    │
│               └────────────┬───────────────┘                    │
│                            │                                    │
│  ┌─────────────────────────▼─────────────────────────────┐      │
│  │              Refresh Engine                           │      │
│  │  ┌──────────┐  ┌──────────┐  ┌─────────────────────┐  │      │
│  │  │ Frontier │  │ DAG      │  │ Scheduler           │  │      │
│  │  │ Tracker  │  │ Resolver │  │ (canonical schedule)│  │      │
│  │  └──────────┘  └──────────┘  └─────────────────────┘  │      │
│  └───────────────────────────────────────────────────────┘      │
│                                                                 │
│  ┌────────────────────────────────────────────────────────┐     │
│  │                    Catalog (pgstream.*)                │     │
│  │  pgs_stream_tables │ pgs_dependencies │ pgs_refresh_history│  │
│  └────────────────────────────────────────────────────────┘     │
│                                                                 │
│  ┌──────────────────────────────────────────────────────┐       │
│  │                  Monitoring Layer                    │       │
│  │  st_refresh_stats │ slot_health │ check_cdc_health    │       │
│  │  explain_st │ views │ NOTIFY alerting               │       │
│  └──────────────────────────────────────────────────────┘       │
└─────────────────────────────────────────────────────────────────┘

Component Details

1. SQL API Layer (src/api.rs)

The public entry point for users. All operations are exposed as #[pg_extern] functions in the pgstream schema:

• create_stream_table — Applies a chain of auto-rewrite passes (view inlining → DISTINCT ON → GROUPING SETS → scalar subquery → SubLinks in OR → multi-PARTITION BY windows), parses the defining query, builds an operator tree, creates the storage table, registers CDC slots, populates the catalog, and optionally performs an initial full refresh.
• alter_stream_table — Modifies schedule, refresh mode, or status (ACTIVE/SUSPENDED).
• drop_stream_table — Removes the storage table, catalog entries, and cleans up CDC slots.
• refresh_stream_table — Triggers a manual refresh (same path as automatic scheduling).
• pgs_status — Returns a summary of all registered stream tables.

2. Catalog (src/catalog.rs)

The catalog manages persistent metadata stored in PostgreSQL tables within the pgstream schema:

Schema creation is handled by extension_sql!() macros that run at CREATE EXTENSION time.

Entity-Relationship Diagram

erDiagram
    pgs_stream_tables {
        bigserial pgs_id PK
        oid pgs_relid UK "OID of materialized storage table"
        text pgs_name
        text pgs_schema
        text defining_query
        text original_query "User's original SQL (pre-inlining)"
        text schedule "Duration or cron expression"
        text refresh_mode "FULL | DIFFERENTIAL"
        text status "INITIALIZING | ACTIVE | SUSPENDED | ERROR"
        boolean is_populated
        timestamptz data_timestamp "Freshness watermark"
        jsonb frontier "DBSP-style version frontier"
        timestamptz last_refresh_at
        int consecutive_errors
        boolean needs_reinit
        float8 auto_threshold
        float8 last_full_ms
        timestamptz created_at
        timestamptz updated_at
    }

    pgs_dependencies {
        bigint pgs_id PK,FK "References pgs_stream_tables.pgs_id"
        oid source_relid PK "OID of source table"
        text source_type "TABLE | STREAM_TABLE | VIEW"
        text_arr columns_used "Column-level lineage"
        text cdc_mode "TRIGGER | TRANSITIONING | WAL"
        text slot_name "Replication slot (WAL mode)"
        pg_lsn decoder_confirmed_lsn "WAL decoder progress"
        timestamptz transition_started_at "Trigger→WAL transition start"
    }

    pgs_refresh_history {
        bigserial refresh_id PK
        bigint pgs_id FK "References pgs_stream_tables.pgs_id"
        timestamptz data_timestamp
        timestamptz start_time
        timestamptz end_time
        text action "NO_DATA | FULL | DIFFERENTIAL | REINITIALIZE | SKIP"
        bigint rows_inserted
        bigint rows_deleted
        text error_message
        text status "RUNNING | COMPLETED | FAILED | SKIPPED"
        text initiated_by "SCHEDULER | MANUAL | INITIAL"
        timestamptz freshness_deadline
    }

    pgs_change_tracking {
        oid source_relid PK "OID of tracked source table"
        text slot_name "Trigger function name"
        pg_lsn last_consumed_lsn
        bigint_arr tracked_by_pgs_ids "ST IDs sharing this source"
    }

    pgs_stream_tables ||--o{ pgs_dependencies : "has sources"
    pgs_stream_tables ||--o{ pgs_refresh_history : "has refresh history"
    pgs_stream_tables }o--o{ pgs_change_tracking : "tracks via pgs_ids array"

Note: Change buffer tables (pgstream_changes.changes_<oid>) are created dynamically per source table OID and live in the separate pgstream_changes schema.

3. CDC / Change Data Capture (src/cdc.rs, src/wal_decoder.rs)

pg_stream uses a hybrid CDC architecture that starts with triggers and optionally transitions to WAL-based (logical replication) capture for lower write-side overhead.

Trigger Mode (default)

1. Trigger Management — Creates AFTER INSERT OR UPDATE OR DELETE row-level triggers (pg_stream_cdc_<oid>) on each tracked source table. Each trigger fires a PL/pgSQL function (pg_stream_cdc_fn_<oid>()) that writes changes to the buffer table.
2. Change Buffering — Decoded changes are written to per-source change buffer tables in the pgstream_changes schema. Each row captures the LSN (pg_current_wal_lsn()), transaction ID, action type (I/U/D), and the new/old row data as JSONB via to_jsonb().
3. Cleanup — Consumed changes are deleted after each successful refresh via delete_consumed_changes(), bounded by the upper LSN to prevent unbounded scans.
4. Lifecycle — Triggers and trigger functions are automatically created when a source table is first tracked and dropped when the last stream table referencing a source is removed.

The trigger approach was chosen as the default for transaction safety (triggers can be created in the same transaction as DDL), simplicity (no slot management, no wal_level = logical requirement), and immediate visibility (changes are visible in buffer tables as soon as the source transaction commits).

WAL Mode (optional, automatic transition)

When pg_stream.cdc_mode is set to 'auto' or 'wal' and wal_level = logical is available, the system transitions from trigger-based to WAL-based CDC after the first successful refresh:

1. WAL Availability Detection — At stream table creation, checks whether wal_level = logical is configured. If so, the source dependency is marked for WAL transition.
2. WAL Decoder Background Worker — A dedicated background worker (src/wal_decoder.rs) polls logical replication slots and writes decoded changes into the same change buffer tables used by triggers, ensuring a uniform format for the DVM engine.
3. Transition Orchestration — The transition is a three-step process: (a) create a replication slot, (b) wait for the decoder to catch up to the trigger's last confirmed LSN, (c) drop the trigger and switch the dependency to WAL mode. If the decoder doesn't catch up within pg_stream.wal_transition_timeout (default 300s), the system falls back to triggers.
4. CDC Mode Tracking — Each source dependency in pgs_dependencies carries a cdc_mode column (TRIGGER / TRANSITIONING / WAL) and WAL-specific metadata (slot_name, decoder_confirmed_lsn, transition_started_at).

See ADR-001 and ADR-002 in plans/adrs/PLAN_ADRS.md for the original design rationale and plans/sql/PLAN_HYBRID_CDC.md for the full implementation plan.

4. DVM Engine (src/dvm/)

The Differential View Maintenance engine is the core of the system. It transforms the defining SQL query into an executable operator tree that can compute deltas efficiently.

Auto-Rewrite Pipeline (src/dvm/parser.rs)

Before the defining query is parsed into an operator tree, it passes through a chain of auto-rewrite passes that normalize SQL constructs the DVM parser doesn't handle directly:

The view inlining pass (#0) runs first so that view definitions containing DISTINCT ON, GROUPING SETS, etc. are further rewritten by downstream passes. Nested views are expanded via a fixpoint loop (max depth 10).

Query Parser (src/dvm/parser.rs)

Parses the defining query using PostgreSQL's internal parser (via pgrx raw_parser) and extracts:

• WITH clause — CTE definitions (non-recursive: inline expansion or shared delta; recursive: detected for mode gating)
• Target list — output columns
• FROM clause — source tables, joins, subqueries, and CTE references
• WHERE clause — filters
• GROUP BY / aggregate functions
• DISTINCT / UNION ALL / INTERSECT / EXCEPT

The parser produces an OpTree — a tree of operator nodes. CTE handling follows a tiered approach:

1. Tier 1 (Inline Expansion) — Non-recursive CTEs referenced once are expanded into Subquery nodes, equivalent to subqueries in FROM.
2. Tier 2 (Shared Delta) — Non-recursive CTEs referenced multiple times produce CteScan nodes that share a single delta computation via a CTE registry and delta cache.
3. Tier 3a/3b/3c (Recursive) — Recursive CTEs (WITH RECURSIVE) are detected via query_has_recursive_cte(). In FULL mode, the query executes as-is. In DIFFERENTIAL mode, the strategy is auto-selected: semi-naive evaluation for INSERT-only changes, Delete-and-Rederive (DRed) for mixed changes, or recomputation fallback when CTE columns don't match ST storage.

Operators (src/dvm/operators/)

Each operator knows how to generate a delta query — given a set of changes to its inputs, it produces the corresponding changes to its output:

See DVM_OPERATORS.md for detailed descriptions.

Diff Engine (src/dvm/diff.rs)

Generates the final diff SQL that:

1. Computes the delta from the operator tree
2. Produces ('+', row) for inserts and ('-', row) for deletes
3. Applies the diff via DELETE matching old rows and INSERT for new rows

5. DAG / Dependency Graph (src/dag.rs)

Stream tables can depend on other stream tables (cascading), forming a Directed Acyclic Graph:

• Cycle detection — Prevents circular dependencies at creation time using DFS.
• Topological ordering — Determines refresh order: upstream STs must be refreshed before downstream STs.
• Cascade operations — When a source table changes, all transitive dependents are identified for refresh.

6. Version / Frontier Tracking (src/version.rs)

Implements a per-source frontier (JSONB map of source_oid → LSN) to track exactly how far each stream table has consumed changes:

• Read frontier — Before refresh, read the frontier to know where to start consuming changes.
• Advance frontier — After a successful refresh, the frontier is updated to the latest consumed LSN.
• Consistent snapshots — The frontier ensures that each refresh processes a contiguous, non-overlapping window of changes.

Delayed View Semantics (DVS) Guarantee

The contents of every stream table are logically equivalent to evaluating its defining query at some past point in time — the data_timestamp. The scheduler refreshes STs in topological order so that when ST B references upstream ST A, A has already been refreshed to the target data_timestamp before B runs its delta query against A's contents. The frontier lifecycle is:

1. Created — on first full refresh; records the LSN of each source at that moment.
2. Advanced — on each differential refresh; the old frontier becomes the lower bound and the new frontier (with fresh LSNs) the upper bound. The DVM engine reads changes in [old, new].
3. Reset — on reinitialize; a fresh frontier is created from scratch.

7. Refresh Engine (src/refresh.rs)

Orchestrates the complete refresh cycle:

┌──────────────┐
│  Check State │ → Is ST active? Has it been populated?
└──────┬───────┘
       │
 ┌─────▼──────┐
 │ Drain CDC  │ → Read WAL changes into change buffer tables
 └─────┬──────┘
       │
 ┌─────▼──────────────┐
 │ Determine Action   │ → FULL, DIFFERENTIAL, NO_DATA, REINITIALIZE, or SKIP?
 │                    │   (adaptive: if change ratio > pg_stream.differential_max_change_ratio,
 │                    │    downgrade DIFFERENTIAL → FULL automatically)
 └─────┬──────────────┘
       │
 ┌─────▼──────┐
 │ Execute    │ → Full: TRUNCATE + INSERT ... SELECT
 │            │   Differential: Generate & apply delta SQL
 └─────┬──────┘
       │
 ┌─────▼──────────────┐
 │ Record History     │ → Write to pgstream.pgs_refresh_history
 └─────┬──────────────┘
       │
 ┌─────▼──────────────┐
 │ Advance Frontier   │ → Update JSONB frontier in catalog
 └─────┬──────────────┘
       │
 ┌─────▼──────────────┐
 │ Reset Error Count  │ → On success, reset consecutive_errors to 0
 └──────────────────────┘

8. Background Worker & Scheduling (src/scheduler.rs)

Registration & Lifecycle

pg_stream registers one PostgreSQL background worker — the scheduler — during _PG_init() (extension load). Because it is registered at startup, pg_stream must appear in shared_preload_libraries, which requires a server restart.

┌──────────────────────────────────────────────────────────────────┐
│                  PostgreSQL postmaster                           │
│                                                                  │
│  shared_preload_libraries = 'pg_stream'                          │
│       │                                                          │
│       ▼                                                          │
│  _PG_init()                                                      │
│    ├─ Register GUCs (pg_stream.enabled, scheduler_interval_ms …) │
│    ├─ Register shared memory (PgStreamSharedState, atomics)      │
│    └─ BackgroundWorkerBuilder::new("pg_stream scheduler")        │
│         .set_start_time(RecoveryFinished)                        │
│         .set_restart_time(5s)       ← auto-restart on crash      │
│         .load()                                                  │
│                                                                  │
│  After recovery finishes:                                        │
│       │                                                          │
│       ▼                                                          │
│  pg_stream_scheduler_main()         ← background worker starts   │
│    ├─ Attach SIGHUP + SIGTERM handlers                           │
│    ├─ Connect to SPI (database = "postgres")                     │
│    ├─ Crash recovery: mark stale RUNNING records as FAILED       │
│    └─ Enter main loop ─────────────────────────┐                 │
│         │                                      │                 │
│         ▼                                      │                 │
│     wait_latch(scheduler_interval_ms)          │                 │
│         │                                      │                 │
│     ┌───▼───────────────────────────────┐      │                 │
│     │ SIGTERM? → log + break            │      │                 │
│     │ pg_stream.enabled = false? → skip │      │                 │
│     │ Otherwise → scheduler tick        │      │                 │
│     └───┬───────────────────────────────┘      │                 │
│         │                                      │                 │
│         └──────────── loop ────────────────────┘                 │
└──────────────────────────────────────────────────────────────────┘

Key lifecycle properties:

Scheduler Tick

Each tick of the main loop performs the following steps inside a single transaction:

1. DAG rebuild — Compare the shared-memory DAG_REBUILD_SIGNAL counter against the local copy. If it advanced (a CREATE, ALTER, or DROP stream table occurred), rebuild the in-memory dependency graph (StDag) from the catalog.
2. Topological traversal — Walk stream tables in dependency order (upstream before downstream). This ensures that when ST B references ST A, A is refreshed first.
3. Per-ST evaluation — For each active ST:
   • Skip if in retry backoff (exponential, per-ST).
   • Skip if schedule/cron says not yet due.
   • Skip if an advisory lock indicates a concurrent refresh.
   • Check upstream change buffers for pending rows.
4. Execute refresh — Acquire an advisory lock → record RUNNING in history → run FULL / DIFFERENTIAL / REINITIALIZE → store new frontier → release lock → record completion.
5. WAL transitions — Advance any trigger→WAL CDC mode transitions (src/wal_decoder.rs).
6. Slot health — Check replication slot health and emit NOTIFY alerts.
7. Prune retry state — Remove backoff entries for STs that no longer exist.

Sequential Processing

The scheduler processes stream tables sequentially within a single background worker. Although pg_stream.max_concurrent_refreshes (default 4) exists as a GUC, it currently only prevents a manual pgstream.refresh_stream_table() call from overlapping with the scheduler on the same ST — it does not spawn additional workers. All STs are refreshed one at a time in topological order.

The PostgreSQL GUC max_worker_processes (default 8) sets the server-wide budget for all background workers (autovacuum, parallel query, logical replication, extensions). pg_stream consumes one slot from that budget.

Retry & Error Handling

Each ST maintains an in-memory RetryState (reset on scheduler restart):

• Retryable errors (SPI failures, lock contention, slot issues) trigger exponential backoff.
• Permanent errors (schema mismatch, user errors) skip backoff but increment consecutive_errors.
• When consecutive_errors reaches pg_stream.max_consecutive_errors (default 3), the ST is auto-suspended and a NOTIFY alert is emitted.
• Schema errors additionally set needs_reinit, triggering a REINITIALIZE on the next successful cycle.

Scheduling Policy

Automatic refresh scheduling uses canonical periods (48·2ⁿ seconds, n = 0, 1, 2, …) snapped to the user's schedule:

• Picks the smallest canonical period ≤ schedule.
• For DOWNSTREAM schedule (NULL schedule), the ST refreshes only when explicitly triggered or when a downstream ST needs it.
• Advisory locks prevent concurrent refreshes of the same ST.
• The scheduler is driven by the background worker polling at the pg_stream.scheduler_interval_ms GUC interval.

Shared Memory (src/shmem.rs)

The scheduler background worker and user sessions share a PgStreamSharedState structure protected by a PgLwLock. Key fields:

A separate PgAtomic<AtomicU64> named DAG_REBUILD_SIGNAL is incremented by API functions (create, alter, drop) after catalog mutations. The scheduler compares its local copy against the atomic counter to detect when to rebuild its in-memory DAG without holding a lock.

A second PgAtomic<AtomicU64> named CACHE_GENERATION tracks DDL events that may invalidate cached delta or MERGE templates across backends. When DDL hooks fire (view change, ALTER TABLE, function change) or API functions mutate the catalog, CACHE_GENERATION is bumped. Each backend maintains a thread-local generation counter; on the next refresh, if the shared generation has advanced, the backend flushes its delta template cache, MERGE template cache, and prepared statements.

9. DDL Tracking (src/hooks.rs)

Event triggers monitor DDL changes to source tables and functions:

• _on_ddl_end — Fires on ALTER TABLE to detect column adds/drops/type changes. If a source table used by a ST is altered, the ST's needs_reinit flag is set. Also detects CREATE OR REPLACE FUNCTION / ALTER FUNCTION — if the function appears in a ST's functions_used catalog column, the ST is marked for reinit.
• _on_sql_drop — Fires on DROP TABLE to set needs_reinit for affected STs. Also detects DROP FUNCTION and marks affected STs for reinit.
• Function name extraction — object_identity strings (e.g., public.my_func(integer, text)) are parsed to extract the bare function name, which is matched against the functions_used TEXT[] column in pgs_stream_tables.

Reinitialization is deferred until the next refresh cycle, which then performs a REINITIALIZE action (drop and recreate the storage table from the updated query).

10. Error Handling (src/error.rs)

Centralized error types using thiserror:

• PgStreamError variants cover catalog access, SQL execution, CDC, DVM, DAG, and config errors.
• Each refresh failure increments consecutive_errors.
• When consecutive_errors reaches pg_stream.max_consecutive_errors (default 3), the ST is moved to ERROR status and suspended from automatic refresh.
• Manual intervention (ALTER ... status => 'ACTIVE') resets the counter.

11. Monitoring (src/monitor.rs)

Provides observability functions:

• st_refresh_stats — Aggregate statistics (total/successful/failed refreshes, avg duration, staleness status).
• get_refresh_history — Per-ST audit trail.
• get_staleness — Current staleness in seconds.
• slot_health — Checks replication slot state and WAL retention.
• check_cdc_health — Per-source CDC health status including mode, slot lag, confirmed LSN, and alerts.
• explain_st — Describes the DVM plan for a given ST.
• Views — pgstream.stream_tables_info (computed staleness) and pgstream.pg_stat_stream_tables (combined stats).

NOTIFY Alerting

Operational events are broadcast via PostgreSQL NOTIFY on the pg_stream_alert channel. Clients can subscribe with LISTEN pg_stream_alert; and receive JSON-formatted events:

12. Row ID Hashing (src/hash.rs)

Provides deterministic 64-bit row identifiers using xxHash (xxh64) with a fixed seed. Two SQL functions are exposed:

• pgstream.pg_stream_hash(text) — Hash a single text value; used for simple single-column row IDs.
• pgstream.pg_stream_hash_multi(text[]) — Hash multiple values (separated by a record-separator byte \x1E) for composite keys (join row IDs, GROUP BY keys).

Row IDs are written into every stream table's storage as an internal __pgs_row_id BIGINT column and are used by the delta application phase to match DELETE candidates precisely.

13. Configuration (src/config.rs)

Twelve GUC (Grand Unified Configuration) variables control runtime behavior, plus five performance-tuning GUCs. See CONFIGURATION.md for details.

Data Flow: End-to-End Refresh

 Source Table INSERT/UPDATE/DELETE
           │
           ▼
 Hybrid CDC Layer:
   ┌─────────────────────────────────────────────┐
   │ TRIGGER mode: Row-Level AFTER Trigger        │
   │   pg_stream_cdc_fn_<oid>() → buffer table    │
   │                                              │
   │ WAL mode: Logical Replication Slot           │
   │   wal_decoder bgworker → same buffer table   │
   └─────────────────────────────────────────────┘
           │
           ▼
 Change Buffer Table (pgstream_changes.changes_<oid>)
   Columns: change_id, lsn, xid, action (I/U/D), row_data (jsonb)
           │
           ▼
 DVM Engine: generate delta SQL from operator tree
   - Scan operator reads from change buffer
   - Filter/Project/Join transform the deltas
   - Aggregate recomputes affected groups
           │
           ▼
 Diff Engine: produce (+/-) diff rows
           │
           ▼
 Delta Application:
   DELETE FROM storage WHERE __pgs_row_id IN (removed)
   INSERT INTO storage SELECT ... FROM (added)
           │
           ▼
 Frontier Update: advance per-source LSN
           │
           ▼
 History Record: log to pgstream.pgs_refresh_history

Module Map

src/
├── lib.rs           # Extension entry, module declarations, _PG_init
├── bin/
│   └── pgrx_embed.rs# pgrx SQL entity embedding (generated)
├── api.rs           # SQL API functions (create/alter/drop/refresh/status)
├── catalog.rs       # Catalog CRUD operations
├── cdc.rs           # Change data capture (triggers + WAL transition)
├── config.rs        # GUC variable registration
├── dag.rs           # Dependency graph (cycle detection, topo sort)
├── error.rs         # Centralized error types
├── hash.rs          # xxHash row ID generation (pg_stream_hash / pg_stream_hash_multi)
├── hooks.rs         # DDL event trigger handlers (_on_ddl_end, _on_sql_drop)
├── shmem.rs         # Shared memory state (PgStreamSharedState, DAG_REBUILD_SIGNAL, CACHE_GENERATION)
├── dvm/
│   ├── mod.rs       # DVM module root + recursive CTE orchestration
│   ├── parser.rs    # Query → OpTree converter (CTE extraction, subquery, window support)
│   ├── diff.rs      # Delta SQL generation (CTE delta cache)
│   ├── row_id.rs    # Row ID generation
│   └── operators/
│       ├── mod.rs           # Operator trait + registry
│       ├── scan.rs          # Table scan (CDC passthrough)
│       ├── filter.rs        # WHERE clause filtering
│       ├── project.rs       # Column projection
│       ├── join.rs          # Inner join
│       ├── join_common.rs   # Shared join utilities (snapshot subqueries, column disambiguation)
│       ├── outer_join.rs    # LEFT/RIGHT outer join
│       ├── full_join.rs     # FULL OUTER JOIN (8-part delta)
│       ├── aggregate.rs     # GROUP BY + aggregate functions (39 AggFunc variants)
│       ├── distinct.rs      # DISTINCT deduplication
│       ├── union_all.rs     # UNION ALL merging
│       ├── intersect.rs     # INTERSECT / INTERSECT ALL (dual-count LEAST)
│       ├── except.rs        # EXCEPT / EXCEPT ALL (dual-count GREATEST)
│       ├── subquery.rs      # Subquery / inlined CTE delegation
│       ├── cte_scan.rs      # Shared CTE delta (multi-reference)
│       ├── recursive_cte.rs # Recursive CTE (semi-naive + DRed + recomputation)
│       ├── window.rs        # Window function (partition recomputation)
│       ├── lateral_function.rs  # LATERAL SRF (row-scoped recomputation)
│       ├── lateral_subquery.rs  # LATERAL correlated subquery
│       ├── semi_join.rs     # EXISTS / IN subquery (semi-join delta)
│       ├── anti_join.rs     # NOT EXISTS / NOT IN subquery (anti-join delta)
│       └── scalar_subquery.rs   # Correlated scalar subquery in SELECT
├── monitor.rs       # Monitoring & observability functions
├── refresh.rs       # Refresh orchestration
├── scheduler.rs     # Automatic scheduling with canonical periods
├── version.rs       # Frontier / LSN tracking
└── wal_decoder.rs   # WAL-based CDC (logical replication slot polling, transitions)

Extension Control File (pg_stream.control)

The pg_stream.control file in the repository root is required by PostgreSQL's extension infrastructure. It declares the extension's description, default version, shared-library path, and privilege requirements. PostgreSQL reads this file when CREATE EXTENSION pg_stream; is executed.

During packaging (cargo pgrx package), pgrx replaces the @CARGO_VERSION@ placeholder with the version from Cargo.toml and copies the file into the target's share/extension/ directory alongside the SQL migration scripts.

DVM Operators

This document describes the Differential View Maintenance (DVM) operators implemented by pgstream. Each operator transforms a stream of row-level changes (deltas) propagated from source tables through the operator tree.

Prior Art

• Budiu, M. et al. (2023). "DBSP: Automatic Incremental View Maintenance." VLDB 2023. (comparison)
• Gupta, A. & Mumick, I.S. (1999). Materialized Views: Techniques, Implementations, and Applications. MIT Press.
• Koch, C. et al. (2014). "DBToaster: Higher-order Delta Processing for Dynamic, Frequently Fresh Views." VLDB Journal.
• PostgreSQL 9.4+ — Materialized views with REFRESH MATERIALIZED VIEW CONCURRENTLY.

Overview

When a stream table is created, the defining SQL query is parsed into a tree of DVM operators. During an differential refresh, changes flow bottom-up through this tree:

         Aggregate
            │
         Project
            │
          Filter
            │
    ┌───────┴───────┐
   Join             │
  ┌─┴─┐            │
Scan(A) Scan(B)   Scan(C)

Each operator implements a differentiation rule: given the delta (Δ) to its input(s), it produces the corresponding delta to its output. This is conceptually similar to automatic differentiation in calculus.

The general contract:

• Input: a set of ('+', row) and ('-', row) tuples (inserts and deletes)
• Output: a set of ('+', row) and ('-', row) tuples

Updates are modeled as a delete of the old row followed by an insert of the new row.

Operators

Scan

Module: src/dvm/operators/scan.rs

The leaf operator. Reads CDC changes from a source table's change buffer.

Delta Rule:

$$\Delta(\text{Scan}(R)) = \Delta R$$

The scan operator is a direct passthrough — inserts in the source become inserts in the output, deletes become deletes.

SQL Generation:

SELECT op, row_data FROM pgstream_changes.changes_<oid>
WHERE xid >= <last_consumed_xid>

Notes:

• Each source table has a dedicated change buffer table created by the CDC module.
• Row data is stored as JSONB with column names as keys.
• The __pgs_row_id column (xxHash of primary key) is included for deduplication.

Filter

Module: src/dvm/operators/filter.rs

Applies a WHERE clause predicate to the delta stream.

Delta Rule:

$$\Delta(\sigma_p(R)) = \sigma_p(\Delta R)$$

Filtering is applied to the deltas in the same way as to the base data — only rows satisfying the predicate pass through.

SQL Generation:

SELECT * FROM (<input_delta>) AS d
WHERE <predicate>

Example:

If the defining query is:

SELECT * FROM orders WHERE status = 'shipped'

And a new row (id=5, status='pending') is inserted, it does not appear in the delta output. If (id=3, status='shipped') is inserted, it passes through.

Edge Cases:

• For updates that change the predicate column (e.g., status from 'pending' to 'shipped'), the CDC produces a delete of the old row and insert of the new row. The filter passes the insert (matches) and blocks the delete (doesn't match the old row against the predicate), correctly resulting in a net insert.

Project

Module: src/dvm/operators/project.rs

Applies column projection from the target list.

Delta Rule:

$$\Delta(\pi_L(R)) = \pi_L(\Delta R)$$

Projects the same columns from the delta that the query projects from the base data.

SQL Generation:

SELECT <target_columns> FROM (<input_delta>) AS d

Notes:

• Projection is applied after filtering for efficiency.
• Computed expressions in the target list (e.g., price * quantity AS total) are evaluated on the delta rows.

Join (Inner)

Module: src/dvm/operators/join.rs

Implements inner join between two inputs.

Delta Rule:

For $R \bowtie S$:

$$\Delta(R \bowtie S) = (\Delta R \bowtie S) \cup (R' \bowtie \Delta S)$$

Where $R' = R \cup \Delta R$ (the new state of R after applying deltas).

In practice, when only one side has changes (common case), the delta join simplifies to joining the changed rows against the current state of the other side.

SQL Generation:

-- Changes to left side joined with current right side
SELECT '+' AS op, l.*, r.*
FROM (<left_delta> WHERE op = '+') AS l
JOIN <right_table> AS r ON <join_condition>

UNION ALL

-- Current left side joined with changes to right side
SELECT '+' AS op, l.*, r.*
FROM <left_table> AS l
JOIN (<right_delta> WHERE op = '+') AS r ON <join_condition>

(And corresponding DELETE queries for op = '-'.)

Notes:

• The join uses the current state of the non-changed side, not the change buffer.
• For equi-joins, this is efficient — the join key narrows the scan.
• Non-equi joins (theta joins) may require broader scans.

Outer Join

Module: src/dvm/operators/outer_join.rs (LEFT JOIN), src/dvm/operators/full_join.rs (FULL JOIN)

Implements LEFT, RIGHT, and FULL OUTER JOIN.

RIGHT JOIN Handling:

RIGHT JOIN is automatically converted to a LEFT JOIN with swapped left/right operands during query parsing. This normalization happens transparently — the user can write RIGHT JOIN and the parser rewrites it to an equivalent LEFT JOIN before the operator tree is constructed.

Delta Rule:

Similar to inner join, but additionally handles NULL-padded rows:

$$\Delta(R \text{ LEFT JOIN } S) = (\Delta R \bowtie_L S) \cup (R' \bowtie_L \Delta S)$$

With special handling for:

• Rows in ΔR that have no match in S → emit ('+', row, NULLs)
• Rows in ΔS that create a first match for an R row → emit ('-', row, NULLs) and ('+', row, s_data)
• Rows in ΔS that remove the last match for an R row → emit ('-', row, s_data) and ('+', row, NULLs)

SQL Generation (LEFT JOIN):

Uses anti-join detection (via NOT EXISTS) to correctly handle the NULL padding transitions.

FULL OUTER JOIN Delta Rule:

FULL OUTER JOIN extends the LEFT JOIN delta with symmetric right-side handling. The delta is computed as an 8-part UNION ALL:

1. Parts 1–5: Same as LEFT JOIN delta (inserted/deleted rows from both sides, with NULL-padding transitions)
2. Parts 6–7: Symmetric anti-join transitions for the right side (rows in ΔL that remove/create the last/first match for an S row)
3. Part 8: Right-side insertions that have no match in the left side → emit ('+', NULLs, s_data)

Each part uses pre-computed delta flags (__has_ins_*, __has_del_*) to efficiently detect first-match/last-match transitions without redundant subqueries.

Nested Join Support:

Module: src/dvm/operators/join_common.rs

All join operators (inner, left, full) support nested children — i.e., a join whose left or right operand is itself another join. The join_common module provides shared helpers:

• build_snapshot_sql() — returns the table reference for simple (Scan) operands, or a parenthesized subquery with disambiguated columns for nested join operands
• rewrite_join_condition() — rewrites column references in ON conditions to use the correct alias prefixes for nested children (e.g., o.cust_id → dl.o__cust_id)

This enables queries with 3 or more joined tables, e.g.:

SELECT o.id, c.name, p.title
FROM orders o
JOIN customers c ON o.cust_id = c.id
JOIN products p ON o.prod_id = p.id

Limitations:

• FULL OUTER JOIN delta computation can be expensive due to dual-side NULL tracking (8 UNION ALL parts).
• Performance degrades with high-cardinality join keys.
• NATURAL JOIN is supported — common columns are resolved automatically and synthesized into an explicit equi-join condition.

Aggregate

Module: src/dvm/operators/aggregate.rs

Handles GROUP BY with aggregate functions (COUNT, SUM, AVG, MIN, MAX, BOOL_AND, BOOL_OR, STRING_AGG, ARRAY_AGG, JSON_AGG, JSONB_AGG, BIT_AND, BIT_OR, BIT_XOR, JSON_OBJECT_AGG, JSONB_OBJECT_AGG, STDDEV_POP, STDDEV_SAMP, VAR_POP, VAR_SAMP, MODE, PERCENTILE_CONT, PERCENTILE_DISC, JSON_ARRAYAGG, JSON_OBJECTAGG) and the FILTER (WHERE …) and WITHIN GROUP (ORDER BY …) clauses.

Delta Rule:

$$\Delta(\gamma_{G, \text{agg}}(R)) = \gamma_{G, \text{agg}}(R' \text{ WHERE } G \in \text{affected_keys}) - \gamma_{G, \text{agg}}(R \text{ WHERE } G \in \text{affected_keys})$$

Where:

• $G$ = grouping columns
• affected_keys = the set of group key values that appear in ΔR
• $R'$ = $R \cup \Delta R$ (the new state)

Strategy:

1. Identify affected groups — Collect all group key values that appear in the delta (either inserted or deleted rows).
2. Recompute old values — Query the storage table for current aggregate values of affected groups.
3. Recompute new values — Query the updated source for new aggregate values of affected groups.
4. Diff — For each affected group:
   • If old exists and new differs → emit ('-', old) and ('+', new)
   • If old exists and new is gone → emit ('-', old) (group eliminated)
   • If no old and new exists → emit ('+', new) (new group appeared)

Supported Aggregate Functions:

FILTER Clause:

All aggregate functions support the FILTER (WHERE …) clause:

SELECT COUNT(*) FILTER (WHERE status = 'active') AS active_count FROM orders GROUP BY region

The filter predicate is applied within the delta computation — only rows matching the filter contribute to the aggregate delta. Filtered aggregates are excluded from the P5 direct-bypass optimization.

SQL Generation:

The aggregate operator uses a 3-CTE pipeline:

1. Merge CTE — Joins affected group keys against old (storage) and new (source) aggregate values, producing __pgs_meta_action ('I' for new-only groups, 'D' for disappeared groups, 'U' for changed groups).
2. LATERAL VALUES expansion — A single-pass LATERAL (VALUES ...) clause expands each merge row into insert and delete actions, avoiding a 4-branch UNION ALL:

FROM merge_cte m,
LATERAL (VALUES
    ('I', m.new_count, m.new_total),
    ('D', m.old_count, m.old_total)
) v(action, count_val, val_total)
WHERE (m.__pgs_meta_action = 'I' AND v.action = 'I')
   OR (m.__pgs_meta_action = 'D' AND v.action = 'D')
   OR (m.__pgs_meta_action = 'U')

3. Final projection — Emits ('+', row) and ('-', row) tuples for the refresh engine.

MIN/MAX Merge Strategy:

MIN and MAX use a semi-algebraic strategy with two cases:

1. Non-extremum deletion — When the deleted row is NOT the current minimum (or maximum), the merge uses LEAST(old_value, new_inserts) for MIN or GREATEST(old_value, new_inserts) for MAX. This is fully algebraic and requires no rescan.

2. Extremum deletion — When the row holding the current minimum (or maximum) IS deleted, the new value cannot be computed from the delta alone. The merge expression returns NULL as a sentinel, which triggers the change-detection guard (IS DISTINCT FROM) to emit the group for re-aggregation. The MERGE layer treats this as a DELETE + INSERT pair, recomputing the group from source data. This is still more efficient than a full table refresh since only affected groups are rescanned.

Distinct

Module: src/dvm/operators/distinct.rs

Implements SELECT DISTINCT using reference counting.

Delta Rule:

$$\Delta(\delta(R)) = { r \in \Delta R : \text{count}(r, R) = 0 \land \text{count}(r, R') > 0 } - { r \in \Delta R : \text{count}(r, R) > 0 \land \text{count}(r, R') = 0 }$$

In other words:

• A row enters the output when its count transitions from 0 to ≥1
• A row leaves the output when its count transitions from ≥1 to 0

Strategy:

Maintains a hidden __pgs_dup_count column in the storage table to track how many times each distinct row appears in the pre-distinct input.

1. On insert: increment count. If count was 0, emit ('+', row).
2. On delete: decrement count. If count becomes 0, emit ('-', row).

Notes:

• The duplicate count is not visible in user queries against the storage table (projected away by the view layer).
• Duplicate counting uses __pgs_row_id (xxHash) for efficient lookups.

Union All

Module: src/dvm/operators/union_all.rs

Merges deltas from two branches.

Delta Rule:

$$\Delta(R \cup_{\text{all}} S) = \Delta R \cup_{\text{all}} \Delta S$$

Simply concatenates the delta streams from both branches.

SQL Generation:

SELECT * FROM (<left_delta>)
UNION ALL
SELECT * FROM (<right_delta>)

Notes:

• Column count and types must match between branches.
• Each branch is independently processed through its own operator sub-tree.
• This is the simplest operator since UNION ALL preserves all duplicates.

Intersect

Module: src/dvm/operators/intersect.rs

Implements INTERSECT and INTERSECT ALL using dual-count per-branch multiplicity tracking.

Delta Rule:

$$\Delta(R \cap S): \text{emit rows where } \min(\text{count}_L, \text{count}_R) \text{ crosses the 0 boundary}$$

• INTERSECT (set): a row is present when both branches contain it.
• INTERSECT ALL (bag): a row appears $\min(\text{count}_L, \text{count}_R)$ times.

SQL Generation (3-CTE chain):

1. Delta CTE — tags rows from left/right child deltas with branch indicator ('L'/'R') and computes per-row net_count.
2. Merge CTE — joins with the storage table to compute old and new per-branch counts (__pgs_count_l, __pgs_count_r).
3. Final CTE — detects boundary crossings using LEAST(old_count_l, old_count_r) vs LEAST(new_count_l, new_count_r).

Notes:

• Storage table requires hidden columns __pgs_count_l and __pgs_count_r for multiplicity tracking.
• Both set and bag variants use the same 3-CTE structure; only the boundary logic stays the same (both use LEAST).

Except

Module: src/dvm/operators/except.rs

Implements EXCEPT and EXCEPT ALL using dual-count per-branch multiplicity tracking.

Delta Rule:

$$\Delta(R - S): \text{emit rows where } \max(0, \text{count}_L - \text{count}_R) \text{ crosses the 0 boundary}$$

• EXCEPT (set): a row is present when it exists in the left but not the right branch.
• EXCEPT ALL (bag): a row appears $\max(0, \text{count}_L - \text{count}_R)$ times.

SQL Generation (3-CTE chain):

1. Delta CTE — same as Intersect: tags rows from both child deltas with branch indicator.
2. Merge CTE — joins with storage table for old/new per-branch counts.
3. Final CTE — detects boundary crossings using GREATEST(0, old_count_l - old_count_r) vs GREATEST(0, new_count_l - new_count_r).

Notes:

• EXCEPT is not commutative — left branch is the positive input, right is subtracted.
• Storage table requires hidden columns __pgs_count_l and __pgs_count_r.
• Same 3-CTE structure as Intersect with different effective-count function.

Subquery

Module: src/dvm/operators/subquery.rs

Handles both inlined CTEs and explicit subqueries in FROM ((SELECT ...) AS alias).

Delta Rule:

$$\Delta(\rho_{\text{alias}}(Q)) = \rho_{\text{alias}}(\Delta Q)$$

A subquery wrapper is transparent for differentiation — it delegates to its child's delta and optionally renames output columns to match the subquery's column aliases.

SQL Generation:

-- If column aliases differ from child output columns:
SELECT __pgs_row_id, __pgs_action, child_col1 AS alias_col1, child_col2 AS alias_col2
FROM (<child_delta>)

If the child columns already match the aliases, the subquery is a pure passthrough — no additional CTE is emitted.

Notes:

• This operator enables both CTE support (Tier 1) and standalone subqueries in FROM.
• Column aliases on subqueries (FROM (...) AS x(a, b)) are handled by emitting a thin renaming CTE.
• The subquery body is fully differentiated as a normal operator sub-tree.

CTE Scan (Shared Delta)

Module: src/dvm/operators/cte_scan.rs

Handles multi-reference CTEs by computing the CTE body's delta once and reusing it across all references (Tier 2).

Delta Rule:

$$\Delta(\text{CteScan}(\text{id}, Q)) = \text{cache}[\text{id}] \quad \text{(computed once, reused)}$$

When a CTE is referenced multiple times in a query, each reference produces a CteScan node with the same cte_id. The diff engine differentiates the CTE body once and caches the result. Subsequent CteScan nodes for the same CTE reuse the cached delta.

SQL Generation:

-- First reference: differentiates the CTE body and stores result in cache
-- Subsequent references: point to the same system CTE name
SELECT __pgs_row_id, __pgs_action, <columns>
FROM __pgs_cte_<cte_name>_delta  -- shared across all references

If column aliases are present, a thin renaming CTE is added on top of the cached delta.

Notes:

• Without CteScan (Tier 1), multi-reference CTEs are inlined: each reference duplicates the full operator sub-tree. CteScan (Tier 2) eliminates this duplication.
• The CTE body is pre-differentiated in dependency order (earlier CTEs before later ones that reference them).
• Column alias support follows the same pattern as the Subquery operator.

Recursive CTEs

Recursive CTEs (WITH RECURSIVE) are supported via two strategies depending on the refresh mode:

FULL Mode

Recursive CTEs work out-of-the-box with refresh_mode = 'FULL'. The defining query is executed as-is via INSERT INTO ... SELECT ..., and PostgreSQL handles the iterative evaluation internally.

DIFFERENTIAL Mode (Three-Strategy Incremental Maintenance)

Recursive CTEs with refresh_mode = 'DIFFERENTIAL' use an automatic three-strategy approach, selected based on column compatibility and change type:

Strategy 1: Semi-Naive Evaluation (INSERT-only changes)

When only INSERT changes are present in the change buffer, pg_stream uses semi-naive evaluation — the standard technique for incremental fixpoint computation. The base case is differentiated normally through the DVM operator tree, then the resulting delta is propagated through the recursive term using a nested WITH RECURSIVE:

WITH RECURSIVE
  __pgs_base_delta AS (
    -- Normal DVM differentiation of the base case (INSERT rows only)
    <differentiated base case>
  ),
  __pgs_rec_delta AS (
    -- Seed: base case delta rows
    SELECT cols FROM __pgs_base_delta WHERE __pgs_action = 'I'
    UNION ALL
    -- Seed: new base rows joining existing ST storage
    SELECT cols FROM <recursive term with self_ref = ST_storage, base = change_buffer>
    UNION ALL
    -- Propagation: recursive term applied to growing delta
    SELECT cols FROM <recursive term with self_ref = __pgs_rec_delta, base = full>
  )
SELECT pgstream.pg_stream_hash(...) AS __pgs_row_id, 'I' AS __pgs_action, cols
FROM __pgs_rec_delta

The cost is proportional to the number of new rows produced by the change, not the full result set.

Strategy 2: Delete-and-Rederive / DRed (mixed INSERT/DELETE/UPDATE changes)

When the change buffer contains DELETE or UPDATE changes, simple propagation is insufficient — a deleted base row may have transitively derived many recursive rows, some of which may still be derivable from alternative paths. DRed handles this in four phases:

1. Insert propagation — semi-naive evaluation for the INSERT portion (same as Strategy 1)
2. Over-deletion cascade — propagate base-case deletions through the recursive term against ST storage to find all transitively-derived rows that might be invalidated
3. Rederivation — re-execute the recursive CTE from the remaining (non-deleted) base rows to restore any over-deleted rows that have alternative derivations
4. Combine — final delta = inserts + (over-deletions − rederived rows)

This avoids full recomputation while correctly handling deletions with alternative derivation paths.

Strategy 3: Recomputation Fallback

When the CTE defines more columns than the outer SELECT projects (column mismatch), the incremental strategies cannot be used because the ST storage table lacks columns needed for recursive self-joins. In this case, the full defining query is re-executed and anti-joined against current storage:

WITH __pgs_recomp_new AS (
    SELECT pgstream.pg_stream_hash(row_to_json(sub)::text) AS __pgs_row_id, col1, col2, ...
    FROM (<defining_query>) sub
),
__pgs_recomp_ins AS (
    SELECT n.__pgs_row_id, 'I'::text AS __pgs_action, n.col1, n.col2, ...
    FROM __pgs_recomp_new n
    LEFT JOIN <storage_table> s ON s.__pgs_row_id = n.__pgs_row_id
    WHERE s.__pgs_row_id IS NULL
),
__pgs_recomp_del AS (
    SELECT s.__pgs_row_id, 'D'::text AS __pgs_action, s.col1, s.col2, ...
    FROM <storage_table> s
    LEFT JOIN __pgs_recomp_new n ON n.__pgs_row_id = s.__pgs_row_id
    WHERE n.__pgs_row_id IS NULL
)
SELECT * FROM __pgs_recomp_ins
UNION ALL
SELECT * FROM __pgs_recomp_del

The cost is proportional to the full result set size.

Strategy Selection

Notes:

• Non-linear recursion (multiple self-references in the recursive term) is rejected — PostgreSQL restricts the recursive term to reference the CTE at most once.
• The __pgs_row_id column (xxHash of the JSON-serialized row) is used for row identity.
• For write-heavy workloads on very large recursive result sets with frequent mixed changes, refresh_mode = 'FULL' may still be more efficient than DRed.

Window Functions

Module: src/dvm/operators/window.rs

Handles window functions (ROW_NUMBER, RANK, DENSE_RANK, SUM() OVER, etc.) using partition-based recomputation.

Delta Rule:

When any row in a partition changes (insert, update, or delete), the entire partition's window function output is recomputed:

$$\Delta(\omega_{f, P}(R)) = \omega_{f, P}(R'|{\text{affected partitions}}) - \omega{f, P}(R|_{\text{affected partitions}})$$

Where $P$ is the PARTITION BY key and $f$ is the window function.

Strategy:

1. Identify affected partition keys from the child delta.
2. Delete old window function results for affected partitions from storage.
3. Fetch surviving (unchanged) rows in affected partitions from the child.
4. Recompute the window function on the current input for affected partitions.
5. Emit the recomputed rows as inserts.

SQL Generation:

-- CTE 1: Affected partition keys from delta
WITH affected_partitions AS (
    SELECT DISTINCT <partition_cols> FROM (<child_delta>)
),
-- CTE 2: Surviving rows (not in delta) for affected partitions  
surviving AS (
    SELECT * FROM <storage_table>
    WHERE (<partition_cols>) IN (SELECT * FROM affected_partitions)
    AND __pgs_row_id NOT IN (SELECT __pgs_row_id FROM (<child_delta>) WHERE __pgs_action = 'D')
),
-- CTE 3: Recompute window function
recomputed AS (
    SELECT *, <window_func> OVER (PARTITION BY <partition_cols> ORDER BY <order_cols>) AS <alias>
    FROM surviving
)
-- Delete old results + insert recomputed results
SELECT 'D' AS __pgs_action, ...  -- old rows from affected partitions
UNION ALL
SELECT 'I' AS __pgs_action, ...  -- recomputed rows

Notes:

• The cost is proportional to the size of affected partitions, not the full table. For workloads where changes spread across few partitions, this is efficient.
• All window expressions must share the same PARTITION BY / ORDER BY clause.
• Without PARTITION BY, the entire table is treated as a single partition — any change triggers a full recomputation.

Window Frame Clauses:

Window frame specifications are fully supported:

• Modes: ROWS, RANGE, GROUPS
• Bounds: UNBOUNDED PRECEDING, N PRECEDING, CURRENT ROW, N FOLLOWING, UNBOUNDED FOLLOWING
• Between syntax: BETWEEN <start> AND <end>
• Exclusion: EXCLUDE CURRENT ROW, EXCLUDE GROUP, EXCLUDE TIES, EXCLUDE NO OTHERS

Example: SUM(val) OVER (ORDER BY ts ROWS BETWEEN 3 PRECEDING AND CURRENT ROW)

Named WINDOW Clauses:

Named window definitions are resolved from the query-level WINDOW clause:

SELECT id, SUM(val) OVER w, AVG(val) OVER w
FROM data
WINDOW w AS (PARTITION BY category ORDER BY ts)

The parser resolves OVER w by looking up the window definition from the WINDOW clause and merging partition, order, and frame specifications.

Lateral Function (Set-Returning Functions in FROM)

Module: src/dvm/operators/lateral_function.rs

Handles set-returning functions (SRFs) used in the FROM clause with implicit LATERAL semantics: jsonb_array_elements, jsonb_each, jsonb_each_text, unnest, etc.

Delta Rule:

When a source row changes (insert, update, or delete), the SRF expansion is re-evaluated only for that source row:

$$\Delta(R \ltimes f(R.\text{col})) = (R' \ltimes f(R'.\text{col}))|{\text{changed rows}} - (R \ltimes f(R.\text{col}))|{\text{changed rows}}$$

Where $R$ is the source table, $f$ is the SRF, and changed rows are identified via the child delta.

Strategy (Row-Scoped Recomputation):

1. Propagate the child delta to identify changed source rows.
2. Find all existing ST rows derived from changed source rows (via column matching).
3. Delete old SRF expansions for those source rows.
4. Re-expand the SRF for inserted/updated source rows.
5. Emit deletes + inserts as the final delta.

SQL Generation (4-CTE chain):

-- CTE 1: Changed source rows from child delta
WITH lat_changed AS (
    SELECT DISTINCT "__pgs_row_id", "__pgs_action", <child_cols>
    FROM <child_delta>
),
-- CTE 2: Old ST rows for changed source rows (to be deleted)
lat_old AS (
    SELECT st."__pgs_row_id", st.<all_output_cols>
    FROM <st_table> st
    WHERE EXISTS (
        SELECT 1 FROM lat_changed cs
        WHERE st.<col1> IS NOT DISTINCT FROM cs.<col1>
          AND st.<col2> IS NOT DISTINCT FROM cs.<col2>
          ...
    )
),
-- CTE 3: Re-expand SRF for inserted/updated source rows
lat_expand AS (
    SELECT pg_stream_hash(<all_cols>::text) AS "__pgs_row_id",
           cs.<child_cols>, <srf_alias>.<srf_cols>
    FROM lat_changed cs,
         LATERAL <srf_function>(cs.<arg>) AS <srf_alias>
    WHERE cs."__pgs_action" = 'I'
),
-- CTE 4: Final delta
lat_final AS (
    SELECT "__pgs_row_id", 'D' AS "__pgs_action", <cols> FROM lat_old
    UNION ALL
    SELECT "__pgs_row_id", 'I' AS "__pgs_action", <cols> FROM lat_expand
)

Row Identity:

Content-based: hash(child_columns || srf_result_columns). This is stable as long as the same source row produces the same expanded values.

Supported SRFs:

Notes:

• The cost is proportional to the number of changed source rows × average SRF expansion size, not the full table.
• WITH ORDINALITY is supported — adds a bigint ordinality column to the output.
• ROWS FROM() with multiple functions is not supported (rejected at parse time).
• Column aliases (e.g., AS child(value)) are used to determine output column names; for known SRFs without aliases, the alias name becomes the column name.
• JSON_TABLE (PostgreSQL 17+) — JSON_TABLE(expr, path COLUMNS (...)) is modeled as a LateralFunction and uses the same row-scoped recomputation strategy. Supported column types: regular, EXISTS, formatted, and nested columns with ON ERROR/ON EMPTY behaviors and PASSING clauses.

Lateral Subquery (Correlated Subqueries in FROM)

Module: src/dvm/operators/lateral_subquery.rs

Handles correlated subqueries used in the FROM clause with explicit or implicit LATERAL semantics: FROM t, LATERAL (SELECT ... WHERE ref = t.col) AS alias or FROM t LEFT JOIN LATERAL (...) AS alias ON true.

Delta Rule:

When an outer row changes, the correlated subquery is re-executed only for that row:

$$\Delta(R \ltimes Q(R)) = (R' \ltimes Q(R'))|{\text{changed rows}} - (R \ltimes Q(R))|{\text{changed rows}}$$

Where $R$ is the outer table, $Q(R)$ is the correlated subquery, and changed rows are identified via the child delta.

Strategy (Row-Scoped Recomputation):

1. Propagate the child delta to identify changed outer rows.
2. Find all existing ST rows derived from changed outer rows (via column matching with IS NOT DISTINCT FROM).
3. Delete old subquery expansions for those outer rows.
4. Re-execute the subquery for inserted/updated outer rows using the original outer alias.
5. Emit deletes + inserts as the final delta.

SQL Generation (4-CTE chain):

-- CTE 1: Changed outer rows from child delta
WITH lat_sq_changed AS (
    SELECT DISTINCT "__pgs_row_id", "__pgs_action", <child_cols>
    FROM <child_delta>
),
-- CTE 2: Old ST rows for changed outer rows (to be deleted)
lat_sq_old AS (
    SELECT st."__pgs_row_id", st.<all_output_cols>
    FROM <st_table> st
    WHERE EXISTS (
        SELECT 1 FROM lat_sq_changed cs
        WHERE st.<col1> IS NOT DISTINCT FROM cs.<col1>
          AND st.<col2> IS NOT DISTINCT FROM cs.<col2>
          ...
    )
),
-- CTE 3: Re-execute subquery for inserted/updated outer rows
lat_sq_expand AS (
    SELECT pg_stream_hash(<all_cols>::text) AS "__pgs_row_id",
           <outer_alias>.<child_cols>, <sub_alias>.<sub_cols>
    FROM lat_sq_changed AS <outer_alias>,      -- Original outer alias!
         LATERAL (<subquery_sql>) AS <sub_alias>
    WHERE <outer_alias>."__pgs_action" = 'I'
),
-- CTE 4: Final delta
lat_sq_final AS (
    SELECT "__pgs_row_id", 'D' AS "__pgs_action", <cols> FROM lat_sq_old
    UNION ALL
    SELECT "__pgs_row_id", 'I' AS "__pgs_action", <cols> FROM lat_sq_expand
)

LEFT JOIN LATERAL Handling:

For queries using LEFT JOIN LATERAL (...) ON true, the expand CTE uses LEFT JOIN LATERAL instead of comma syntax and wraps subquery columns in COALESCE for hash stability:

lat_sq_expand AS (
    SELECT pg_stream_hash(<outer_cols>::text || '/' || COALESCE(<sub_cols>::text, '')) AS "__pgs_row_id",
           <outer_alias>.<child_cols>, <sub_alias>.<sub_cols>
    FROM lat_sq_changed AS <outer_alias>
    LEFT JOIN LATERAL (<subquery_sql>) AS <sub_alias> ON true
    WHERE <outer_alias>."__pgs_action" = 'I'
)

Row Identity:

Content-based: hash(outer_columns || '/' || subquery_result_columns). For LEFT JOIN with NULL results, COALESCE ensures a stable hash.

Supported Patterns:

Key Design Decision: Outer Alias Rewriting

The subquery body contains column references to the outer table (e.g., WHERE li.order_id = o.id). In the expansion CTE, the changed-sources CTE is aliased with the original outer table alias (e.g., lat_sq_changed AS o) so that the subquery's column references resolve naturally without rewriting.

Notes:

• The cost is proportional to the number of changed outer rows × average subquery result size, not the full table.
• The subquery is stored as raw SQL (like LateralFunction) because it cannot be independently differentiated — it depends on outer row context.
• Source table OIDs referenced by the subquery body are extracted at parse time for CDC trigger setup.
• ORDER BY + LIMIT inside the subquery are valid (they apply per-outer-row, not to the stream table).

Semi-Join (EXISTS / IN Subquery)

Module: src/dvm/operators/semi_join.rs

Handles WHERE EXISTS (SELECT ... FROM ...) and WHERE col IN (SELECT ...) patterns. The parser transforms these into a SemiJoin operator with a left (outer) child, a right (inner) child, and a join condition.

Delta Rule:

$$\Delta(L \ltimes R) = \Delta L|{R} + L|{\Delta R \text{ causes existence change}}$$

• Part 1: Outer rows that changed and still satisfy the semi-join condition.
• Part 2: Existing outer rows whose semi-join result flipped due to inner changes (a matching inner row was inserted or deleted).

Strategy (Two-Part Delta):

1. Part 1 (outer delta): Filter delta_left to rows that have at least one match in the current right-hand snapshot.
2. Part 2 (inner delta): For each row in the left snapshot, check whether the existence of matching right-hand rows changed between the old and current state. Emit 'I' if a match appeared, 'D' if all matches disappeared.

The "old" right-hand state is reconstructed from the current state by reversing the delta: R_old = (R_current EXCEPT ALL delta_right(action='I')) UNION ALL delta_right(action='D').

Row Identity:

• Part 1: Uses __pgs_row_id from the left delta.
• Part 2: Content-based hash via pg_stream_hash_multi on left-side columns.

Supported Patterns:

Anti-Join (NOT EXISTS / NOT IN Subquery)

Module: src/dvm/operators/anti_join.rs

Handles WHERE NOT EXISTS (SELECT ... FROM ...) and WHERE col NOT IN (SELECT ...) patterns. The inverse of the semi-join operator.

Delta Rule:

$$\Delta(L \triangleright R) = \Delta L|{\neg R} + L|{\Delta R \text{ causes existence change}}$$

• Part 1: Outer rows that changed and have no match in the right-hand snapshot.
• Part 2: Existing outer rows whose anti-join result flipped due to inner changes.

Strategy (Two-Part Delta):

1. Part 1 (outer delta): Filter delta_left to rows with NOT EXISTS in the current right snapshot.
2. Part 2 (inner delta): For each row in the left snapshot, detect existence changes. Emit 'D' if a match appeared (row no longer qualifies), 'I' if all matches disappeared (row now qualifies).

Note the inverted semantics compared to semi-join: a new match means deletion, losing all matches means insertion.

Row Identity: Same as semi-join.

Supported Patterns:

Scalar Subquery (Correlated SELECT Subquery)

Module: src/dvm/operators/scalar_subquery.rs

Handles scalar subqueries appearing in the SELECT list, e.g., SELECT a, (SELECT max(x) FROM t) AS mx FROM s. The subquery must return exactly one row and one column.

Delta Rule:

$$\Delta(L \times q) = \Delta L \times q' + L \times (q' - q)$$

Where $q$ is the scalar subquery value and $q'$ is the updated value.

Strategy (Two-Part Delta):

1. Part 1 (outer delta): Propagate the child delta, appending the current scalar subquery value to each row.
2. Part 2 (scalar value change): When the scalar subquery's result changes, emit deletes for all existing outer rows (with the old scalar value) and re-inserts for all outer rows (with the new value). The old scalar value is reconstructed by reversing the inner delta.

SQL Generation (3 or 4 CTEs):

-- Part 1: child delta + current scalar value
WITH sq_outer AS (
    SELECT *, (<scalar_subquery>) AS "<alias>"
    FROM <child_delta>
),
-- Part 2a: DELETE all outer rows when scalar changed
sq_del AS (
    SELECT "__pgs_row_id", 'D' AS "__pgs_action", <cols>
    FROM <st_table>
    WHERE (<scalar_old>) IS DISTINCT FROM (<scalar_current>)
),
-- Part 2b: INSERT all outer rows with new scalar value
sq_ins AS (
    SELECT pg_stream_hash_multi(...) AS "__pgs_row_id",
           'I' AS "__pgs_action", <cols>, (<scalar_current>) AS "<alias>"
    FROM <source_snapshot>
    WHERE (<scalar_old>) IS DISTINCT FROM (<scalar_current>)
)
-- Final: UNION ALL of all parts
SELECT * FROM sq_outer
UNION ALL SELECT * FROM sq_del
UNION ALL SELECT * FROM sq_ins

Row Identity:

• Part 1: __pgs_row_id from the child delta.
• Part 2: Content-based hash via pg_stream_hash_multi on all output columns.

Notes:

• The scalar subquery is stored as raw SQL (deparsed from the parse tree).
• The old scalar value is approximated using the same EXCEPT ALL / UNION ALL reversal technique as semi/anti-join.
• If the scalar subquery references a table that changes, all outer rows must be re-evaluated — the delta can be large.
• Source OIDs used by the scalar subquery are captured at parse time for CDC trigger registration.

Operator Tree Construction

The DVM engine builds the operator tree by analyzing the parsed query:

1. WITH clause → CTE definitions extracted into a name→body map (non-recursive) or CTE registry (multi-reference)
2. FROM clause → Scan nodes for physical tables; Subquery nodes for inlined CTEs and subqueries in FROM; CteScan nodes for multi-reference CTEs; LateralFunction nodes for SRFs and JSON_TABLE in FROM; LateralSubquery nodes for correlated subqueries in FROM
3. JOIN → Join or OuterJoin wrapping two sub-trees
4. LATERAL SRFs → LateralFunction wrapping the left-hand FROM item as its child
5. LATERAL subqueries → LateralSubquery wrapping the left-hand FROM item as its child (comma syntax or JOIN LATERAL)
6. WHERE subqueries → SemiJoin for EXISTS/IN (subquery), AntiJoin for NOT EXISTS/NOT IN (subquery), extracted from the WHERE clause
7. Scalar subqueries → ScalarSubquery for (SELECT ...) in the SELECT list, wrapping the child tree
8. WHERE → Filter wrapping the scan/join tree (remaining non-subquery predicates)
9. SELECT list → Project for column selection and expressions
10. GROUP BY → Aggregate wrapping the filtered/projected tree
11. DISTINCT → Distinct on top
12. UNION ALL → UnionAll combining two complete sub-trees
13. INTERSECT / EXCEPT → Intersect or Except combining two sub-trees with dual-count tracking
14. Window functions → Window wrapping the sub-tree with PARTITION BY / ORDER BY metadata
15. ORDER BY → silently discarded (storage row order is undefined)
16. LIMIT / OFFSET → rejected with a clear error (stream tables materialize the full result set)

For recursive CTEs (WITH RECURSIVE), the query is parsed into an OpTree with RecursiveCte operator nodes. In DIFFERENTIAL mode, the strategy (semi-naive, DRed, or recomputation) is selected automatically based on column compatibility and change type — see the Recursive CTEs section above for details.

The tree is then traversed bottom-up during delta generation: each operator's generate_delta_sql() method composes its SQL fragment around the output of its child operator(s).

Further Reading

• ARCHITECTURE.md — System-wide component overview
• SQL_REFERENCE.md — Complete function reference
• CONFIGURATION.md — GUC tuning guide

pg_stream — Benchmark Guide

This document explains how the database-level refresh benchmarks work and how to interpret their output.

Overview

The benchmark suite in tests/e2e_bench_tests.rs measures wall-clock refresh time for FULL vs INCREMENTAL mode across a matrix of table sizes, change rates, and query complexities. Each benchmark spawns an isolated PostgreSQL 18.x container via Testcontainers, ensuring reproducible and interference-free measurements.

The core question the benchmarks answer:

How much faster is an INCREMENTAL refresh compared to a FULL refresh, given a specific workload?

Prerequisites

Build the E2E test Docker image before running any benchmarks:

./tests/build_e2e_image.sh

Docker must be running on the host.

Running Benchmarks

All benchmark tests are tagged #[ignore] so they are skipped during normal CI. The --nocapture flag is required to see the printed output tables.

Quick Spot Checks (~5–10 seconds each)

# Simple scan, 10K rows, 1% change rate
cargo test --test e2e_bench_tests --features pg18 -- --ignored --nocapture bench_scan_10k_1pct

# Aggregate query, 100K rows, 1% change rate
cargo test --test e2e_bench_tests --features pg18 -- --ignored --nocapture bench_aggregate_100k_1pct

# Join + aggregate, 100K rows, 10% change rate
cargo test --test e2e_bench_tests --features pg18 -- --ignored --nocapture bench_join_agg_100k_10pct

Zero-Change Latency (~5 seconds)

cargo test --test e2e_bench_tests --features pg18 -- --ignored --nocapture bench_no_data_refresh_latency

Full Matrix (~15–30 minutes)

Runs all 30 combinations and prints a consolidated summary:

cargo test --test e2e_bench_tests --features pg18 -- --ignored --nocapture bench_full_matrix

Run All Benchmarks in Parallel

cargo test --test e2e_bench_tests --features pg18 -- --ignored --nocapture

Note: each test starts its own container, so parallel execution requires sufficient Docker resources.

Benchmark Dimensions

Table Sizes

Change Rates

Query Complexities

DML Mix per Cycle

Each change cycle applies a realistic mix of operations:

What Each Benchmark Does

1. Start a fresh PostgreSQL 18.x container
2. Install the pg_stream extension
3. Create and populate the source table (10K or 100K rows)
4. Create dimension table if needed (for join scenarios)
5. ANALYZE for stable query plans

── FULL mode ──
6. Create a Stream Table in FULL refresh mode
7. For each of 3 cycles:
   a. Apply random DML (updates + deletes + inserts)
   b. ANALYZE
   c. Time the FULL refresh (TRUNCATE + re-execute entire query)
   d. Record refresh_ms and ST row count
8. Drop the FULL-mode ST

── INCREMENTAL mode ──
9. Reset source table to same starting state
10. Create a Stream Table in INCREMENTAL refresh mode
11. For each of 3 cycles:
    a. Apply random DML (same parameters)
    b. ANALYZE
    c. Time the INCREMENTAL refresh (delta query + MERGE)
    d. Record refresh_ms and ST row count

12. Print results table and summary

Both modes start from the same data to ensure a fair comparison. The 3-cycle design captures warm-up effects (cycle 1 may be slower due to plan caching).

Reading the Output

Detail Table

╔══════════════════════════════════════════════════════════════════════════════════════╗
║                    pg_stream Refresh Benchmark Results                      ║
╠════════════╤══════════╤════════╤═════════════╤═══════╤════════════╤═════════════════╣
║ Scenario   │ Rows     │ Chg %  │ Mode        │ Cycle │ Refresh ms │ ST Rows         ║
╠════════════╪══════════╪════════╪═════════════╪═══════╪════════════╪═════════════════╣
║ aggregate  │    10000 │     1% │ FULL        │     1 │       22.1 │               5 ║
║ aggregate  │    10000 │     1% │ FULL        │     2 │        4.8 │               5 ║
║ aggregate  │    10000 │     1% │ FULL        │     3 │        5.3 │               5 ║
║ aggregate  │    10000 │     1% │ INCREMENTAL │     1 │        8.4 │               5 ║
║ aggregate  │    10000 │     1% │ INCREMENTAL │     2 │        4.4 │               5 ║
║ aggregate  │    10000 │     1% │ INCREMENTAL │     3 │        4.6 │               5 ║
╚════════════╧══════════╧════════╧═════════════╧═══════╧════════════╧═════════════════╝

Summary Table

┌─────────────────────────────────────────────────────────────────────────┐
│                        Summary (avg ms per cycle)                       │
├────────────┬──────────┬────────┬─────────────────┬──────────────────────┤
│ Scenario   │ Rows     │ Chg %  │ FULL avg ms     │ INCREMENTAL avg ms   │
├────────────┼──────────┼────────┼─────────────────┼──────────────────────┤
│ aggregate  │    10000 │     1% │       10.7       │        5.8 (  1.8x) │
└────────────┴──────────┴────────┴─────────────────┴──────────────────────┘

The Speedup value in parentheses is FULL avg / INCREMENTAL avg — how many times faster the incremental refresh is compared to a full refresh.

Interpreting the Speedup

What to Expect

Rules of Thumb

Key Patterns to Look For

1. Scaling with table size: For the same change rate, speedup should increase with table size. FULL must re-process all rows; INCREMENTAL processes only the delta.

2. Degradation with change rate: As change rate rises from 1% → 50%, speedup should decrease. At 50%, INCREMENTAL processes half the table which approaches FULL cost.

3. Query complexity amplifies speedup: Aggregate and join queries benefit more from INCREMENTAL because they avoid expensive re-computation. A join_agg at 1% changes should show higher speedup than a simple scan at the same parameters.

4. Cycle 1 warm-up: The first cycle in each mode may be slower due to PostgreSQL plan cache population. Use cycles 2–3 for the steadiest numbers.

5. ST Rows consistency: The ST row count should be similar between FULL and INCREMENTAL for the same scenario (accounting for random DML). Large discrepancies indicate a correctness issue.

Zero-Change Latency

The bench_no_data_refresh_latency test measures the overhead of a refresh when no data has changed — the NO_DATA code path.

┌──────────────────────────────────────────────┐
│ NO_DATA Refresh Latency (10 iterations)      │
├──────────────────────────────────────────────┤
│ Avg:     3.21 ms                             │
│ Max:     5.10 ms                             │
│ Target: < 10 ms                              │
│ Status: ✅ PASS                              │
└──────────────────────────────────────────────┘

A passing result confirms the scheduler's per-cycle overhead is negligible. Values > 10 ms in containerized environments may be acceptable due to Docker overhead; bare-metal PostgreSQL should comfortably meet the target.

Available Tests

Individual Tests (10K rows)

Individual Tests (100K rows)

Special Tests

In-Process Micro-Benchmarks (Criterion.rs)

In addition to the E2E database benchmarks, the project includes two Criterion.rs benchmark suites that measure pure Rust computation time without database overhead. These are useful for tracking performance regressions in the internal query-building and IVM differentiation logic.

Benchmark Suites

refresh_bench — Utility Functions

benches/refresh_bench.rs benchmarks the low-level helper functions used during refresh operations:

diff_operators — IVM Operator Differentiation

benches/diff_operators.rs benchmarks the delta SQL generation for every IVM operator. Each benchmark creates a realistic operator tree and measures differentiate() throughput:

Running Micro-Benchmarks

# Run all Criterion benchmarks
just bench

# Run only refresh utility benchmarks
cargo bench --bench refresh_bench --features pg18

# Run only IVM diff operator benchmarks
just bench-diff
# or equivalently:
cargo bench --bench diff_operators --features pg18

# Output in Bencher-compatible format (for CI integration)
just bench-bencher

Output and Reports

Criterion produces statistical analysis for each benchmark including:

• Mean and standard deviation of execution time
• Throughput (iterations/sec)
• Comparison with previous run — reports improvements/regressions with confidence intervals

HTML reports are generated in target/criterion/ with interactive charts showing distributions and regression history. Open target/criterion/report/index.html to browse all results.

Sample output:

diff_scan/3_columns   time:   [11.834 µs 12.074 µs 12.329 µs]
diff_scan/10_columns  time:   [16.203 µs 16.525 µs 16.869 µs]
diff_aggregate/simple time:   [21.447 µs 21.862 µs 22.301 µs]
diff_inner_join       time:   [25.919 µs 26.421 µs 26.952 µs]

Continuous Benchmarking with Bencher

Bencher provides continuous benchmark tracking in CI, detecting performance regressions on pull requests before they merge.

How It Works

The .github/workflows/benchmarks.yml workflow:

1. On main pushes — runs both Criterion suites and uploads results to Bencher as the baseline. This establishes the expected performance for each benchmark.

2. On pull requests — runs the same benchmarks and compares against the main baseline using a Student's t-test with a 99% upper confidence boundary. If any benchmark regresses beyond the threshold, the PR check fails.

Setup

To enable Bencher for your fork or deployment:

1. Create a Bencher account at bencher.dev and create a project.

2. Add the API token as a GitHub Actions secret:

   • Go to Settings → Secrets and variables → Actions
   • Add BENCHER_API_TOKEN with your Bencher API token

3. Update the project slug in .github/workflows/benchmarks.yml if your Bencher project name differs from pg-stream.

The workflow gracefully degrades — if BENCHER_API_TOKEN is not set, benchmarks still run and upload artifacts but skip Bencher tracking.

Local Bencher-Format Output

To see what Bencher would receive from CI:

just bench-bencher

This runs both suites with --output-format bencher, producing JSON output compatible with bencher run.

Dashboard

Once configured, the Bencher dashboard shows:

• Historical trends for every benchmark across commits
• Statistical thresholds with configurable alerting
• PR annotations highlighting which benchmarks regressed and by how much

Troubleshooting

What Happens When You INSERT a Row?

This tutorial traces the complete lifecycle of a single INSERT statement on a base table that is referenced by a stream table — from the moment the row is written to the moment the stream table reflects the change.

Setup: A Real-World Example

Suppose you run an e-commerce platform. You have an orders table and a stream table that maintains a running total per customer:

-- Base table
CREATE TABLE orders (
    id    SERIAL PRIMARY KEY,
    customer TEXT NOT NULL,
    amount   NUMERIC(10,2) NOT NULL
);

-- Stream table: always-fresh customer totals
SELECT pgstream.create_stream_table(
    'customer_totals',
    $$
      SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count
      FROM orders GROUP BY customer
    $$,
    '1m',           -- refresh when data is staler than 1 minute
    'DIFFERENTIAL'  -- only process changed rows, not the full table
);

After creation, customer_totals is a real PostgreSQL table:

SELECT * FROM customer_totals;
-- (empty — no orders yet)

Phase 1: The INSERT

A new order arrives:

INSERT INTO orders (customer, amount) VALUES ('alice', 49.99);

What happens inside PostgreSQL

When create_stream_table() was called, pg_stream installed an AFTER INSERT OR UPDATE OR DELETE trigger on the orders table. This trigger fires automatically — the user's INSERT statement triggers it transparently.

The trigger function (pgstream_changes.pg_stream_cdc_fn_<oid>()) executes inside the same transaction as the INSERT and writes a single row into the change buffer table:

pgstream_changes.changes_16384    (where 16384 = orders table OID)
┌───────────┬─────────────┬────────┬─────────┬──────────┬──────────┬────────────┐
│ change_id │ lsn         │ action │ pk_hash  │ new_id   │ new_cust │ new_amount │
├───────────┼─────────────┼────────┼─────────┼──────────┼──────────┼────────────┤
│ 1         │ 0/1A3F2B80  │ I      │ -837291 │ 1        │ alice    │ 49.99      │
└───────────┴─────────────┴────────┴─────────┴──────────┴──────────┴────────────┘

Key details:

• lsn: The current WAL Log Sequence Number (pg_current_wal_lsn()), used to bound which changes belong to which refresh cycle.
• action: 'I' for INSERT, 'U' for UPDATE, 'D' for DELETE.
• pk_hash: A pre-computed hash of the primary key (orders.id), used later for efficient row matching.
• new_* columns: The actual column values from NEW, stored as native PostgreSQL types (not JSONB). There are no old_* values for INSERTs.

The trigger adds zero overhead to the user's transaction commit beyond this single INSERT into the buffer table. There is no JSONB serialization, no logical replication slot, and no external process involved.

Phase 2: The Scheduler Wakes Up

A background worker called the scheduler runs inside PostgreSQL (registered via shared_preload_libraries). It wakes up every pg_stream.scheduler_interval_ms milliseconds (default: 1000ms) and performs a tick:

1. Rebuild the DAG (if any stream tables were created/dropped since last tick) — a dependency graph of all stream tables and their source tables.
2. Topological sort — determine the refresh order so that stream tables depending on other stream tables are refreshed after their dependencies.
3. For each stream table, check: has its staleness exceeded its schedule?

For customer_totals with a '1m' schedule, the scheduler compares:

• now() minus data_timestamp (the freshness watermark from the last refresh)
• Against the schedule: 60 seconds

If more than 60 seconds have elapsed and the stream table isn't already being refreshed, the scheduler begins a refresh.

Phase 3: Frontier Advancement

Before executing the refresh, the scheduler creates a new frontier — a snapshot of how far to read changes from each source table:

Previous frontier: { orders(16384): lsn = 0/1A3F2A00 }
New frontier:      { orders(16384): lsn = 0/1A3F2C00 }

The frontier is a DBSP-inspired version vector. Each source table has its own LSN cursor. The refresh will process all changes in the buffer table where lsn > previous_frontier_lsn AND lsn <= new_frontier_lsn.

This means:

• Changes committed before the previous refresh are already reflected.
• Changes committed after the new frontier will be picked up in the next cycle.
• The INSERT we made (lsn = 0/1A3F2B80) falls within this window.

Phase 4: Change Detection — Is There Anything to Do?

Before running the full delta query, the scheduler runs a short-circuit check: does the change buffer actually have any rows in the LSN window?

SELECT count(*)::bigint FROM (
    SELECT 1 FROM pgstream_changes.changes_16384
    WHERE lsn > '0/1A3F2A00'::pg_lsn
    AND lsn <= '0/1A3F2C00'::pg_lsn
    LIMIT <threshold>
) __pgs_capped

This query also checks the adaptive threshold: if the number of changes exceeds a percentage of the source table size (default: 10%), the scheduler falls back to a FULL refresh instead of DIFFERENTIAL, because applying thousands of individual deltas would be slower than a bulk reload.

For our single INSERT, the count is 1 — well below the threshold. The scheduler proceeds with a DIFFERENTIAL refresh.

Phase 5: Delta Query Generation (DVM Engine)

This is where the Differential View Maintenance (DVM) engine does its work. The defining query:

SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count
FROM orders GROUP BY customer

is parsed into an operator tree:

Aggregate(GROUP BY customer, SUM(amount), COUNT(*))
  └── Scan(orders)

The DVM engine differentiates each operator — converting it from "compute the full result" to "compute only what changed":

Step 1: Differentiate the Scan

The Scan(orders) operator becomes a read from the change buffer:

-- Reads only changes in the LSN window, splitting UPDATEs into DELETE+INSERT
WITH __pgs_raw AS (
    SELECT c.pk_hash, c.action,
           c."new_customer", c."old_customer",
           c."new_amount", c."old_amount"
    FROM pgstream_changes.changes_16384 c
    WHERE c.lsn > '0/1A3F2A00'::pg_lsn
    AND   c.lsn <= '0/1A3F2C00'::pg_lsn
)
-- INSERT rows: take new_* values
SELECT pk_hash AS __pgs_row_id, 'I' AS __pgs_action,
       "new_customer" AS customer, "new_amount" AS amount
FROM __pgs_raw WHERE action IN ('I', 'U')
UNION ALL
-- DELETE rows: take old_* values
SELECT pk_hash AS __pgs_row_id, 'D' AS __pgs_action,
       "old_customer" AS customer, "old_amount" AS amount
FROM __pgs_raw WHERE action IN ('D', 'U')

For our single INSERT, this produces:

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
-837291      | I            | alice    | 49.99

Step 2: Differentiate the Aggregate

The Aggregate differentiation is the heart of incremental maintenance. Instead of re-computing SUM(amount) over the entire orders table, it computes:

-- Delta for SUM: add new values, subtract deleted values
SELECT customer,
       SUM(CASE WHEN __pgs_action = 'I' THEN amount
                WHEN __pgs_action = 'D' THEN -amount END) AS total,
       SUM(CASE WHEN __pgs_action = 'I' THEN 1
                WHEN __pgs_action = 'D' THEN -1 END) AS order_count,
       pgstream.pg_stream_hash(customer::text) AS __pgs_row_id,
       'I' AS __pgs_action
FROM <scan_delta>
GROUP BY customer

For our INSERT of ('alice', 49.99), this yields:

customer | total  | order_count | __pgs_row_id | __pgs_action
---------|--------|-------------|--------------|-------------
alice    | +49.99 | +1          | 7283194      | I

The stream table uses reference counting: it tracks __pgs_count (how many source rows contribute to each group). When __pgs_count reaches 0, the group row is deleted.

Phase 6: MERGE Into the Stream Table

The delta is applied to the customer_totals storage table using a single SQL MERGE statement:

MERGE INTO public.customer_totals AS st
USING (<delta_query>) AS d
ON st.__pgs_row_id = d.__pgs_row_id
WHEN MATCHED AND d.__pgs_action = 'D' THEN DELETE
WHEN MATCHED AND d.__pgs_action = 'I' THEN
    UPDATE SET customer = d.customer, total = d.total, order_count = d.order_count
WHEN NOT MATCHED AND d.__pgs_action = 'I' THEN
    INSERT (__pgs_row_id, customer, total, order_count)
    VALUES (d.__pgs_row_id, d.customer, d.total, d.order_count)

Since alice didn't exist before, this is a NOT MATCHED → INSERT. The stream table now contains:

SELECT * FROM customer_totals;
 customer | total | order_count
----------|-------|------------
 alice    | 49.99 | 1

Phase 7: Cleanup and Bookkeeping

After the MERGE succeeds:

1. Consumed changes are deleted from the buffer table:

   DELETE FROM pgstream_changes.changes_16384
   WHERE lsn > '0/1A3F2A00'::pg_lsn
   AND lsn <= '0/1A3F2C00'::pg_lsn
   

2. The frontier is saved to the catalog as JSONB, so the next refresh knows where to start.

3. The refresh is recorded in pgstream.pgs_refresh_history:

   refresh_id | pgs_id | action       | rows_inserted | rows_deleted | status    | initiated_by
   1          | 1      | DIFFERENTIAL | 1             | 0            | COMPLETED | SCHEDULER
   

4. The data timestamp on the stream table is advanced, resetting the staleness clock.

5. The MERGE template is cached in thread-local storage. The next refresh for this stream table skips SQL parsing, operator tree construction, and differentiation — it only substitutes LSN values into the cached template. This saves ~45ms per refresh cycle.

What About UPDATE and DELETE?

UPDATE

UPDATE orders SET amount = 59.99 WHERE id = 1;

The trigger writes a single row with action = 'U', capturing both OLD and NEW values:

action | new_amount | old_amount | new_customer | old_customer
-------|------------|------------|--------------|-------------
U      | 59.99      | 49.99      | alice        | alice

The scan differentiation splits this into:

• DELETE old: (alice, 49.99) with action 'D'
• INSERT new: (alice, 59.99) with action 'I'

The aggregate differentiation computes: +59.99 - 49.99 = +10.00 for alice's total. The MERGE updates the existing row.

DELETE

DELETE FROM orders WHERE id = 1;

The trigger writes action = 'D' with the OLD values. The aggregate differentiation computes -49.99 for the total and -1 for the count. If the __pgs_count reaches 0 (no more orders for alice), the MERGE deletes alice's row from the stream table entirely.

Performance: Why This Is Fast

For a table with 10 million rows and 100 changed rows, a DIFFERENTIAL refresh processes only those 100 rows. A FULL refresh would need to scan all 10 million.

Next in This Series

• What Happens When You UPDATE a Row? — D+I split, group key changes, net-effect for multiple UPDATEs
• What Happens When You DELETE a Row? — Reference counting, group deletion, INSERT+DELETE cancellation
• What Happens When You TRUNCATE a Table? — Why TRUNCATE bypasses triggers and how to recover

What Happens When You UPDATE a Row?

This tutorial traces what happens when an UPDATE statement hits a base table that is referenced by a stream table. It covers the trigger capture, the scan-level decomposition into DELETE + INSERT, and how each DVM operator propagates the change — including cases where the group key changes, where JOINs are involved, and where multiple UPDATEs happen within a single refresh window.

Prerequisite: Read WHAT_HAPPENS_ON_INSERT.md first — it introduces the full 7-phase lifecycle. This tutorial focuses on how UPDATE differs.

Setup

Same e-commerce example:

CREATE TABLE orders (
    id       SERIAL PRIMARY KEY,
    customer TEXT NOT NULL,
    amount   NUMERIC(10,2) NOT NULL
);

SELECT pgstream.create_stream_table(
    'customer_totals',
    $$
      SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count
      FROM orders GROUP BY customer
    $$,
    '1m',
    'DIFFERENTIAL'
);

-- Seed some data
INSERT INTO orders (customer, amount) VALUES
    ('alice', 49.99),
    ('alice', 30.00),
    ('bob',   75.00);

After the first refresh, the stream table contains:

customer | total | order_count
---------|-------|------------
alice    | 79.99 | 2
bob      | 75.00 | 1

Case 1: Simple Value UPDATE (Same Group Key)

UPDATE orders SET amount = 59.99 WHERE id = 1;

Alice's first order changes from 49.99 to 59.99. The customer (group key) stays the same.

Phase 1: Trigger Capture

The AFTER UPDATE trigger fires and writes one row to the change buffer with both OLD and NEW values:

pgstream_changes.changes_16384
┌───────────┬─────────────┬────────┬──────────┬──────────┬────────────┬──────────┬────────────┐
│ change_id │ lsn         │ action │ new_cust │ new_amt  │ old_cust   │ old_amt  │ pk_hash    │
├───────────┼─────────────┼────────┼──────────┼──────────┼────────────┼──────────┼────────────┤
│ 4         │ 0/1A3F3000  │ U      │ alice    │ 59.99    │ alice      │ 49.99    │ -837291    │
└───────────┴─────────────┴────────┴──────────┴──────────┴────────────┴──────────┴────────────┘

Key difference from INSERT: the trigger writes both new_* and old_* columns. The pk_hash is computed from NEW.id.

Phase 2–4: Scheduler, Frontier, Change Detection

Identical to the INSERT flow. The scheduler detects one change row in the LSN window.

Phase 5: Scan Differentiation — The U → D+I Split

This is where UPDATE handling diverges fundamentally. The scan delta operator decomposes the UPDATE into two events:

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
-837291      | D            | alice    | 49.99     ← old values (DELETE)
-837291      | I            | alice    | 59.99     ← new values (INSERT)

Why split into D+I? This is a core IVM principle. Downstream operators (aggregates, joins, filters) don't have special "update" logic — they only understand insertions and deletions. By decomposing the UPDATE:

• The DELETE event subtracts the old values from running aggregates
• The INSERT event adds the new values

This algebraic approach handles arbitrary operator trees without operator-specific update logic.

Phase 5 (continued): Aggregate Differentiation

The aggregate operator processes both events against the alice group:

-- DELETE event: subtract old values
alice: total += CASE WHEN action='D' THEN -49.99 END  →  -49.99
alice: count += CASE WHEN action='D' THEN -1 END       →  -1

-- INSERT event: add new values
alice: total += CASE WHEN action='I' THEN +59.99 END  →  +59.99
alice: count += CASE WHEN action='I' THEN +1 END       →  +1

Net effect on alice's group:

total delta:  -49.99 + 59.99 = +10.00
count delta:  -1 + 1 = 0

The aggregate emits this as an INSERT (because the group still exists and its value changed):

customer | total  | order_count | __pgs_row_id | __pgs_action
---------|--------|-------------|--------------|-------------
alice    | +10.00 | 0           | 7283194      | I

Phase 6: MERGE

The MERGE updates the existing row:

-- MERGE WHEN MATCHED AND action = 'I' THEN UPDATE:
-- alice's total: 79.99 + 10.00 = 89.99  (via reference counting)
-- alice's count: 2 + 0 = 2

Wait — that's not right. The MERGE doesn't add deltas; it replaces the row. The aggregate delta query actually computes the new absolute value by combining the stored state with the delta:

COALESCE(existing.total, 0) + delta.total  → 79.99 + 10.00 = 89.99
COALESCE(existing.__pgs_count, 0) + delta.__pgs_count → 2 + 0 = 2

Result:

SELECT * FROM customer_totals;
 customer | total | order_count
----------|-------|------------
 alice    | 89.99 | 2            ← was 79.99
 bob      | 75.00 | 1

Case 2: Group Key Change (Customer Reassignment)

UPDATE orders SET customer = 'bob' WHERE id = 2;

Alice's second order (amount=30.00) is reassigned to Bob. The group key itself changes.

Trigger Capture

change_id | lsn         | action | new_cust | new_amt | old_cust | old_amt | pk_hash
5         | 0/1A3F3100  | U      | bob      | 30.00   | alice    | 30.00   | 4521038

The old and new customer values differ.

Scan Delta: D+I Split

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
4521038      | D            | alice    | 30.00    ← removes from alice's group
4521038      | I            | bob      | 30.00    ← adds to bob's group

Aggregate Delta

The aggregate groups by customer, so the DELETE and INSERT land in different groups:

Group "alice":
  total delta:  -30.00
  count delta:  -1

Group "bob":
  total delta:  +30.00
  count delta:  +1

After MERGE

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 alice    | 59.99  | 1            ← lost one order (-30.00)
 bob      | 105.00 | 2            ← gained one order (+30.00)

This is why the D+I decomposition is essential. Without it, you'd need special "move between groups" logic. With it, the standard aggregate differentiation handles group key changes naturally.

Case 3: UPDATE That Deletes a Group

-- Alice only has one order left. Reassign it to bob.
UPDATE orders SET customer = 'bob' WHERE id = 1;

Aggregate Delta

Group "alice":
  total delta:    -59.99
  count delta:    -1
  new __pgs_count: 1 - 1 = 0  → group vanishes!

Group "bob":
  total delta:    +59.99
  count delta:    +1

When __pgs_count reaches 0, the aggregate emits a DELETE for alice's group:

customer | total | __pgs_row_id | __pgs_action
---------|-------|--------------|-------------
alice    | —     | 7283194      | D             ← group removed
bob      | ...   | 9182734      | I             ← group updated

The MERGE deletes alice's row entirely:

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 bob      | 165.00 | 3

Case 4: Multiple UPDATEs on the Same Row (Within One Refresh Window)

What if a row is updated multiple times before the next refresh?

UPDATE orders SET amount = 10.00 WHERE id = 3;  -- bob: 75 → 10
UPDATE orders SET amount = 20.00 WHERE id = 3;  -- bob: 10 → 20
UPDATE orders SET amount = 30.00 WHERE id = 3;  -- bob: 20 → 30

The change buffer now has 3 rows for pk_hash of order #3:

change_id | action | old_amt | new_amt
6         | U      | 75.00   | 10.00
7         | U      | 10.00   | 20.00
8         | U      | 20.00   | 30.00

Net-Effect Computation

The scan delta uses a split fast-path design. Since order #3 has multiple changes (cnt > 1), it takes the multi-change path with window functions:

FIRST_VALUE(action) OVER (PARTITION BY pk_hash ORDER BY change_id)  → 'U'
LAST_VALUE(action) OVER (...)                                        → 'U'

Both first and last actions are 'U', so:

• DELETE: emits using old values from the earliest change (change_id=6): old_amt = 75.00
• INSERT: emits using new values from the latest change (change_id=8): new_amt = 30.00

Net delta:

__pgs_row_id | __pgs_action | amount
-------------|--------------|-------
pk_hash_3    | D            | 75.00    ← original value before all changes
pk_hash_3    | I            | 30.00    ← final value after all changes

The aggregate sees -75.00 + 30.00 = -45.00. This is correct regardless of the intermediate values. The intermediate rows (10.00, 20.00) are never seen.

Case 5: INSERT + UPDATE in Same Window

INSERT INTO orders (customer, amount) VALUES ('charlie', 100.00);
UPDATE orders SET amount = 200.00 WHERE customer = 'charlie';

Both happen before the next refresh. The buffer has:

change_id | action | old_amt | new_amt
9         | I      | NULL    | 100.00
10        | U      | 100.00  | 200.00

Net-effect analysis:

• first_action = 'I' (row didn't exist before this window)
• last_action = 'U' (row exists after)

Result:

• No DELETE emitted (first_action = 'I' means the row was born in this window)
• INSERT with final values: (charlie, 200.00)

The aggregate sees a pure insertion of (charlie, 200.00) — the intermediate value of 100.00 never appears.

Case 6: UPDATE + DELETE in Same Window

UPDATE orders SET amount = 999.99 WHERE id = 3;
DELETE FROM orders WHERE id = 3;

Net-effect:

• first_action = 'U' (row existed before)
• last_action = 'D' (row no longer exists)

Result:

• DELETE with original old values from the first change
• No INSERT (last_action = 'D')

The aggregate correctly sees only a removal.

Case 7: UPDATE with JOINs

Consider a stream table that joins two tables:

CREATE TABLE customers (
    id   SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    tier TEXT NOT NULL DEFAULT 'standard'
);

CREATE TABLE orders (
    id          SERIAL PRIMARY KEY,
    customer_id INT REFERENCES customers(id),
    amount      NUMERIC(10,2)
);

SELECT pgstream.create_stream_table(
    'order_details',
    $$
      SELECT c.name, c.tier, o.amount
      FROM orders o
      JOIN customers c ON o.customer_id = c.id
    $$,
    '1m',
    'DIFFERENTIAL'
);

Now update a customer's tier:

UPDATE customers SET tier = 'premium' WHERE name = 'alice';

How the JOIN Delta Works

The join differentiation follows the formula:

$$\Delta(L \bowtie R) = (\Delta L \bowtie R) \cup (L \bowtie \Delta R) - (\Delta L \bowtie \Delta R)$$

Since only the customers table changed:

• $\Delta L$ = changes to orders (empty)
• $\Delta R$ = changes to customers (alice's tier: standard → premium)

So:

• Part 1: $\Delta\text{orders} \bowtie \text{customers}$ = empty (no order changes)
• Part 2: $\text{orders} \bowtie \Delta\text{customers}$ = all of alice's orders joined with her tier change
• Part 3: $\Delta\text{orders} \bowtie \Delta\text{customers}$ = empty (no order changes)

Part 2 produces the delta: for each of alice's orders, DELETE the old row (with tier='standard') and INSERT a new row (with tier='premium').

The stream table is updated to reflect the new tier across all of alice's order rows.

Performance Summary

In all cases, the work is proportional to the number of changed rows, not the total table size. A single UPDATE on a billion-row table produces the same delta cost as on a 10-row table.

Next in This Series

• What Happens When You INSERT a Row? — The full 7-phase lifecycle (start here if you haven't already)
• What Happens When You DELETE a Row? — Reference counting, group deletion, INSERT+DELETE cancellation
• What Happens When You TRUNCATE a Table? — Why TRUNCATE bypasses triggers and how to recover

What Happens When You DELETE a Row?

This tutorial traces what happens when a DELETE statement hits a base table that is referenced by a stream table. It covers the trigger capture, how the scan delta emits a single DELETE event, and how each DVM operator propagates the removal — including group deletion, partial group reduction, JOINs, cascading deletes within a single refresh window, and the important edge case where a DELETE cancels a prior INSERT.

Prerequisite: Read WHAT_HAPPENS_ON_INSERT.md first — it introduces the full 7-phase lifecycle (trigger → scheduler → frontier → change detection → DVM delta → MERGE → cleanup). This tutorial focuses on how DELETE differs.

Setup

Same e-commerce example used throughout the series:

CREATE TABLE orders (
    id       SERIAL PRIMARY KEY,
    customer TEXT NOT NULL,
    amount   NUMERIC(10,2) NOT NULL
);

SELECT pgstream.create_stream_table(
    'customer_totals',
    $$
      SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count
      FROM orders GROUP BY customer
    $$,
    '1m',
    'DIFFERENTIAL'
);

-- Seed some data
INSERT INTO orders (customer, amount) VALUES
    ('alice', 50.00),
    ('alice', 30.00),
    ('bob',   75.00),
    ('bob',   25.00);

After the first refresh, the stream table contains:

customer | total  | order_count
---------|--------|------------
alice    | 80.00  | 2
bob      | 100.00 | 2

Case 1: Delete One Row (Group Survives)

DELETE FROM orders WHERE id = 2;  -- alice's 30.00 order

Alice still has one remaining order (id=1, amount=50.00). The group shrinks but doesn't vanish.

Phase 1: Trigger Capture

The AFTER DELETE trigger fires and writes one row to the change buffer with only OLD values:

pgstream_changes.changes_16384
┌───────────┬─────────────┬────────┬──────────┬──────────┬────────────┬──────────┬────────────┐
│ change_id │ lsn         │ action │ new_cust │ new_amt  │ old_cust   │ old_amt  │ pk_hash    │
├───────────┼─────────────┼────────┼──────────┼──────────┼────────────┼──────────┼────────────┤
│ 5         │ 0/1A3F3000  │ D      │ NULL     │ NULL     │ alice      │ 30.00    │ 4521038    │
└───────────┴─────────────┴────────┴──────────┴──────────┴────────────┴──────────┴────────────┘

Key difference from INSERT and UPDATE:

• new_* columns are all NULL — the row no longer exists, so there are no NEW values
• old_* columns contain the deleted row's data — this is what gets subtracted
• pk_hash is computed from OLD.id (the deleted row's primary key)

Phase 2–4: Scheduler, Frontier, Change Detection

Identical to the INSERT flow. The scheduler detects one change row in the LSN window.

Phase 5: Scan Differentiation — Pure DELETE

Unlike UPDATE (which splits into D+I), a DELETE produces a single event:

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
4521038      | D            | alice    | 30.00

The scan delta applies the net-effect filtering rule:

• first_action = 'D' → row existed before the refresh window
• last_action = 'D' → row does not exist after

Result: emit a DELETE using old values. No INSERT is emitted (because last_action = 'D').

This is the simplest path through the scan delta — one change, one PK, one DELETE event.

Phase 5 (continued): Aggregate Differentiation

The aggregate operator processes the DELETE event against the alice group:

-- DELETE event: subtract old values from alice's group
__ins_count = 0         -- no inserts
__del_count = 1         -- one deletion
__ins_total = 0         -- no amount added
__del_total = 30.00     -- 30.00 removed

The merge CTE joins this delta with the existing stream table state:

new_count = old_count + ins_count - del_count = 2 + 0 - 1 = 1  (still > 0)

Since new_count > 0 and the group already existed (old_count = 2), the action is classified as 'U' (update). The aggregate emits the group with its new values:

customer | total | order_count | __pgs_row_id | __pgs_action
---------|-------|-------------|--------------|-------------
alice    | 50.00 | 1           | 7283194      | I

Note: the 'U' meta-action is emitted as __pgs_action = 'I' because the MERGE treats it as an update-via-INSERT (see aggregate final CTE: CASE WHEN __pgs_meta_action = 'D' THEN 'D' ELSE 'I' END).

Phase 6: MERGE

The MERGE statement matches alice's existing row and updates it:

MERGE INTO customer_totals AS st
USING (...delta...) AS d
ON st.__pgs_row_id = d.__pgs_row_id
WHEN MATCHED AND d.__pgs_action = 'I' THEN
  UPDATE SET customer = d.customer, total = d.total, order_count = d.order_count, ...

Result:

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 alice    | 50.00  | 1            ← was 80.00 / 2
 bob      | 100.00 | 2

Phase 7: Cleanup

The change buffer rows in the consumed LSN window are deleted:

DELETE FROM pgstream_changes.changes_16384
WHERE lsn > '0/1A3F2FFF'::pg_lsn AND lsn <= '0/1A3F3000'::pg_lsn;

Case 2: Delete Last Row in Group (Group Vanishes)

-- Alice has one order left (id=1, amount=50.00). Delete it.
DELETE FROM orders WHERE id = 1;

Trigger Capture

change_id | lsn         | action | old_cust | old_amt | pk_hash
6         | 0/1A3F3100  | D      | alice    | 50.00   | -837291

Scan Delta

Single DELETE event:

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
-837291      | D            | alice    | 50.00

Aggregate Delta

Group "alice":
  ins_count = 0
  del_count = 1
  new_count = old_count + 0 - 1 = 1 - 1 = 0  → group vanishes!

When new_count drops to 0 (or below), the aggregate classifies this as action 'D' (delete). The reference count has reached zero — no rows contribute to this group anymore.

The aggregate emits a DELETE for alice's group:

customer | __pgs_row_id | __pgs_action
---------|--------------|-------------
alice    | 7283194      | D

MERGE

The MERGE matches alice's existing row and deletes it:

WHEN MATCHED AND d.__pgs_action = 'D' THEN DELETE

Result:

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 bob      | 100.00 | 2

Alice's row is completely removed from the stream table. This is the correct behavior — with zero contributing rows, the group should not exist.

Case 3: Delete Multiple Rows (Same Group, Same Window)

-- Delete both of bob's orders before the next refresh
DELETE FROM orders WHERE id = 3;  -- bob, 75.00
DELETE FROM orders WHERE id = 4;  -- bob, 25.00

The change buffer has two rows with different pk_hash values (different PKs):

change_id | action | old_cust | old_amt | pk_hash
7         | D      | bob      | 75.00   | pk_hash_3
8         | D      | bob      | 25.00   | pk_hash_4

Scan Delta

Each PK has exactly one change, so both take the single-change fast path:

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
pk_hash_3    | D            | bob      | 75.00
pk_hash_4    | D            | bob      | 25.00

Two DELETE events, both targeting bob's group.

Aggregate Delta

The aggregate sums both deletions:

Group "bob":
  ins_count = 0
  del_count = 2
  del_total = 75.00 + 25.00 = 100.00
  new_count = 2 + 0 - 2 = 0  → group vanishes!

The aggregate emits a DELETE for bob's group.

MERGE

Bob's row is deleted from the stream table. With both alice and bob gone (from Cases 1+2+3), the stream table is now empty.

Case 4: INSERT + DELETE in Same Window (Cancellation)

What if a row is inserted and then deleted before the next refresh?

INSERT INTO orders (customer, amount) VALUES ('charlie', 200.00);
DELETE FROM orders WHERE customer = 'charlie';

The change buffer has:

change_id | action | new_cust | new_amt | old_cust | old_amt | pk_hash
9         | I      | charlie  | 200.00  | NULL     | NULL    | pk_hash_new
10        | D      | NULL     | NULL    | charlie  | 200.00  | pk_hash_new

Net-Effect Computation

Both changes share the same pk_hash. The pk_stats CTE finds cnt = 2, so this goes through the multi-change path:

first_action = FIRST_VALUE(action) OVER (...) → 'I'
last_action  = LAST_VALUE(action)  OVER (...) → 'D'

The scan delta applies the net-effect filtering:

• DELETE branch: requires first_action != 'I' → FAILS (first_action = 'I')
• INSERT branch: requires last_action != 'D' → FAILS (last_action = 'D')

Result: zero events emitted. The INSERT and DELETE completely cancel each other out.

The aggregate never sees charlie. The stream table is unchanged. This is correct — the row was born and died within the same refresh window, so it should have no visible effect.

Case 5: UPDATE + DELETE in Same Window

UPDATE orders SET amount = 999.99 WHERE id = 3;  -- bob: 75 → 999.99
DELETE FROM orders WHERE id = 3;

The change buffer:

change_id | action | old_amt | new_amt
11        | U      | 75.00   | 999.99
12        | D      | 999.99  | NULL

Net-Effect Computation

Same pk_hash, cnt = 2:

first_action = 'U'  (row existed before this window)
last_action  = 'D'  (row no longer exists)

Filtering:

• DELETE branch: first_action != 'I' → OK. Emit DELETE with old values from the earliest change: old_amt = 75.00
• INSERT branch: last_action != 'D' → FAILS. No INSERT emitted.

Net delta:

__pgs_row_id | __pgs_action | amount
-------------|--------------|-------
pk_hash_3    | D            | 75.00

The intermediate value of 999.99 never appears. The aggregate sees only the removal of the original value (75.00), which is correct — that's the value that was previously accounted for in the stream table.

Case 6: DELETE with JOINs

Consider a stream table that joins two tables:

CREATE TABLE customers (
    id   SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    tier TEXT NOT NULL DEFAULT 'standard'
);

CREATE TABLE orders (
    id          SERIAL PRIMARY KEY,
    customer_id INT REFERENCES customers(id),
    amount      NUMERIC(10,2)
);

SELECT pgstream.create_stream_table(
    'order_details',
    $$
      SELECT c.name, c.tier, o.amount
      FROM orders o
      JOIN customers c ON o.customer_id = c.id
    $$,
    '1m',
    'DIFFERENTIAL'
);

Seed data:

INSERT INTO customers VALUES (1, 'alice', 'premium'), (2, 'bob', 'standard');
INSERT INTO orders VALUES (1, 1, 50.00), (2, 1, 30.00), (3, 2, 75.00);

After refresh, the stream table has:

name  | tier     | amount
------|----------|-------
alice | premium  | 50.00
alice | premium  | 30.00
bob   | standard | 75.00

Now delete an order:

DELETE FROM orders WHERE id = 2;  -- alice's 30.00 order

How the JOIN Delta Works

The join differentiation formula:

$$\Delta(L \bowtie R) = (\Delta L \bowtie R) \cup (L \bowtie \Delta R) - (\Delta L \bowtie \Delta R)$$

Since only the orders table changed:

• $\Delta L$ = changes to orders (one DELETE: order #2)
• $\Delta R$ = changes to customers (empty)

So:

• Part 1: $\Delta\text{orders} \bowtie \text{customers}$ = the deleted order joined with its customer
• Part 2: $\text{orders} \bowtie \Delta\text{customers}$ = empty (no customer changes)
• Part 3: $\Delta\text{orders} \bowtie \Delta\text{customers}$ = empty (customers unchanged)

Part 1 produces:

name  | tier    | amount | __pgs_action
------|---------|--------|-------------
alice | premium | 30.00  | D

The deleted order is joined with alice's customer record to produce a DELETE delta row with the complete joined values.

MERGE

The MERGE matches the row (alice, premium, 30.00) and deletes it:

SELECT * FROM order_details;
 name  | tier     | amount
-------|----------|-------
 alice | premium  | 50.00      ← alice's remaining order
 bob   | standard | 75.00

What About Deleting From the Dimension Table?

DELETE FROM customers WHERE id = 2;  -- remove bob entirely

Now $\Delta R$ has a DELETE for bob, while $\Delta L$ is empty:

• Part 1: $\Delta\text{orders} \bowtie \text{customers}$ = empty
• Part 2: $\text{orders} \bowtie \Delta\text{customers}$ = bob's order(s) joined with deleted customer record

Part 2 produces DELETE events for every order that referenced bob:

name | tier     | amount | __pgs_action
-----|----------|--------|-------------
bob  | standard | 75.00  | D

After MERGE, bob's rows vanish from the stream table.

Note: This assumes referential integrity — if orders still references customer #2, a foreign key constraint would prevent the DELETE in practice. But from the IVM perspective, the join delta correctly handles the removal regardless.

Case 7: Bulk DELETE

DELETE FROM orders WHERE amount < 50.00;

This deletes multiple rows across potentially multiple groups. The trigger fires once per row (it's a FOR EACH ROW trigger), writing one change buffer entry per deleted row:

change_id | action | old_cust | old_amt | pk_hash
13        | D      | alice    | 30.00   | pk_hash_2
14        | D      | bob      | 25.00   | pk_hash_4

Scan Delta

Each deleted PK is independent (different pk_hash values), so each takes the single-change fast path. Two DELETE events:

__pgs_row_id | __pgs_action | customer | amount
-------------|--------------|----------|-------
pk_hash_2    | D            | alice    | 30.00
pk_hash_4    | D            | bob      | 25.00

Aggregate Delta

The aggregate groups these by customer:

Group "alice":
  del_count = 1, del_total = 30.00
  new_count = 2 - 1 = 1  (survives)

Group "bob":
  del_count = 1, del_total = 25.00
  new_count = 2 - 1 = 1  (survives)

Both groups survive (count > 0), so the aggregate emits UPDATE (as 'I') events with new values:

customer | total | order_count
---------|-------|------------
alice    | 50.00 | 1
bob      | 75.00 | 1

The MERGE updates both rows. All work is proportional to the number of deleted rows (2), not the total table size.

Case 8: TRUNCATE (What Doesn't Work)

TRUNCATE orders;

TRUNCATE does not fire row-level triggers. The change buffer receives zero entries. The next refresh cycle will detect no changes and skip the stream table.

This means the stream table becomes stale — it still shows the old aggregate values while the base table is empty.

Workaround: After a TRUNCATE, manually reinitialize:

SELECT pgstream.refresh_stream_table('customer_totals');

With the FULL refresh mode (or by setting the reinitialize flag), the stream table is recomputed from scratch, producing correct results.

Why no TRUNCATE support? PostgreSQL fires statement-level AFTER TRUNCATE triggers, but they don't provide OLD row data. The trigger can detect that a TRUNCATE happened, but cannot enumerate which rows were removed. Supporting TRUNCATE would require reading the stream table to determine what to subtract — effectively a full refresh anyway.

How DELETE Differs From INSERT and UPDATE — A Summary

The Reference Counting Principle

The core insight behind incremental DELETE handling is reference counting. Every aggregate group in the stream table maintains an internal counter (__pgs_count) that tracks how many source rows contribute to the group:

Stream table internal state:
customer | total | order_count | __pgs_count (hidden)
---------|-------|-------------|---------------------
alice    | 80.00 | 2           | 2
bob      | 100.00| 2           | 2

• INSERT → __pgs_count += 1
• DELETE → __pgs_count -= 1
• UPDATE → __pgs_count += 0 (D cancels I for same-group updates)

When __pgs_count reaches 0:

• The group has zero contributing rows
• The aggregate emits a DELETE event
• The MERGE removes the row from the stream table

This is mathematically rigorous — the stream table always reflects the correct result of the defining query over the current base table contents, incrementally maintained through algebraic delta operations.

Performance Summary

In all cases, the work is proportional to the number of changed rows, not the total table size. Deleting 3 rows from a billion-row table produces the same delta cost as from a 10-row table.

Next in This Series

• What Happens When You INSERT a Row? — The full 7-phase lifecycle (start here if you haven't already)
• What Happens When You UPDATE a Row? — D+I split, group key changes, net-effect for multiple UPDATEs
• What Happens When You TRUNCATE a Table? — Why TRUNCATE bypasses triggers and how to recover

What Happens When You TRUNCATE a Table?

This tutorial explains what happens when a TRUNCATE statement hits a base table that is referenced by a stream table. Unlike INSERT, UPDATE, and DELETE — which are fully tracked by the CDC trigger — TRUNCATE is a special case that bypasses row-level triggers entirely. Understanding this gap is essential for operating pg_stream correctly.

Prerequisite: Read WHAT_HAPPENS_ON_INSERT.md first — it introduces the 7-phase lifecycle. This tutorial explains why TRUNCATE breaks that lifecycle and how to recover.

Setup

Same e-commerce example used throughout the series:

CREATE TABLE orders (
    id       SERIAL PRIMARY KEY,
    customer TEXT NOT NULL,
    amount   NUMERIC(10,2) NOT NULL
);

SELECT pgstream.create_stream_table(
    'customer_totals',
    $$
      SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count
      FROM orders GROUP BY customer
    $$,
    '1m',
    'DIFFERENTIAL'
);

-- Seed some data
INSERT INTO orders (customer, amount) VALUES
    ('alice', 50.00),
    ('alice', 30.00),
    ('bob',   75.00),
    ('bob',   25.00);

After the first refresh, the stream table contains:

customer | total  | order_count
---------|--------|------------
alice    | 80.00  | 2
bob      | 100.00 | 2

Case 1: TRUNCATE the Base Table

TRUNCATE orders;

All four rows are removed instantly.

What Happens at the Trigger Level: Nothing

PostgreSQL's TRUNCATE command does not fire row-level AFTER triggers. This is by design in PostgreSQL — TRUNCATE is a DDL-like operation that removes all rows without scanning them individually. The per-row AFTER INSERT OR UPDATE OR DELETE trigger that pg_stream installs is simply never invoked.

The change buffer remains empty:

pgstream_changes.changes_16384
┌───────────┬─────────────┬────────┬──────────┬──────────┐
│ change_id │ lsn         │ action │ ...      │ ...      │
├───────────┼─────────────┼────────┼──────────┼──────────┤
│ (empty)   │             │        │          │          │
└───────────┴─────────────┴────────┴──────────┴──────────┘

No rows. Zero change events captured.

What Happens at the Scheduler: Skip

On the next refresh cycle, the scheduler:

1. Checks the change buffer for rows in the LSN window
2. Finds zero rows
3. Classifies the refresh action as NO_DATA
4. Advances the data timestamp but does not modify the stream table

The stream table still shows the old values:

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 alice    | 80.00  | 2            ← STALE! orders table is empty
 bob      | 100.00 | 2            ← STALE! orders table is empty

The stream table is now stale. It reflects data that no longer exists in the base table.

Why This Happens

The entire incremental view maintenance pipeline depends on the change buffer to know what changed. Without change events, the DVM has no deltas to apply. The stream table's reference-counted aggregates still think there are 4 orders contributing to two groups.

This is not a bug — it's a fundamental limitation of trigger-based CDC. The trigger can only fire when PostgreSQL executes individual row operations, and TRUNCATE deliberately skips per-row processing for performance.

Case 2: How to Recover — Manual Refresh

The fix is straightforward. Force a manual refresh:

SELECT pgstream.refresh_stream_table('customer_totals');

This executes a full refresh regardless of the stream table's configured refresh mode:

1. TRUNCATE the stream table itself (clearing the stale data)
2. Re-execute the defining query: SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count FROM orders GROUP BY customer
3. INSERT the results into the stream table
4. Update the frontier so future differential refreshes start from the current LSN

Since the orders table is empty, the defining query returns zero rows. The stream table becomes empty too:

SELECT pgstream.refresh_stream_table('customer_totals');

SELECT * FROM customer_totals;
 customer | total | order_count
----------|-------|------------
 (0 rows)                        ← correct: orders is empty

The stream table is now consistent again. Future INSERT/UPDATE/DELETE operations on orders will be captured normally by the trigger and propagated incrementally.

Case 3: TRUNCATE Then INSERT (Common ETL Pattern)

A common data loading pattern is:

BEGIN;
TRUNCATE orders;
INSERT INTO orders (customer, amount) VALUES
    ('charlie', 100.00),
    ('charlie', 200.00),
    ('dave',    150.00);
COMMIT;

What the Change Buffer Sees

• TRUNCATE: 0 events (not captured)
• INSERT charlie 100.00: 1 event (captured)
• INSERT charlie 200.00: 1 event (captured)
• INSERT dave 150.00: 1 event (captured)

The change buffer has 3 INSERT events — but knows nothing about the 4 rows that were removed by TRUNCATE.

What the Scheduler Does

The scheduler sees 3 pending changes and runs a DIFFERENTIAL refresh:

Aggregate delta (from INSERTs only):
  Group "charlie": ins_count = 2, ins_total = 300.00
  Group "dave":    ins_count = 1, ins_total = 150.00

The MERGE applies these deltas to the existing stream table:

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 alice    | 80.00  | 2            ← STALE! alice has no orders
 bob      | 100.00 | 2            ← STALE! bob has no orders
 charlie  | 300.00 | 2            ← correct
 dave     | 150.00 | 1            ← correct

The new data (charlie, dave) is correct. But the old data (alice, bob) persists because the TRUNCATE that removed their orders was never captured. The reference counts for alice and bob were never decremented.

How to Fix

SELECT pgstream.refresh_stream_table('customer_totals');

SELECT * FROM customer_totals;
 customer | total  | order_count
----------|--------|------------
 charlie  | 300.00 | 2            ← correct
 dave     | 150.00 | 1            ← correct

Case 4: TRUNCATE a Dimension Table in a JOIN

Consider a stream table that joins two tables:

CREATE TABLE customers (
    id   SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    tier TEXT NOT NULL DEFAULT 'standard'
);

CREATE TABLE orders (
    id          SERIAL PRIMARY KEY,
    customer_id INT REFERENCES customers(id),
    amount      NUMERIC(10,2)
);

SELECT pgstream.create_stream_table(
    'order_details',
    $$
      SELECT c.name, c.tier, o.amount
      FROM orders o
      JOIN customers c ON o.customer_id = c.id
    $$,
    '1m',
    'DIFFERENTIAL'
);

Now truncate the dimension table:

TRUNCATE customers CASCADE;

The CASCADE also truncates orders (due to the foreign key). Neither TRUNCATE fires row-level triggers. The change buffer is empty for both tables.

The stream table continues to show all the old joined rows as if nothing happened. The only recovery is a manual refresh:

SELECT pgstream.refresh_stream_table('order_details');

Case 5: FULL Mode Stream Tables Are Immune

If the stream table uses FULL refresh mode instead of DIFFERENTIAL:

SELECT pgstream.create_stream_table(
    'customer_totals_full',
    $$
      SELECT customer, SUM(amount) AS total, COUNT(*) AS order_count
      FROM orders GROUP BY customer
    $$,
    '1m',
    'FULL'
);

A FULL-mode stream table doesn't use the change buffer at all. Every refresh cycle:

1. TRUNCATEs the stream table
2. Re-executes the defining query
3. Inserts all results

So after a TRUNCATE of the base table, the next scheduled refresh automatically picks up the correct state — no manual intervention needed. The trade-off is that every refresh recomputes from scratch, which is more expensive for large result sets.

Why PostgreSQL Doesn't Fire Row Triggers on TRUNCATE

Understanding the PostgreSQL internals helps explain why this limitation exists:

TRUNCATE is fundamentally different from DELETE. It's an O(1) operation that replaces the table's storage files, while DELETE is O(N) — scanning every row and recording each removal in WAL.

PostgreSQL does support statement-level AFTER TRUNCATE triggers:

CREATE TRIGGER after_truncate_trigger
    AFTER TRUNCATE ON orders
    FOR EACH STATEMENT
    EXECUTE FUNCTION some_function();

However, statement-level TRUNCATE triggers:

• Do not receive OLD row data (there's no OLD record)
• Cannot enumerate which rows were removed
• Only know that a TRUNCATE happened on a specific table

This means a TRUNCATE trigger could detect the event but cannot generate the per-row DELETE events that the DVM pipeline needs.

Alternative: DELETE FROM Instead of TRUNCATE

If you need the stream table to stay consistent without manual intervention, use DELETE FROM instead of TRUNCATE:

-- Instead of: TRUNCATE orders;
DELETE FROM orders;

This is slower (O(N) vs O(1)) but fires the row-level DELETE trigger for every row. The change buffer captures all removals, and the next differential refresh correctly decrements all reference counts, removing groups whose count reaches zero.

For tables with millions of rows, DELETE FROM can be slow and generate significant WAL. In those cases, TRUNCATE followed by a manual refresh is often the pragmatic choice.

Best Practices

1. Always Refresh After TRUNCATE

Make it a habit: if you TRUNCATE a base table, immediately refresh all dependent stream tables:

TRUNCATE orders;
SELECT pgstream.refresh_stream_table('customer_totals');

2. Use DELETE FROM for Small Tables

For tables with fewer than ~100K rows, DELETE FROM is fast enough and keeps everything consistent automatically.

3. Wrap TRUNCATE + Refresh in a Function

For ETL pipelines, create a helper:

CREATE OR REPLACE FUNCTION reload_orders(data jsonb) RETURNS void AS $$
BEGIN
    TRUNCATE orders;
    INSERT INTO orders SELECT * FROM jsonb_populate_recordset(null::orders, data);
    PERFORM pgstream.refresh_stream_table('customer_totals');
END;
$$ LANGUAGE plpgsql;

4. Consider FULL Mode for ETL-Heavy Tables

If a table is routinely truncated and reloaded, FULL refresh mode may be simpler than DIFFERENTIAL — it naturally handles TRUNCATE because it recomputes from scratch every cycle.

5. Monitor for Staleness

pg_stream emits monitoring alerts when a stream table's data deviates from expected freshness. After a TRUNCATE, the data_timestamp still advances (since the scheduler sees no pending changes), but the actual data is stale. The most reliable detection is comparing the stream table results against a fresh query:

-- Quick consistency check
SELECT count(*) FROM customer_totals;  -- still shows rows?
SELECT count(*) FROM orders;            -- should be 0 after TRUNCATE

How TRUNCATE Compares to Other Operations

Summary

TRUNCATE is the one common DML-like operation that falls outside pg_stream's automatic change tracking. The trigger-based CDC architecture captures INSERT, UPDATE, and DELETE perfectly — but TRUNCATE bypasses row-level triggers by design.

The key takeaways:

1. TRUNCATE does not fire row-level triggers — the change buffer stays empty
2. The stream table becomes stale — showing data that no longer exists in the base table
3. Manual refresh fixes it — SELECT pgstream.refresh_stream_table('name') recomputes from scratch
4. FULL mode is immune — every refresh recomputes regardless of change tracking
5. DELETE FROM is the trigger-safe alternative — slower but keeps everything consistent automatically
6. After TRUNCATE, always refresh — make this a standard part of your ETL workflow

Next in This Series

• What Happens When You INSERT a Row? — The full 7-phase lifecycle (start here if you haven't already)
• What Happens When You UPDATE a Row? — D+I split, group key changes, net-effect for multiple UPDATEs
• What Happens When You DELETE a Row? — Reference counting, group deletion, INSERT+DELETE cancellation

Frequently Asked Questions

General

What is pg_stream?

pg_stream is a PostgreSQL 18 extension that implements stream tables — declarative, automatically-refreshing materialized views with Differential View Maintenance (DVM). You define a SQL query and a refresh schedule; the extension handles change capture, delta computation, and incremental refresh automatically.

It is inspired by the DBSP differential dataflow framework. See DBSP_COMPARISON.md for a detailed comparison.

How is this different from PostgreSQL materialized views?

What PostgreSQL versions are supported?

PostgreSQL 18.x exclusively. The extension uses features specific to PostgreSQL 18.

Does pg_stream require wal_level = logical?

No. pg_stream uses lightweight row-level triggers for change data capture, not logical replication. You do not need to set wal_level = logical or configure max_replication_slots.

Is pg_stream production-ready?

pg_stream is under active development. It has a comprehensive test suite (700+ unit tests, 290+ end-to-end tests), but users should evaluate it against their specific workloads before deploying to production.

Installation & Setup

How do I install pg_stream?

1. Add pg_stream to shared_preload_libraries in postgresql.conf:

   shared_preload_libraries = 'pg_stream'
   

2. Restart PostgreSQL.
3. Run:

   CREATE EXTENSION pg_stream;
   

See INSTALL.md for platform-specific instructions and pre-built release artifacts.

What are the minimum configuration requirements?

Only shared_preload_libraries = 'pg_stream' is mandatory (requires a restart). All other settings have sensible defaults. max_worker_processes = 8 is recommended.

Can I install pg_stream on a managed PostgreSQL service (RDS, Cloud SQL, etc.)?

It depends on whether the service allows custom extensions and shared_preload_libraries. Since pg_stream does not require wal_level = logical, it avoids one of the most common restrictions on managed services. Check your provider's documentation for custom extension support.

How do I uninstall pg_stream?

1. Drop all stream tables first (or they will be cascade-dropped):

   SELECT pgstream.drop_stream_table(pgs_name) FROM pgstream.pgs_stream_tables;
   

2. Drop the extension:

   DROP EXTENSION pg_stream CASCADE;
   

3. Remove pg_stream from shared_preload_libraries and restart PostgreSQL.

Creating & Managing Stream Tables

How do I create a stream table?

SELECT pgstream.create_stream_table(
    'order_totals',                                           -- name
    'SELECT customer_id, SUM(amount) AS total
     FROM orders GROUP BY customer_id',                       -- defining query
    '5m',                                                     -- refresh schedule
    'DIFFERENTIAL'                                            -- refresh mode
);

What is the difference between FULL and DIFFERENTIAL refresh mode?

• FULL — Truncates the stream table and re-runs the entire defining query every refresh cycle. Simple but expensive for large result sets.
• DIFFERENTIAL — Computes only the delta (changes since the last refresh) using the DVM engine and applies it via a MERGE statement. Much faster when only a small fraction of source data changes between refreshes.

When should I use FULL vs. DIFFERENTIAL?

Use DIFFERENTIAL (default) when:

• Source tables are large and changes between refreshes are small
• The defining query uses supported operators (most common SQL is supported)

Use FULL when:

• The defining query uses unsupported aggregates (CORR, COVAR_*, REGR_*)
• Source tables are small and a full recompute is cheap
• You see frequent adaptive fallbacks to FULL (check refresh history)

What schedule formats are supported?

Duration strings:

Cron expressions:

CALCULATED mode: Pass NULL as the schedule to inherit the schedule from downstream dependents.

What is the minimum allowed schedule?

The pg_stream.min_schedule_seconds GUC (default: 60) sets the floor. Schedules shorter than this value are rejected. Set to 1 for development/testing.

Can a stream table reference another stream table?

Yes. Stream tables can depend on other stream tables. The scheduler automatically refreshes them in topological order (upstream first). Circular dependencies are detected and rejected at creation time.

-- ST1: aggregates orders
SELECT pgstream.create_stream_table('order_totals',
    'SELECT customer_id, SUM(amount) AS total FROM orders GROUP BY customer_id',
    '1m', 'DIFFERENTIAL');

-- ST2: filters ST1
SELECT pgstream.create_stream_table('big_customers',
    'SELECT customer_id, total FROM pgstream.order_totals WHERE total > 1000',
    '1m', 'DIFFERENTIAL');

How do I change a stream table's schedule or mode?

-- Change schedule
SELECT pgstream.alter_stream_table('order_totals', schedule => '10m');

-- Switch refresh mode
SELECT pgstream.alter_stream_table('order_totals', refresh_mode => 'FULL');

-- Suspend
SELECT pgstream.alter_stream_table('order_totals', status => 'SUSPENDED');

-- Resume
SELECT pgstream.alter_stream_table('order_totals', status => 'ACTIVE');

Can I change the defining query of a stream table?

Not directly. You must drop and recreate the stream table:

SELECT pgstream.drop_stream_table('order_totals');
SELECT pgstream.create_stream_table('order_totals', '<new query>', '5m', 'DIFFERENTIAL');

How do I trigger a manual refresh?

SELECT pgstream.refresh_stream_table('order_totals');

This works even when pg_stream.enabled = false (scheduler disabled).

SQL Support

What SQL features are supported in defining queries?

Most common SQL is supported in both FULL and DIFFERENTIAL modes:

• Table scans, projections, WHERE/HAVING filters
• INNER, LEFT, RIGHT, FULL OUTER JOIN (including multi-table joins)
• GROUP BY with 25+ aggregate functions (COUNT, SUM, AVG, MIN, MAX, BOOL_AND/OR, STRING_AGG, ARRAY_AGG, JSON_AGG, JSONB_AGG, BIT_AND/OR/XOR, STDDEV, VARIANCE, MODE, PERCENTILE_CONT/DISC, and more)
• FILTER (WHERE ...) on aggregates
• DISTINCT
• Set operations: UNION ALL, UNION, INTERSECT, INTERSECT ALL, EXCEPT, EXCEPT ALL
• Subqueries: EXISTS, NOT EXISTS, IN (subquery), NOT IN (subquery), scalar subqueries
• Non-recursive and recursive CTEs
• Window functions (ROW_NUMBER, RANK, SUM OVER, etc.)
• LATERAL joins with set-returning functions and correlated subqueries
• CASE, COALESCE, NULLIF, GREATEST, LEAST, BETWEEN, IS DISTINCT FROM

See DVM Operators for the complete list.

What SQL features are NOT supported?

The following are rejected with clear error messages and suggested rewrites:

The following were previously rejected but are now supported via automatic parse-time rewrites:

Each rejected feature is explained in detail in the Why Are These SQL Features Not Supported? section below.

What happens to ORDER BY in defining queries?

ORDER BY is accepted but silently discarded. Row order in a stream table is undefined (consistent with PostgreSQL's CREATE MATERIALIZED VIEW behavior). Apply ORDER BY when querying the stream table, not in the defining query.

Which aggregates support DIFFERENTIAL mode?

Algebraic (O(changes), fully incremental): COUNT, SUM, AVG

Semi-algebraic (incremental with occasional group rescan): MIN, MAX

Group-rescan (affected groups re-aggregated from source): STRING_AGG, ARRAY_AGG, JSON_AGG, JSONB_AGG, BOOL_AND, BOOL_OR, BIT_AND, BIT_OR, BIT_XOR, JSON_OBJECT_AGG, JSONB_OBJECT_AGG, STDDEV, STDDEV_POP, STDDEV_SAMP, VARIANCE, VAR_POP, VAR_SAMP, MODE, PERCENTILE_CONT, PERCENTILE_DISC, CORR, COVAR_POP, COVAR_SAMP, REGR_AVGX, REGR_AVGY, REGR_COUNT, REGR_INTERCEPT, REGR_R2, REGR_SLOPE, REGR_SXX, REGR_SXY, REGR_SYY

37 aggregate function variants are supported in total.

Change Data Capture (CDC)

How does pg_stream capture changes to source tables?

pg_stream installs AFTER INSERT/UPDATE/DELETE row-level PL/pgSQL triggers on each source table. These triggers write change records (action, old/new row data as JSONB, LSN, transaction ID) into per-source buffer tables in the pgstream_changes schema.

What is the overhead of CDC triggers?

Approximately 20–55 μs per row (PL/pgSQL dispatch + row_to_json() + buffer INSERT). At typical write rates (<1000 writes/sec per source table), this adds less than 5% DML latency overhead.

What happens when I TRUNCATE a source table?

TRUNCATE is now captured via a statement-level AFTER TRUNCATE trigger that writes a T marker row to the change buffer. When the differential refresh engine detects this marker, it automatically falls back to a full refresh for that cycle, ensuring the stream table stays consistent.

Previously, TRUNCATE bypassed row-level triggers entirely. This is no longer a concern — both FULL and DIFFERENTIAL mode stream tables handle TRUNCATE correctly.

Are CDC triggers automatically cleaned up?

Yes. When the last stream table referencing a source is dropped, the trigger and its associated change buffer table are automatically removed.

What happens if a source table is dropped or altered?

pg_stream has DDL event triggers that detect ALTER TABLE and DROP TABLE on source tables. When detected:

• Affected stream tables are marked with needs_reinit = true
• The next refresh cycle performs a full reinitialization (drops and recreates the storage table)
• A reinitialize_needed NOTIFY alert is sent

How do I check if a source table has switched from trigger-based CDC to WAL-based CDC?

When you enable hybrid CDC (pg_stream.cdc_mode = 'auto'), pg_stream starts capturing changes with triggers and can automatically transition to WAL-based logical replication once conditions are met. There are several ways to check the current CDC mode for each source table:

1. Query the dependency catalog directly:

SELECT d.source_relid, c.relname AS source_table, d.cdc_mode,
       d.slot_name, d.decoder_confirmed_lsn, d.transition_started_at
FROM pgstream.pgs_dependencies d
JOIN pg_class c ON c.oid = d.source_relid;

The cdc_mode column shows one of three values:

• TRIGGER — changes are captured via row-level triggers (the default)
• TRANSITIONING — the system is in the process of switching from triggers to WAL
• WAL — changes are captured via logical replication

2. Use the built-in health check function:

SELECT source_table, cdc_mode, slot_name, lag_bytes, alert
FROM pgstream.check_cdc_health();

This returns a row per source table with the current mode, replication slot lag (for WAL-mode sources), and any alert conditions such as slot_lag_exceeds_threshold or replication_slot_missing.

3. Listen for real-time transition notifications:

LISTEN pg_stream_cdc_transition;

pg_stream sends a NOTIFY with a JSON payload whenever a transition starts, completes, or is rolled back. Example payload:

{
  "event": "transition_complete",
  "source_table": "public.orders",
  "old_mode": "TRANSITIONING",
  "new_mode": "WAL",
  "slot_name": "pg_stream_slot_16384"
}

This lets you integrate CDC mode changes into your monitoring stack without polling.

4. Check the global GUC setting:

SHOW pg_stream.cdc_mode;

This shows the desired global behavior (trigger, auto, or wal), not the per-table actual state. The per-table state lives in pgs_dependencies.cdc_mode as described above.

See CONFIGURATION.md for details on the pg_stream.cdc_mode and pg_stream.wal_transition_timeout GUCs.

Is it safe to add triggers to a stream table while the source table is switching CDC modes?

Yes, this is completely safe. CDC mode transitions and user-defined triggers operate on different tables and do not interfere with each other:

• CDC transitions affect how changes are captured from source tables (e.g., orders). The transition switches the capture mechanism from row-level triggers on the source table to WAL-based logical replication.
• User-defined triggers live on stream tables (e.g., order_totals) and control how the refresh engine applies changes to the materialized output.

Because these are independent concerns, you can freely add, modify, or remove triggers on a stream table at any point — including during an active CDC transition on its source tables.

How it works in practice:

1. The refresh engine checks for user-defined triggers on the stream table at the start of each refresh cycle (via a fast pg_trigger lookup, <0.1 ms).
2. If user triggers are detected, the engine uses explicit DELETE / UPDATE / INSERT statements instead of MERGE, so your triggers fire with correct TG_OP, OLD, and NEW values.
3. The change data consumed by the refresh engine has the same format regardless of whether it came from CDC triggers or WAL decoding — so the trigger detection and the CDC mode are fully decoupled.

A trigger added between two refresh cycles will simply be picked up on the next cycle. The only (theoretical) edge case is adding a trigger in the tiny window during a single refresh transaction, between the trigger-detection check and the MERGE execution — but since both happen within the same transaction, this is virtually impossible in practice.

Performance & Tuning

How do I tune the scheduler interval?

The pg_stream.scheduler_interval_ms GUC controls how often the scheduler checks for stale stream tables (default: 1000 ms).

What is the adaptive fallback to FULL?

When the number of pending changes exceeds pg_stream.differential_max_change_ratio (default: 15%) of the source table size, DIFFERENTIAL mode automatically falls back to FULL for that refresh cycle. This prevents pathological delta queries on bulk changes.

• Set to 0.0 to always use DIFFERENTIAL (even on large change sets)
• Set to 1.0 to effectively always use FULL
• Default 0.15 (15%) is a good balance

How many concurrent refreshes can run?

Controlled by pg_stream.max_concurrent_refreshes (default: 4, range: 1–32). Each concurrent refresh uses a background worker. Increase this if you have many stream tables and available CPU/IO.

How do I check if my stream tables are keeping up?

-- Quick overview
SELECT pgs_name, status, staleness, stale
FROM pgstream.stream_tables_info;

-- Detailed statistics
SELECT pgs_name, total_refreshes, avg_duration_ms, consecutive_errors, stale
FROM pgstream.pg_stat_stream_tables;

-- Recent refresh history for a specific ST
SELECT * FROM pgstream.get_refresh_history('order_totals', 10);

What is __pgs_row_id?

Every stream table has a __pgs_row_id BIGINT PRIMARY KEY column. It stores a 64-bit xxHash of the row's group-by key (or all columns for non-aggregate queries). The refresh engine uses it for delta MERGE operations (matching DELETEs and INSERTs by row ID).

You should ignore this column in your queries. It is an implementation detail.