A SQL-to-SQL compiler for incremental computation.

A grouped aggregate over one table. This is all you write.

CREATE MATERIALIZED VIEW regional_totals AS
    SELECT region, SUM(amount) AS total, COUNT(*) AS cnt
    FROM sales
    GROUP BY region;

The view is stored in openivm_data_regional_totals with a hidden helper column (the non-NULL count behind SUM, so a group that loses all its values returns NULL, not 0). A plain view hides it; a delta table is created for sales.

CREATE TABLE openivm_data_regional_totals AS
SELECT    region, sum(amount) AS total, count_star() AS cnt,
          count(amount) AS openivm_nonnull_sum_count_1
FROM      sales
GROUP BY  region;

CREATE VIEW regional_totals AS
    SELECT * EXCLUDE (openivm_nonnull_sum_count_1) FROM openivm_data_regional_totals;

CREATE TABLE IF NOT EXISTS openivm_delta_sales AS
    SELECT *, 1::INTEGER AS openivm_multiplicity,
           make_timestamp(epoch_us(now())) AS openivm_timestamp
    FROM sales LIMIT 0;

CREATE UNIQUE INDEX openivm_data_regional_totalsopenivm_index
    ON openivm_data_regional_totals(region);

On refresh, only rows newer than the view's cursor are read. Duplicate deltas are consolidated first (a +1 and a −1 for the same tuple cancel), then the aggregate is applied to the delta itself, keeping the multiplicity as a group key.

WITH
t0_scan (t0_region, t0_amount, t0_openivm_multiplicity) AS (
    SELECT  region, amount, openivm_multiplicity
    FROM    openivm_delta_sales
    WHERE   openivm_timestamp >= $last_update
),
t1_aggregate (t8_region, t8_amount, t9_openivm_multiplicity) AS (
    SELECT    t0_region, t0_amount, sum(t0_openivm_multiplicity)
    FROM      t0_scan
    GROUP BY  t0_region, t0_amount
),
t2_block (t2_region, total, cnt, openivm_nonnull_sum_count_1, t1_openivm_multiplicity) AS (
    SELECT    t8_region, sum(t8_amount), count_star(), count(t8_amount),
              CAST(t9_openivm_multiplicity AS INTEGER)
    FROM      t1_aggregate
    WHERE     t9_openivm_multiplicity != 0
    GROUP BY  t8_region, CAST(t9_openivm_multiplicity AS INTEGER)
)
INSERT INTO openivm_delta_regional_totals
    (region, total, cnt, openivm_nonnull_sum_count_1, openivm_multiplicity)
SELECT * FROM t2_block;

The view's own delta is folded with SUM(multiplicity × value) and merged on the group key. Groups whose count drops to zero are deleted. Downstream views can consume openivm_delta_regional_totals before it is cleared.

WITH refresh_cte AS (
    SELECT region,
           sum(openivm_multiplicity * total) AS total,
           sum(openivm_multiplicity * cnt)   AS cnt,
           sum(openivm_multiplicity * openivm_nonnull_sum_count_1) AS openivm_nonnull_sum_count_1
    FROM   openivm_delta_regional_totals
    WHERE  openivm_timestamp > $last_update
    GROUP BY region
)
MERGE INTO openivm_data_regional_totals v USING refresh_cte d
ON v.region IS NOT DISTINCT FROM d.region
WHEN MATCHED THEN UPDATE SET
    total = CASE WHEN COALESCE(v.openivm_nonnull_sum_count_1 + d.openivm_nonnull_sum_count_1,
                               v.openivm_nonnull_sum_count_1, d.openivm_nonnull_sum_count_1) = 0
                 THEN NULL
                 ELSE COALESCE(v.total + d.total, v.total, d.total) END,
    cnt   = COALESCE(v.cnt + d.cnt, v.cnt, d.cnt),
    openivm_nonnull_sum_count_1 = COALESCE(v.openivm_nonnull_sum_count_1 + d.openivm_nonnull_sum_count_1,
                                           v.openivm_nonnull_sum_count_1, d.openivm_nonnull_sum_count_1)
WHEN NOT MATCHED THEN INSERT (region, total, cnt, openivm_nonnull_sum_count_1)
    VALUES (d.region, d.total, d.cnt, d.openivm_nonnull_sum_count_1);

DELETE FROM openivm_data_regional_totals WHERE COALESCE(cnt, 0) = 0;

Consumed deltas are removed once every dependent view has moved past them, and the cursor advances to MAX(timestamp) + 1µs of what this refresh actually saw, not now(), so a concurrent insert is never skipped or applied twice.

DELETE FROM openivm_delta_regional_totals;

DELETE FROM openivm_delta_sales
WHERE openivm_timestamp < (SELECT MIN(last_update) FROM openivm_delta_tables
                           WHERE table_name = 'openivm_delta_sales');

UPDATE openivm_delta_tables
SET last_update = COALESCE((SELECT MAX(openivm_timestamp) + INTERVAL '1 microsecond'
                            FROM openivm_delta_sales),
                           make_timestamp(epoch_us(now()))),
    last_refresh_ts = make_timestamp(epoch_us(now()))
WHERE view_name = 'regional_totals' AND table_name = 'openivm_delta_sales';