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_<key> with a hidden non-NULL count, so a group that loses all its values returns NULL rather than 0. Readers see openivm_visible_<key>, the published result, through a plain view. Changes to sales are captured in openivm_delta_sales.

CREATE TABLE openivm_data_<key> 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 TABLE openivm_visible_<key> AS
    SELECT * EXCLUDE (openivm_nonnull_sum_count_1), CAST(0 AS BIGINT) AS openivm_ordinal
    FROM openivm_data_<key>;

CREATE VIEW regional_totals AS
    SELECT * EXCLUDE (openivm_ordinal) FROM openivm_visible_<key>;

CREATE UNIQUE INDEX openivm_data_<key>openivm_index ON openivm_data_<key>(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_<key>
    (region, total, cnt, openivm_nonnull_sum_count_1, openivm_multiplicity)
SELECT * FROM t2_block;

The view's delta is folded with SUM(multiplicity × value) and merged on the group key; groups whose count drops to zero are deleted.

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_<key>
    WHERE  openivm_timestamp > $last_update
    GROUP BY region
)
MERGE INTO openivm_data_<key> 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_<key> WHERE COALESCE(cnt, 0) = 0;

Only the affected groups are compared with the published table. Their signed difference replaces the old visible rows and is recorded as a delta, which is what downstream views consume.

CREATE TEMP TABLE openivm_publish_changes_<key> AS
SELECT region, total, cnt, openivm_ordinal, SUM(openivm_publish_weight) AS openivm_publish_weight
FROM (
    SELECT *, -1 AS openivm_publish_weight            -- old visible rows of affected groups
    FROM openivm_visible_<key> v
    WHERE EXISTS (SELECT 1 FROM openivm_delta_<key> d WHERE v.region IS NOT DISTINCT FROM d.region)
    UNION ALL
    SELECT *, +1 FROM openivm_publish_next_<key>      -- their new rows
) signed_rows
GROUP BY region, total, cnt, openivm_ordinal
HAVING SUM(openivm_publish_weight) <> 0;

-- then: append the signed changes to openivm_delta_openivm_visible_<key>,
-- delete the old rows from openivm_visible_<key> and insert the new ones

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_<key>;

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 = '<key>' AND table_name = 'openivm_delta_sales';