diff --git a/CHANGELOG.md b/CHANGELOG.md index aa5d0acc..d69446de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -142,6 +142,25 @@ true until the next version shipped. long, and refuses, before creating anything, a call that would create more than 1000 partitions (`22023`) and a parent with a partition bounded by `MAXVALUE` (`55000`). +- Incremental rollups, phase 1 (#1294). `pgcolumnar.create_rollup` defines an + aggregate per `date_bin` time bucket and `group_by` key (count, sum, min, max, + avg), kept as a view over a partials table; `pgcolumnar.refresh_rollup` folds + only the row groups appended since the last refresh, in one statement, reading + each new group through its own TID range. Late rows are folded into their + buckets, rows committed during a refresh are folded by the next one, and two + concurrent refreshes fold each row once. The rollup equals a full recompute: + exactly for counts, integer sums and averages, min and max, and within 1e-12 + for `float8` averages. Rows that `expire` retires keep their buckets in the + rollup, so the rollup's history outlives the raw data. This phase refuses any + other delete or update in the source (`0A000`) and a rewritten source + (`55000`, until `pgcolumnar.rebase_rollup`). Each group's TID range comes from + the access method's own mapping (`pgcolumnar.row_tid`), not a block size + restated in SQL. + The rollup catalogs carry row-level security, so each role sees only the + rollups it owns. Found while it was written: a lateral per-group subquery + pulled up into a join read every row group (114 blocks of a 103-block table + for one new group), because PostgreSQL builds a TID range scan only from a + relation's own restrictions; `OFFSET 0` keeps it a scan (15 blocks). ### Changed diff --git a/docs/how-to.md b/docs/how-to.md index e7d32c0c..606a77ec 100644 --- a/docs/how-to.md +++ b/docs/how-to.md @@ -457,6 +457,56 @@ When a partition is dropped by retention, its pgcolumnar catalog rows go with it `test/partman_columnar.sh` checks each of these statements. +## Keep a rollup up to date + +A rollup keeps an aggregate per time bucket, such as hourly averages, and folds +in only the rows added since the last refresh. Queries read the rollup instead +of the raw table. + +```sql +SELECT pgcolumnar.create_rollup( + 'metrics_hourly', 'metrics', 'ts', '1 hour', '2026-01-01 00:00:00+00', + group_by => '{device}', + agg_functions => '{count,sum,min,max,avg}', + agg_columns => '{*,value,value,value,value}'); + +SELECT pgcolumnar.refresh_rollup('metrics_hourly'); -- returns rows folded +SELECT * FROM metrics_hourly WHERE bucket >= now() - interval '1 day'; +``` + +The rollup is a view with a `bucket` column, the `group_by` columns, and one +column for each aggregate. The aggregate columns are named `count`, or the +function and the column, as in `sum_value` and `avg_value`. The buckets come +from `date_bin(width, ts, origin)`, so every session gets the same buckets, +whatever its time zone. The width cannot include months. + +A refresh reads only the row groups added since the last refresh. Rows that +arrive late, with old timestamps, are still folded into their buckets. Rows +that commit while a refresh runs are folded by the next refresh. Two refreshes +that run at the same time fold each row once. A refresh after a small append +reads a small part of the table. + +The rollup equals what the same `GROUP BY` over the whole table returns. Count, +sum, min and max are exact, and so is the average of an integer column. The +average of a `float8` column can differ from a full recompute in the last digits. + +**Retention keeps the rollup's history.** When `pgcolumnar.expire`, or the +maintenance daemon, drops old rows of the source, the rollup keeps its rows for +those buckets. So, for example, hourly averages can outlive the raw data. + +**This version folds appended rows only.** If rows of the source are deleted or +updated in any other way, `refresh_rollup` refuses (`0A000`) and changes +nothing. If the source +is rewritten, for example by `pgcolumnar.vacuum`, `refresh_rollup` refuses with +`55000`. If the rollup was current before the rewrite, run +`pgcolumnar.rebase_rollup` to record the rewritten table and continue. + +Run `refresh_rollup` at READ COMMITTED. A role that is not a superuser needs +`USAGE` on schema `pgcolumnar`, which a superuser grants. Each role then sees +and refreshes only the rollups that it owns. + +`test/rollup.sh` checks each of these statements. + ## Keep tables optimized automatically A background daemon can compact and recluster tables on a schedule. The threshold diff --git a/docs/sql-reference.md b/docs/sql-reference.md index 76546e30..c4f25322 100644 --- a/docs/sql-reference.md +++ b/docs/sql-reference.md @@ -591,6 +591,37 @@ refused before it creates any (`22023`). The function takes `SHARE UPDATE EXCLUSIVE` on the parent, never `ACCESS EXCLUSIVE`. See [Create time partitions ahead](how-to.md#create-time-partitions-ahead). +### pgcolumnar.create_rollup(rollup name, source regclass, ts_column name, width interval, origin timestamptz, group_by name[], agg_functions text[], agg_columns name[], access_method name DEFAULT 'pgcolumnar') returns regclass + +Creates a rollup of `source` in the source's schema. A rollup is a view of +`date_bin(width, ts_column, origin)` buckets per `group_by` key, with one column +per aggregate. Each aggregate is a function from `agg_functions` (`count`, +`sum`, `min`, `max` or `avg`) applied to the column at the same position in +`agg_columns`. `*` goes only with `count`. `ts_column` must be `timestamptz`. +The data is kept in a table named `__partials`, with access method +`access_method`. The function refuses a width with months, an unknown function, +arrays of different lengths, and a column the source does not have (`22023`). +See [Keep a rollup up to date](how-to.md#keep-a-rollup-up-to-date). + +### pgcolumnar.refresh_rollup(rollup regclass) returns bigint + +Folds the rows of the row groups added to the source since the last refresh, +and returns the number of rows folded. Runs at READ COMMITTED only (`25000`). +Row groups that `pgcolumnar.expire` retired are dropped from the refresh's +record, and the rollup keeps their buckets. Refuses after any other delete or +update in the source (`0A000`), and after a rewrite of the source (`55000`). + +### pgcolumnar.row_tid(row_number bigint) returns tid + +Returns the item pointer that a columnar table gives the row with this row +number. Row numbers start at 1 (`22003` otherwise). + +### pgcolumnar.rebase_rollup(rollup regclass) returns bigint + +Records the source's current row groups as folded, without changing the rollup, +and returns the number of row groups recorded. Use it after a rewrite of the +source, when the rollup was current before the rewrite. + ### pgcolumnar.import_arrow(rel regclass, path text) returns bigint Inserts the rows of an Arrow IPC stream file at `path` into the existing table diff --git a/pgcolumnar--1.0-alpha4--1.0-alpha5.sql b/pgcolumnar--1.0-alpha4--1.0-alpha5.sql index 838800d7..8fd0c65c 100644 --- a/pgcolumnar--1.0-alpha4--1.0-alpha5.sql +++ b/pgcolumnar--1.0-alpha4--1.0-alpha5.sql @@ -982,3 +982,434 @@ CREATE TABLE pgcolumnar.retired_group ( ); CREATE UNIQUE INDEX retired_group_pkey ON pgcolumnar.retired_group USING btree (storage_id, group_number); + + +/* --------------------------------------------------------------------------- + * Incremental rollups, phase 1: append-only (#1294). + * + * A rollup is a VIEW over a partials table. The partials table holds, per + * (bucket, group_by...) key, the parts each aggregate combines from: count, sum, + * min, max, and for avg a sum and a count. The view turns those into the + * aggregates. refresh_rollup folds the rows of row groups that are new since the + * last refresh, in ONE statement, so the group list it reads, the rows it folds + * and the groups it records all come from one snapshot (#1294 point 1). + * + * pgcolumnar.rollup and pgcolumnar.rollup_group are readable and writable by + * anyone, row-level security limiting each role to the rollups it owns: the + * refresh runs as its caller and writes them in that statement. + * ------------------------------------------------------------------------- */ +CREATE TABLE pgcolumnar.rollup ( + rollup regclass PRIMARY KEY, -- the view + partials regclass NOT NULL, -- the table behind it + source regclass NOT NULL, + source_storage bigint NOT NULL, -- the source's storage id when recorded + ts_column name NOT NULL, + width interval NOT NULL, + origin timestamptz NOT NULL, + group_by name[] NOT NULL, + agg_functions text[] NOT NULL, + agg_columns name[] NOT NULL +); + +CREATE TABLE pgcolumnar.rollup_group ( + rollup regclass NOT NULL, + group_number bigint NOT NULL, + row_count bigint NOT NULL, + deleted_count bigint NOT NULL, + PRIMARY KEY (rollup, group_number) +); + +ALTER TABLE pgcolumnar.rollup ENABLE ROW LEVEL SECURITY; +ALTER TABLE pgcolumnar.rollup_group ENABLE ROW LEVEL SECURITY; +CREATE POLICY rollup_owner ON pgcolumnar.rollup + USING (pg_has_role(current_user, + (SELECT c.relowner FROM pg_class c WHERE c.oid = rollup), 'USAGE')); +CREATE POLICY rollup_group_owner ON pgcolumnar.rollup_group + USING (pg_has_role(current_user, + (SELECT c.relowner FROM pg_class c WHERE c.oid = rollup), 'USAGE')); +GRANT SELECT, INSERT, UPDATE, DELETE ON pgcolumnar.rollup, pgcolumnar.rollup_group TO PUBLIC; + +/* + * The source's row groups, with each one's deleted count. SECURITY DEFINER + * because row_group and delete_vector carry no GRANT; the caller must be able to + * SELECT the source. STABLE, so it reads with the calling statement's snapshot + * and refresh_rollup keeps to one snapshot. + */ +-- The item pointer the access method gives a row number (#1319 review). +CREATE FUNCTION pgcolumnar.row_tid(row_number bigint) + RETURNS tid + LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE + AS 'MODULE_PATHNAME', 'pgcolumnar_row_tid'; + +CREATE FUNCTION pgcolumnar.rollup_source_groups(source regclass) + RETURNS TABLE (group_number bigint, first_row_number bigint, row_count bigint, + deleted_count bigint) + LANGUAGE plpgsql STABLE SECURITY DEFINER + SET search_path = pg_catalog, pg_temp +AS $rsg$ +BEGIN + PERFORM pgcolumnar.require_caller_select(source); + RETURN QUERY + SELECT g.group_number, g.first_row_number, g.row_count, + coalesce(d.deleted_count, 0)::bigint + FROM pgcolumnar.row_group g + LEFT JOIN pgcolumnar.delete_vector d + ON d.storage_id = g.storage_id AND d.group_number = g.group_number + WHERE g.storage_id = pgcolumnar.get_storage_id(source); +END +$rsg$; + +/* + * The source's row groups that retention retired (pgcolumnar.retired_group, + * written only by expire, #1314). SECURITY DEFINER because that catalog carries + * no GRANT; the caller must be able to SELECT the source. + */ +CREATE FUNCTION pgcolumnar.rollup_source_retired(source regclass) + RETURNS TABLE (group_number bigint) + LANGUAGE plpgsql STABLE SECURITY DEFINER + SET search_path = pg_catalog, pg_temp +AS $rsr$ +BEGIN + PERFORM pgcolumnar.require_caller_select(source); + RETURN QUERY + SELECT r.group_number FROM pgcolumnar.retired_group r + WHERE r.storage_id = pgcolumnar.get_storage_id(source); +END +$rsr$; + +CREATE FUNCTION pgcolumnar.create_rollup(rollup name, source regclass, ts_column name, + width interval, origin timestamptz, + group_by name[], agg_functions text[], + agg_columns name[], + access_method name DEFAULT 'pgcolumnar') + RETURNS regclass + LANGUAGE plpgsql + SET search_path = pg_catalog, pg_temp +AS $cr$ +DECLARE + nsp name; + tstype regtype; + cols text := ''; + vcols text := ''; + gcols text := ''; + i int; + fn text; + col name; + ctype regtype; + ptype text; + partials text; + view_oid regclass; +BEGIN + IF width IS NULL OR width <= interval '0' OR extract(month FROM width) <> 0 + OR extract(year FROM width) <> 0 THEN + RAISE EXCEPTION 'a rollup width must be positive and have no months or years' + USING ERRCODE = 'invalid_parameter_value', + DETAIL = 'date_bin, which buckets the rows, takes no width in months.'; + END IF; + IF agg_functions IS NULL OR agg_columns IS NULL + OR cardinality(agg_functions) = 0 + OR cardinality(agg_functions) <> cardinality(agg_columns) THEN + RAISE EXCEPTION 'agg_functions and agg_columns must be non-empty and of the same length' + USING ERRCODE = 'invalid_parameter_value'; + END IF; + SELECT a.atttypid::regtype INTO tstype FROM pg_attribute a + WHERE a.attrelid = source AND a.attname = ts_column AND NOT a.attisdropped; + IF tstype IS DISTINCT FROM 'timestamptz'::regtype THEN + RAISE EXCEPTION 'column "%" of % is not a timestamptz column', ts_column, source + USING ERRCODE = 'invalid_parameter_value'; + END IF; + + -- Each aggregate checked against the five that combine across partial + -- results, and each column against the source: structured input, never SQL + -- text (#1294 point 11). + FOR i IN 1 .. cardinality(agg_functions) LOOP + fn := agg_functions[i]; + col := agg_columns[i]; + IF fn NOT IN ('count', 'sum', 'min', 'max', 'avg') THEN + RAISE EXCEPTION '"%" is not an aggregate a rollup can combine', fn + USING ERRCODE = 'invalid_parameter_value', + HINT = 'Use count, sum, min, max or avg.'; + END IF; + IF col = '*' THEN + IF fn <> 'count' THEN + RAISE EXCEPTION '* goes only with count, not %', fn + USING ERRCODE = 'invalid_parameter_value'; + END IF; + cols := cols || ', count bigint NOT NULL'; + vcols := vcols || ', count'; + CONTINUE; + END IF; + SELECT a.atttypid::regtype INTO ctype FROM pg_attribute a + WHERE a.attrelid = source AND a.attname = col AND a.attnum > 0 AND NOT a.attisdropped; + IF ctype IS NULL THEN + RAISE EXCEPTION 'column "%" does not exist in %', col, source + USING ERRCODE = 'invalid_parameter_value'; + END IF; + ptype := CASE WHEN ctype IN ('int2'::regtype, 'int4'::regtype, 'int8'::regtype, + 'numeric'::regtype) THEN 'numeric' + WHEN ctype IN ('float4'::regtype, 'float8'::regtype) THEN 'float8' + ELSE NULL END; + IF fn IN ('sum', 'avg') AND ptype IS NULL THEN + RAISE EXCEPTION '% of a % column cannot be folded', fn, ctype + USING ERRCODE = 'invalid_parameter_value'; + END IF; + CASE fn + WHEN 'count' THEN + cols := cols || format(', %I bigint NOT NULL', 'count_' || col); + vcols := vcols || format(', %I', 'count_' || col); + WHEN 'sum' THEN + cols := cols || format(', %I %s', 'sum_' || col, ptype); + vcols := vcols || format(', %I', 'sum_' || col); + WHEN 'min', 'max' THEN + cols := cols || format(', %I %s', fn || '_' || col, ctype); + vcols := vcols || format(', %I', fn || '_' || col); + WHEN 'avg' THEN + cols := cols || format(', %I %s, %I bigint NOT NULL', 'avg_' || col || '__sum', ptype, + 'avg_' || col || '__n'); + vcols := vcols || format(', (%I / nullif(%I, 0))::%s AS %I', 'avg_' || col || '__sum', + 'avg_' || col || '__n', ptype, 'avg_' || col); + END CASE; + END LOOP; + + IF group_by IS NOT NULL THEN + FOREACH col IN ARRAY group_by LOOP + SELECT a.atttypid::regtype INTO ctype FROM pg_attribute a + WHERE a.attrelid = source AND a.attname = col AND a.attnum > 0 AND NOT a.attisdropped; + IF ctype IS NULL THEN + RAISE EXCEPTION 'group_by column "%" does not exist in %', col, source + USING ERRCODE = 'invalid_parameter_value'; + END IF; + gcols := gcols || format(', %I %s', col, ctype); + END LOOP; + END IF; + + -- In the source's schema: this function's search_path is pg_catalog, so + -- current_schema() would name the system schema, not the caller's. + SELECT n.nspname INTO nsp FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.oid = source; + partials := rollup || '__partials'; + EXECUTE format('CREATE TABLE %I.%I (bucket timestamptz%s%s) USING %I', + nsp, partials, gcols, cols, access_method); + EXECUTE format('CREATE VIEW %I.%I AS SELECT bucket%s%s FROM %I.%I', + nsp, rollup, + (SELECT coalesce(string_agg(format(', %I', g), ''), '') FROM unnest(group_by) g), + vcols, nsp, partials); + view_oid := format('%I.%I', nsp, rollup)::regclass; + INSERT INTO pgcolumnar.rollup VALUES + (view_oid, format('%I.%I', nsp, partials)::regclass, source, + pgcolumnar.get_storage_id(source), ts_column, width, origin, + coalesce(group_by, '{}'), agg_functions, agg_columns); + RETURN view_oid; +END +$cr$; + +COMMENT ON FUNCTION pgcolumnar.create_rollup(name, regclass, name, interval, timestamptz, name[], text[], name[], name) + IS 'create an incremental rollup: a view of date_bin(width, ts_column, origin) buckets per group_by key, with count, sum, min, max or avg per (agg_functions, agg_columns); fill it with refresh_rollup (#1294)'; + +/* + * The fold, one statement for one snapshot. Built per rollup from its catalog + * row: every identifier through %I and every value through %L. + */ +CREATE FUNCTION pgcolumnar.refresh_rollup(rollup regclass) + RETURNS bigint + LANGUAGE plpgsql + SET search_path = pg_catalog, pg_temp +AS $rr$ +DECLARE + r pgcolumnar.rollup; + changed text; + keys text := ''; + keyeq text := ''; + rowcols text; + rcols name[]; + dsel text := ''; + icols text := ''; + isel text := ''; + uset text := ''; + i int; + fn text; + col name; + c text; + g name; + folded bigint; +BEGIN + -- A REPEATABLE READ caller's snapshot predates the lock below, so a refresh + -- that waited for another would fold the same groups again. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'refresh_rollup runs only at READ COMMITTED, not %', + upper(current_setting('transaction_isolation')) + USING ERRCODE = 'invalid_transaction_state'; + END IF; + SELECT * INTO r FROM pgcolumnar.rollup x WHERE x.rollup = refresh_rollup.rollup; + IF NOT FOUND THEN + RAISE EXCEPTION '% is not a rollup', rollup USING ERRCODE = 'undefined_object'; + END IF; + -- Self-conflicting, so two refreshes run one after the other; readers pass. + EXECUTE format('LOCK TABLE %s IN SHARE UPDATE EXCLUSIVE MODE', r.partials); + + IF pgcolumnar.get_storage_id(r.source) <> r.source_storage THEN + RAISE EXCEPTION 'the source of rollup % was rewritten since its last refresh', rollup + USING ERRCODE = 'object_not_in_prerequisite_state', + HINT = 'If the rollup is current, pgcolumnar.rebase_rollup records the source''s new row groups.'; + END IF; + -- Retention keeps history (the owner's decision on #1294): a recorded group + -- that expire retired leaves the snapshot, and its buckets keep the rows they + -- folded. Retirement is permanent, so this need not share the fold's snapshot. + DELETE FROM pgcolumnar.rollup_group rg + WHERE rg.rollup = r.rollup + AND rg.group_number IN (SELECT t.group_number FROM pgcolumnar.rollup_source_retired(r.source) t); + -- Phase 1 folds appends only: any other recorded group that is gone, or has + -- new deletions, needs phase 2's dirty buckets. + SELECT string_agg(rg.group_number::text, ', ' ORDER BY rg.group_number) INTO changed + FROM pgcolumnar.rollup_group rg + LEFT JOIN pgcolumnar.rollup_source_groups(r.source) s USING (group_number) + WHERE rg.rollup = r.rollup + AND (s.group_number IS NULL OR s.deleted_count <> rg.deleted_count); + IF changed IS NOT NULL THEN + RAISE EXCEPTION 'rollup %: row groups % of its source were deleted from or rewritten', rollup, changed + USING ERRCODE = 'feature_not_supported', + DETAIL = 'This version folds appended row groups only; changed ones need dirty buckets (phase 2).'; + END IF; + + FOREACH g IN ARRAY r.group_by LOOP + keys := keys || format(', %I', g); + keyeq := keyeq || format(' AND p.%I IS NOT DISTINCT FROM d.%I', g, g); + END LOOP; + rcols := ARRAY[r.ts_column]; + FOR i IN 1 .. cardinality(r.agg_functions) LOOP + fn := r.agg_functions[i]; + col := r.agg_columns[i]; + IF col = '*' THEN + dsel := dsel || ', count(*) AS count'; + icols := icols || ', count'; + isel := isel || ', d.count'; + uset := uset || ', count = p.count + d.count'; + CONTINUE; + END IF; + IF NOT col = ANY (rcols) THEN rcols := rcols || col; END IF; + CASE fn + WHEN 'count' THEN + c := 'count_' || col; + dsel := dsel || format(', count(%I) AS %I', col, c); + uset := uset || format(', %I = p.%I + d.%I', c, c, c); + WHEN 'sum' THEN + c := 'sum_' || col; + dsel := dsel || format(', sum(%I)::%s AS %I', col, + (SELECT format_type(a.atttypid, a.atttypmod) FROM pg_attribute a + WHERE a.attrelid = r.partials AND a.attname = c), c); + uset := uset || format(', %I = CASE WHEN p.%I IS NULL THEN d.%I WHEN d.%I IS NULL THEN p.%I ELSE p.%I + d.%I END', + c, c, c, c, c, c, c); + WHEN 'min', 'max' THEN + c := fn || '_' || col; + dsel := dsel || format(', %s(%I) AS %I', fn, col, c); + uset := uset || format(', %I = %s(p.%I, d.%I)', c, + CASE fn WHEN 'min' THEN 'least' ELSE 'greatest' END, c, c); + WHEN 'avg' THEN + c := 'avg_' || col || '__sum'; + dsel := dsel || format(', sum(%I)::%s AS %I, count(%I) AS %I', col, + (SELECT format_type(a.atttypid, a.atttypmod) FROM pg_attribute a + WHERE a.attrelid = r.partials AND a.attname = c), c, + col, 'avg_' || col || '__n'); + uset := uset || format(', %I = CASE WHEN p.%I IS NULL THEN d.%I WHEN d.%I IS NULL THEN p.%I ELSE p.%I + d.%I END', + c, c, c, c, c, c, c) + || format(', %I = p.%I + d.%I', 'avg_' || col || '__n', + 'avg_' || col || '__n', 'avg_' || col || '__n'); + icols := icols || format(', %I', 'avg_' || col || '__n'); + isel := isel || format(', d.%I', 'avg_' || col || '__n'); + END CASE; + icols := icols || format(', %I', c); + isel := isel || format(', d.%I', c); + END LOOP; + FOREACH g IN ARRAY r.group_by LOOP + IF NOT g = ANY (rcols) THEN rcols := rcols || g; END IF; + END LOOP; + -- each source column once: sum, min and max of v all read the same v + SELECT string_agg(format('%I', x), ', ' ORDER BY o) INTO rowcols + FROM unnest(rcols) WITH ORDINALITY u(x, o); + + -- ONE statement: the group list, the rows of the new groups (one TID range + -- per group, so only their row groups are read), the fold into the partials + -- and the record of the groups folded all see the same snapshot. A group + -- that commits while this runs is in none of them, and the next refresh + -- takes it. pgcolumnar.rollup_pause holds the statement open for that arm. + EXECUTE format($f$ + WITH cur AS ( + SELECT * FROM pgcolumnar.rollup_source_groups(%1$L::regclass) + ), held AS ( + SELECT pg_sleep(CASE WHEN current_setting('pgcolumnar.rollup_pause', true) = 'on' + THEN 2 ELSE 0 END) AS slept + ), new_groups AS ( + SELECT c.* FROM cur c + WHERE NOT EXISTS (SELECT 1 FROM pgcolumnar.rollup_group rg + WHERE rg.rollup = %2$L::regclass AND rg.group_number = c.group_number) + ), rows AS MATERIALIZED ( + SELECT s.* FROM new_groups ng, held, + LATERAL (SELECT %3$s FROM %1$s + WHERE ctid >= pgcolumnar.row_tid(ng.first_row_number) + AND ctid <= pgcolumnar.row_tid(ng.first_row_number + ng.row_count - 1) + -- OFFSET 0 keeps this subquery from being pulled up into a join: + -- PostgreSQL builds a TID range scan only from a relation's own + -- restrictions, and as a join clause the range read every group. + OFFSET 0) s + ), delta AS ( + SELECT date_bin(%4$L::interval, %5$I, %6$L::timestamptz) AS bucket%7$s%8$s + FROM rows GROUP BY 1%9$s + ), upd AS ( + UPDATE %10$s p SET %11$s + FROM delta d WHERE p.bucket IS NOT DISTINCT FROM d.bucket%12$s + RETURNING p.bucket%13$s + ), ins AS ( + INSERT INTO %10$s (bucket%7$s%14$s) + SELECT d.bucket%15$s%16$s FROM delta d + WHERE NOT EXISTS (SELECT 1 FROM upd p WHERE p.bucket IS NOT DISTINCT FROM d.bucket%12$s) + ), recorded AS ( + INSERT INTO pgcolumnar.rollup_group + SELECT %2$L::regclass, ng.group_number, ng.row_count, ng.deleted_count FROM new_groups ng + ) + SELECT count(*) FROM rows + $f$, + r.source, r.rollup, rowcols, r.width, r.ts_column, r.origin, + keys, dsel, + (SELECT coalesce(string_agg(format(', %s', n + 1), ''), '') FROM generate_series(1, cardinality(r.group_by)) n), + r.partials, substr(uset, 3), keyeq, + (SELECT coalesce(string_agg(format(', p.%I', x), ''), '') FROM unnest(r.group_by) x), + icols, + (SELECT coalesce(string_agg(format(', d.%I', x), ''), '') FROM unnest(r.group_by) x), + isel) + INTO folded; + RETURN folded; +END +$rr$; + +COMMENT ON FUNCTION pgcolumnar.refresh_rollup(regclass) + IS 'fold the rows of the row groups appended to a rollup''s source since its last refresh, in one statement; returns the rows folded (#1294, phase 1: appends only)'; + +CREATE FUNCTION pgcolumnar.rebase_rollup(rollup regclass) + RETURNS bigint + LANGUAGE plpgsql + SET search_path = pg_catalog, pg_temp +AS $rb$ +DECLARE + r pgcolumnar.rollup; + n bigint; +BEGIN + SELECT * INTO r FROM pgcolumnar.rollup x WHERE x.rollup = rebase_rollup.rollup; + IF NOT FOUND THEN + RAISE EXCEPTION '% is not a rollup', rollup USING ERRCODE = 'undefined_object'; + END IF; + EXECUTE format('LOCK TABLE %s IN SHARE UPDATE EXCLUSIVE MODE', r.partials); + -- The caller asserts the rollup is current: the source's groups are recorded + -- as folded, and the rollup's rows are not touched (#1294 point 9). + DELETE FROM pgcolumnar.rollup_group rg WHERE rg.rollup = r.rollup; + INSERT INTO pgcolumnar.rollup_group + SELECT r.rollup, s.group_number, s.row_count, s.deleted_count + FROM pgcolumnar.rollup_source_groups(r.source) s; + GET DIAGNOSTICS n = ROW_COUNT; + UPDATE pgcolumnar.rollup x SET source_storage = pgcolumnar.get_storage_id(r.source) + WHERE x.rollup = r.rollup; + RETURN n; +END +$rb$; + +COMMENT ON FUNCTION pgcolumnar.rebase_rollup(regclass) + IS 'record a rollup''s source row groups as folded without touching the rollup, after the source was rewritten; the caller asserts the rollup is current (#1294)'; diff --git a/pgcolumnar--1.0-alpha5.sql b/pgcolumnar--1.0-alpha5.sql index 893cd1ae..0df96cd1 100644 --- a/pgcolumnar--1.0-alpha5.sql +++ b/pgcolumnar--1.0-alpha5.sql @@ -2173,3 +2173,433 @@ BEGIN RETURN made; END; $create_ahead$; + +/* --------------------------------------------------------------------------- + * Incremental rollups, phase 1: append-only (#1294). + * + * A rollup is a VIEW over a partials table. The partials table holds, per + * (bucket, group_by...) key, the parts each aggregate combines from: count, sum, + * min, max, and for avg a sum and a count. The view turns those into the + * aggregates. refresh_rollup folds the rows of row groups that are new since the + * last refresh, in ONE statement, so the group list it reads, the rows it folds + * and the groups it records all come from one snapshot (#1294 point 1). + * + * pgcolumnar.rollup and pgcolumnar.rollup_group are readable and writable by + * anyone, row-level security limiting each role to the rollups it owns: the + * refresh runs as its caller and writes them in that statement. + * ------------------------------------------------------------------------- */ +CREATE TABLE pgcolumnar.rollup ( + rollup regclass PRIMARY KEY, -- the view + partials regclass NOT NULL, -- the table behind it + source regclass NOT NULL, + source_storage bigint NOT NULL, -- the source's storage id when recorded + ts_column name NOT NULL, + width interval NOT NULL, + origin timestamptz NOT NULL, + group_by name[] NOT NULL, + agg_functions text[] NOT NULL, + agg_columns name[] NOT NULL +); + +CREATE TABLE pgcolumnar.rollup_group ( + rollup regclass NOT NULL, + group_number bigint NOT NULL, + row_count bigint NOT NULL, + deleted_count bigint NOT NULL, + PRIMARY KEY (rollup, group_number) +); + +ALTER TABLE pgcolumnar.rollup ENABLE ROW LEVEL SECURITY; +ALTER TABLE pgcolumnar.rollup_group ENABLE ROW LEVEL SECURITY; +CREATE POLICY rollup_owner ON pgcolumnar.rollup + USING (pg_has_role(current_user, + (SELECT c.relowner FROM pg_class c WHERE c.oid = rollup), 'USAGE')); +CREATE POLICY rollup_group_owner ON pgcolumnar.rollup_group + USING (pg_has_role(current_user, + (SELECT c.relowner FROM pg_class c WHERE c.oid = rollup), 'USAGE')); +GRANT SELECT, INSERT, UPDATE, DELETE ON pgcolumnar.rollup, pgcolumnar.rollup_group TO PUBLIC; + +/* + * The source's row groups, with each one's deleted count. SECURITY DEFINER + * because row_group and delete_vector carry no GRANT; the caller must be able to + * SELECT the source. STABLE, so it reads with the calling statement's snapshot + * and refresh_rollup keeps to one snapshot. + */ +-- The item pointer the access method gives a row number (#1319 review). +CREATE FUNCTION pgcolumnar.row_tid(row_number bigint) + RETURNS tid + LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE + AS 'MODULE_PATHNAME', 'pgcolumnar_row_tid'; + +CREATE FUNCTION pgcolumnar.rollup_source_groups(source regclass) + RETURNS TABLE (group_number bigint, first_row_number bigint, row_count bigint, + deleted_count bigint) + LANGUAGE plpgsql STABLE SECURITY DEFINER + SET search_path = pg_catalog, pg_temp +AS $rsg$ +BEGIN + PERFORM pgcolumnar.require_caller_select(source); + RETURN QUERY + SELECT g.group_number, g.first_row_number, g.row_count, + coalesce(d.deleted_count, 0)::bigint + FROM pgcolumnar.row_group g + LEFT JOIN pgcolumnar.delete_vector d + ON d.storage_id = g.storage_id AND d.group_number = g.group_number + WHERE g.storage_id = pgcolumnar.get_storage_id(source); +END +$rsg$; + +/* + * The source's row groups that retention retired (pgcolumnar.retired_group, + * written only by expire, #1314). SECURITY DEFINER because that catalog carries + * no GRANT; the caller must be able to SELECT the source. + */ +CREATE FUNCTION pgcolumnar.rollup_source_retired(source regclass) + RETURNS TABLE (group_number bigint) + LANGUAGE plpgsql STABLE SECURITY DEFINER + SET search_path = pg_catalog, pg_temp +AS $rsr$ +BEGIN + PERFORM pgcolumnar.require_caller_select(source); + RETURN QUERY + SELECT r.group_number FROM pgcolumnar.retired_group r + WHERE r.storage_id = pgcolumnar.get_storage_id(source); +END +$rsr$; + +CREATE FUNCTION pgcolumnar.create_rollup(rollup name, source regclass, ts_column name, + width interval, origin timestamptz, + group_by name[], agg_functions text[], + agg_columns name[], + access_method name DEFAULT 'pgcolumnar') + RETURNS regclass + LANGUAGE plpgsql + SET search_path = pg_catalog, pg_temp +AS $cr$ +DECLARE + nsp name; + tstype regtype; + cols text := ''; + vcols text := ''; + gcols text := ''; + i int; + fn text; + col name; + ctype regtype; + ptype text; + partials text; + view_oid regclass; +BEGIN + IF width IS NULL OR width <= interval '0' OR extract(month FROM width) <> 0 + OR extract(year FROM width) <> 0 THEN + RAISE EXCEPTION 'a rollup width must be positive and have no months or years' + USING ERRCODE = 'invalid_parameter_value', + DETAIL = 'date_bin, which buckets the rows, takes no width in months.'; + END IF; + IF agg_functions IS NULL OR agg_columns IS NULL + OR cardinality(agg_functions) = 0 + OR cardinality(agg_functions) <> cardinality(agg_columns) THEN + RAISE EXCEPTION 'agg_functions and agg_columns must be non-empty and of the same length' + USING ERRCODE = 'invalid_parameter_value'; + END IF; + SELECT a.atttypid::regtype INTO tstype FROM pg_attribute a + WHERE a.attrelid = source AND a.attname = ts_column AND NOT a.attisdropped; + IF tstype IS DISTINCT FROM 'timestamptz'::regtype THEN + RAISE EXCEPTION 'column "%" of % is not a timestamptz column', ts_column, source + USING ERRCODE = 'invalid_parameter_value'; + END IF; + + -- Each aggregate checked against the five that combine across partial + -- results, and each column against the source: structured input, never SQL + -- text (#1294 point 11). + FOR i IN 1 .. cardinality(agg_functions) LOOP + fn := agg_functions[i]; + col := agg_columns[i]; + IF fn NOT IN ('count', 'sum', 'min', 'max', 'avg') THEN + RAISE EXCEPTION '"%" is not an aggregate a rollup can combine', fn + USING ERRCODE = 'invalid_parameter_value', + HINT = 'Use count, sum, min, max or avg.'; + END IF; + IF col = '*' THEN + IF fn <> 'count' THEN + RAISE EXCEPTION '* goes only with count, not %', fn + USING ERRCODE = 'invalid_parameter_value'; + END IF; + cols := cols || ', count bigint NOT NULL'; + vcols := vcols || ', count'; + CONTINUE; + END IF; + SELECT a.atttypid::regtype INTO ctype FROM pg_attribute a + WHERE a.attrelid = source AND a.attname = col AND a.attnum > 0 AND NOT a.attisdropped; + IF ctype IS NULL THEN + RAISE EXCEPTION 'column "%" does not exist in %', col, source + USING ERRCODE = 'invalid_parameter_value'; + END IF; + ptype := CASE WHEN ctype IN ('int2'::regtype, 'int4'::regtype, 'int8'::regtype, + 'numeric'::regtype) THEN 'numeric' + WHEN ctype IN ('float4'::regtype, 'float8'::regtype) THEN 'float8' + ELSE NULL END; + IF fn IN ('sum', 'avg') AND ptype IS NULL THEN + RAISE EXCEPTION '% of a % column cannot be folded', fn, ctype + USING ERRCODE = 'invalid_parameter_value'; + END IF; + CASE fn + WHEN 'count' THEN + cols := cols || format(', %I bigint NOT NULL', 'count_' || col); + vcols := vcols || format(', %I', 'count_' || col); + WHEN 'sum' THEN + cols := cols || format(', %I %s', 'sum_' || col, ptype); + vcols := vcols || format(', %I', 'sum_' || col); + WHEN 'min', 'max' THEN + cols := cols || format(', %I %s', fn || '_' || col, ctype); + vcols := vcols || format(', %I', fn || '_' || col); + WHEN 'avg' THEN + cols := cols || format(', %I %s, %I bigint NOT NULL', 'avg_' || col || '__sum', ptype, + 'avg_' || col || '__n'); + vcols := vcols || format(', (%I / nullif(%I, 0))::%s AS %I', 'avg_' || col || '__sum', + 'avg_' || col || '__n', ptype, 'avg_' || col); + END CASE; + END LOOP; + + IF group_by IS NOT NULL THEN + FOREACH col IN ARRAY group_by LOOP + SELECT a.atttypid::regtype INTO ctype FROM pg_attribute a + WHERE a.attrelid = source AND a.attname = col AND a.attnum > 0 AND NOT a.attisdropped; + IF ctype IS NULL THEN + RAISE EXCEPTION 'group_by column "%" does not exist in %', col, source + USING ERRCODE = 'invalid_parameter_value'; + END IF; + gcols := gcols || format(', %I %s', col, ctype); + END LOOP; + END IF; + + -- In the source's schema: this function's search_path is pg_catalog, so + -- current_schema() would name the system schema, not the caller's. + SELECT n.nspname INTO nsp FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.oid = source; + partials := rollup || '__partials'; + EXECUTE format('CREATE TABLE %I.%I (bucket timestamptz%s%s) USING %I', + nsp, partials, gcols, cols, access_method); + EXECUTE format('CREATE VIEW %I.%I AS SELECT bucket%s%s FROM %I.%I', + nsp, rollup, + (SELECT coalesce(string_agg(format(', %I', g), ''), '') FROM unnest(group_by) g), + vcols, nsp, partials); + view_oid := format('%I.%I', nsp, rollup)::regclass; + INSERT INTO pgcolumnar.rollup VALUES + (view_oid, format('%I.%I', nsp, partials)::regclass, source, + pgcolumnar.get_storage_id(source), ts_column, width, origin, + coalesce(group_by, '{}'), agg_functions, agg_columns); + RETURN view_oid; +END +$cr$; + +COMMENT ON FUNCTION pgcolumnar.create_rollup(name, regclass, name, interval, timestamptz, name[], text[], name[], name) + IS 'create an incremental rollup: a view of date_bin(width, ts_column, origin) buckets per group_by key, with count, sum, min, max or avg per (agg_functions, agg_columns); fill it with refresh_rollup (#1294)'; + +/* + * The fold, one statement for one snapshot. Built per rollup from its catalog + * row: every identifier through %I and every value through %L. + */ +CREATE FUNCTION pgcolumnar.refresh_rollup(rollup regclass) + RETURNS bigint + LANGUAGE plpgsql + SET search_path = pg_catalog, pg_temp +AS $rr$ +DECLARE + r pgcolumnar.rollup; + changed text; + keys text := ''; + keyeq text := ''; + rowcols text; + rcols name[]; + dsel text := ''; + icols text := ''; + isel text := ''; + uset text := ''; + i int; + fn text; + col name; + c text; + g name; + folded bigint; +BEGIN + -- A REPEATABLE READ caller's snapshot predates the lock below, so a refresh + -- that waited for another would fold the same groups again. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'refresh_rollup runs only at READ COMMITTED, not %', + upper(current_setting('transaction_isolation')) + USING ERRCODE = 'invalid_transaction_state'; + END IF; + SELECT * INTO r FROM pgcolumnar.rollup x WHERE x.rollup = refresh_rollup.rollup; + IF NOT FOUND THEN + RAISE EXCEPTION '% is not a rollup', rollup USING ERRCODE = 'undefined_object'; + END IF; + -- Self-conflicting, so two refreshes run one after the other; readers pass. + EXECUTE format('LOCK TABLE %s IN SHARE UPDATE EXCLUSIVE MODE', r.partials); + + IF pgcolumnar.get_storage_id(r.source) <> r.source_storage THEN + RAISE EXCEPTION 'the source of rollup % was rewritten since its last refresh', rollup + USING ERRCODE = 'object_not_in_prerequisite_state', + HINT = 'If the rollup is current, pgcolumnar.rebase_rollup records the source''s new row groups.'; + END IF; + -- Retention keeps history (the owner's decision on #1294): a recorded group + -- that expire retired leaves the snapshot, and its buckets keep the rows they + -- folded. Retirement is permanent, so this need not share the fold's snapshot. + DELETE FROM pgcolumnar.rollup_group rg + WHERE rg.rollup = r.rollup + AND rg.group_number IN (SELECT t.group_number FROM pgcolumnar.rollup_source_retired(r.source) t); + -- Phase 1 folds appends only: any other recorded group that is gone, or has + -- new deletions, needs phase 2's dirty buckets. + SELECT string_agg(rg.group_number::text, ', ' ORDER BY rg.group_number) INTO changed + FROM pgcolumnar.rollup_group rg + LEFT JOIN pgcolumnar.rollup_source_groups(r.source) s USING (group_number) + WHERE rg.rollup = r.rollup + AND (s.group_number IS NULL OR s.deleted_count <> rg.deleted_count); + IF changed IS NOT NULL THEN + RAISE EXCEPTION 'rollup %: row groups % of its source were deleted from or rewritten', rollup, changed + USING ERRCODE = 'feature_not_supported', + DETAIL = 'This version folds appended row groups only; changed ones need dirty buckets (phase 2).'; + END IF; + + FOREACH g IN ARRAY r.group_by LOOP + keys := keys || format(', %I', g); + keyeq := keyeq || format(' AND p.%I IS NOT DISTINCT FROM d.%I', g, g); + END LOOP; + rcols := ARRAY[r.ts_column]; + FOR i IN 1 .. cardinality(r.agg_functions) LOOP + fn := r.agg_functions[i]; + col := r.agg_columns[i]; + IF col = '*' THEN + dsel := dsel || ', count(*) AS count'; + icols := icols || ', count'; + isel := isel || ', d.count'; + uset := uset || ', count = p.count + d.count'; + CONTINUE; + END IF; + IF NOT col = ANY (rcols) THEN rcols := rcols || col; END IF; + CASE fn + WHEN 'count' THEN + c := 'count_' || col; + dsel := dsel || format(', count(%I) AS %I', col, c); + uset := uset || format(', %I = p.%I + d.%I', c, c, c); + WHEN 'sum' THEN + c := 'sum_' || col; + dsel := dsel || format(', sum(%I)::%s AS %I', col, + (SELECT format_type(a.atttypid, a.atttypmod) FROM pg_attribute a + WHERE a.attrelid = r.partials AND a.attname = c), c); + uset := uset || format(', %I = CASE WHEN p.%I IS NULL THEN d.%I WHEN d.%I IS NULL THEN p.%I ELSE p.%I + d.%I END', + c, c, c, c, c, c, c); + WHEN 'min', 'max' THEN + c := fn || '_' || col; + dsel := dsel || format(', %s(%I) AS %I', fn, col, c); + uset := uset || format(', %I = %s(p.%I, d.%I)', c, + CASE fn WHEN 'min' THEN 'least' ELSE 'greatest' END, c, c); + WHEN 'avg' THEN + c := 'avg_' || col || '__sum'; + dsel := dsel || format(', sum(%I)::%s AS %I, count(%I) AS %I', col, + (SELECT format_type(a.atttypid, a.atttypmod) FROM pg_attribute a + WHERE a.attrelid = r.partials AND a.attname = c), c, + col, 'avg_' || col || '__n'); + uset := uset || format(', %I = CASE WHEN p.%I IS NULL THEN d.%I WHEN d.%I IS NULL THEN p.%I ELSE p.%I + d.%I END', + c, c, c, c, c, c, c) + || format(', %I = p.%I + d.%I', 'avg_' || col || '__n', + 'avg_' || col || '__n', 'avg_' || col || '__n'); + icols := icols || format(', %I', 'avg_' || col || '__n'); + isel := isel || format(', d.%I', 'avg_' || col || '__n'); + END CASE; + icols := icols || format(', %I', c); + isel := isel || format(', d.%I', c); + END LOOP; + FOREACH g IN ARRAY r.group_by LOOP + IF NOT g = ANY (rcols) THEN rcols := rcols || g; END IF; + END LOOP; + -- each source column once: sum, min and max of v all read the same v + SELECT string_agg(format('%I', x), ', ' ORDER BY o) INTO rowcols + FROM unnest(rcols) WITH ORDINALITY u(x, o); + + -- ONE statement: the group list, the rows of the new groups (one TID range + -- per group, so only their row groups are read), the fold into the partials + -- and the record of the groups folded all see the same snapshot. A group + -- that commits while this runs is in none of them, and the next refresh + -- takes it. pgcolumnar.rollup_pause holds the statement open for that arm. + EXECUTE format($f$ + WITH cur AS ( + SELECT * FROM pgcolumnar.rollup_source_groups(%1$L::regclass) + ), held AS ( + SELECT pg_sleep(CASE WHEN current_setting('pgcolumnar.rollup_pause', true) = 'on' + THEN 2 ELSE 0 END) AS slept + ), new_groups AS ( + SELECT c.* FROM cur c + WHERE NOT EXISTS (SELECT 1 FROM pgcolumnar.rollup_group rg + WHERE rg.rollup = %2$L::regclass AND rg.group_number = c.group_number) + ), rows AS MATERIALIZED ( + SELECT s.* FROM new_groups ng, held, + LATERAL (SELECT %3$s FROM %1$s + WHERE ctid >= pgcolumnar.row_tid(ng.first_row_number) + AND ctid <= pgcolumnar.row_tid(ng.first_row_number + ng.row_count - 1) + -- OFFSET 0 keeps this subquery from being pulled up into a join: + -- PostgreSQL builds a TID range scan only from a relation's own + -- restrictions, and as a join clause the range read every group. + OFFSET 0) s + ), delta AS ( + SELECT date_bin(%4$L::interval, %5$I, %6$L::timestamptz) AS bucket%7$s%8$s + FROM rows GROUP BY 1%9$s + ), upd AS ( + UPDATE %10$s p SET %11$s + FROM delta d WHERE p.bucket IS NOT DISTINCT FROM d.bucket%12$s + RETURNING p.bucket%13$s + ), ins AS ( + INSERT INTO %10$s (bucket%7$s%14$s) + SELECT d.bucket%15$s%16$s FROM delta d + WHERE NOT EXISTS (SELECT 1 FROM upd p WHERE p.bucket IS NOT DISTINCT FROM d.bucket%12$s) + ), recorded AS ( + INSERT INTO pgcolumnar.rollup_group + SELECT %2$L::regclass, ng.group_number, ng.row_count, ng.deleted_count FROM new_groups ng + ) + SELECT count(*) FROM rows + $f$, + r.source, r.rollup, rowcols, r.width, r.ts_column, r.origin, + keys, dsel, + (SELECT coalesce(string_agg(format(', %s', n + 1), ''), '') FROM generate_series(1, cardinality(r.group_by)) n), + r.partials, substr(uset, 3), keyeq, + (SELECT coalesce(string_agg(format(', p.%I', x), ''), '') FROM unnest(r.group_by) x), + icols, + (SELECT coalesce(string_agg(format(', d.%I', x), ''), '') FROM unnest(r.group_by) x), + isel) + INTO folded; + RETURN folded; +END +$rr$; + +COMMENT ON FUNCTION pgcolumnar.refresh_rollup(regclass) + IS 'fold the rows of the row groups appended to a rollup''s source since its last refresh, in one statement; returns the rows folded (#1294, phase 1: appends only)'; + +CREATE FUNCTION pgcolumnar.rebase_rollup(rollup regclass) + RETURNS bigint + LANGUAGE plpgsql + SET search_path = pg_catalog, pg_temp +AS $rb$ +DECLARE + r pgcolumnar.rollup; + n bigint; +BEGIN + SELECT * INTO r FROM pgcolumnar.rollup x WHERE x.rollup = rebase_rollup.rollup; + IF NOT FOUND THEN + RAISE EXCEPTION '% is not a rollup', rollup USING ERRCODE = 'undefined_object'; + END IF; + EXECUTE format('LOCK TABLE %s IN SHARE UPDATE EXCLUSIVE MODE', r.partials); + -- The caller asserts the rollup is current: the source's groups are recorded + -- as folded, and the rollup's rows are not touched (#1294 point 9). + DELETE FROM pgcolumnar.rollup_group rg WHERE rg.rollup = r.rollup; + INSERT INTO pgcolumnar.rollup_group + SELECT r.rollup, s.group_number, s.row_count, s.deleted_count + FROM pgcolumnar.rollup_source_groups(r.source) s; + GET DIAGNOSTICS n = ROW_COUNT; + UPDATE pgcolumnar.rollup x SET source_storage = pgcolumnar.get_storage_id(r.source) + WHERE x.rollup = r.rollup; + RETURN n; +END +$rb$; + +COMMENT ON FUNCTION pgcolumnar.rebase_rollup(regclass) + IS 'record a rollup''s source row groups as folded without touching the rollup, after the source was rewritten; the caller asserts the rollup is current (#1294)'; diff --git a/src/columnar_storage.c b/src/columnar_storage.c index 6f140656..8549f70b 100644 --- a/src/columnar_storage.c +++ b/src/columnar_storage.c @@ -668,6 +668,30 @@ PgColumnarRowNumberToItemPointer(uint64 rowNumber, ItemPointer tid) ItemPointerSet(tid, blockno, offset); } +/* + * pgcolumnar.row_tid(row_number bigint) -> tid + * The item pointer the access method gives a row number, for SQL that + * needs a row group's TID range (refresh_rollup, #1294). The mapping lives + * here once: restated in SQL as 291 rows a block it held only at the + * default block size (#1319 review). + */ +PG_FUNCTION_INFO_V1(pgcolumnar_row_tid); +Datum +pgcolumnar_row_tid(PG_FUNCTION_ARGS) +{ + int64 rowNumber = PG_GETARG_INT64(0); + ItemPointer tid; + + if (rowNumber < 1) + ereport(ERROR, + (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE), + errmsg("row number " INT64_FORMAT " is out of range", rowNumber), + errdetail("Row numbers start at 1."))); + tid = (ItemPointer) palloc(sizeof(ItemPointerData)); + PgColumnarRowNumberToItemPointer((uint64) rowNumber, tid); + PG_RETURN_POINTER(tid); /* PG_RETURN_ITEMPOINTER is 16+; tid is by reference */ +} + uint64 PgColumnarItemPointerToRowNumber(ItemPointer tid) { diff --git a/src/columnar_tableam.c b/src/columnar_tableam.c index c1e34006..d72f71e7 100644 --- a/src/columnar_tableam.c +++ b/src/columnar_tableam.c @@ -3663,6 +3663,9 @@ pgcolumnar_get_relation_info(PlannerInfo *root, Oid relationObjectId, * module init * ------------------------------------------------------------------------- */ +/* pgcolumnar.rollup_pause (#1294): read by refresh_rollup through current_setting */ +static bool pgcolumnar_rollup_pause = false; + void _PG_init(void) { @@ -4092,6 +4095,21 @@ _PG_init(void) * time via GetConfigOption; link-local ranges are refused there * unconditionally, list or no list. */ + /* + * A developer pause for pgcolumnar.refresh_rollup (#1294): with it on, the + * refresh sleeps two seconds inside its one statement, after its snapshot + * is taken, so a test can commit rows in that window and show they are + * folded by the next refresh, once. Superuser-only and off by default. + */ + DefineCustomBoolVariable("pgcolumnar.rollup_pause", + "Developer pause: hold refresh_rollup open inside its statement.", + NULL, + &pgcolumnar_rollup_pause, + false, + PGC_SUSET, + GUC_NOT_IN_SAMPLE, + NULL, NULL, NULL); + DefineCustomStringVariable("pgcolumnar.objstore_allowed_endpoints", "Remote endpoints the object-store module may connect to.", "Comma-separated host or host:port entries; empty refuses all remote access.", diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index 7e8ab9ba..7ad4c8dc 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -1929,6 +1929,62 @@ rewrite_storage_oid rewrite_storage_oid the repair gate does not read a partitio rewrite_storage_oid rewrite_storage_oid the repair gate does not read a partition the ALTER did not touch: lp ATTACH PARTITION lp_3 FOR VALUES FROM ('2026-03-01') TO ('2026-04-01') 15;16;17;18;19 2026-10-12 the repair gate opens every inheritor with NoLock again, so it reads the metapage of a partition the ALTER never locked (an Assert on assert builds) rewrite_storage_oid rewrite_storage_oid the repair gate does not read a partition the ALTER did not touch: lp DETACH PARTITION lp_2 15;16;17;18;19 2026-10-12 the repair gate opens every inheritor with NoLock again, so it reads the metapage of a partition the ALTER never locked (an Assert on assert builds) rewrite_storage_oid rewrite_storage_oid the repair gate does not read a partition the ALTER did not touch: lp DETACH PARTITION lp_2 CONCURRENTLY 15;16;17;18;19 2026-10-12 the repair gate opens every inheritor with NoLock again, so it reads the metapage of a partition the ALTER never locked (an Assert on assert builds) +rollup rollup * with anything but count is refused 15;16;17;18;19 2026-10-13 create_rollup allowing * with any function +rollup rollup a refresh after a delete is refused until phase 2 15;16;17;18;19 2026-10-13 refresh_rollup without the changed-group refusal +rollup rollup a refresh after a small append reads under a fifth of the source 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup's per-group subquery without OFFSET 0, so it becomes a join and reads every group +rollup rollup a refresh after appends folds exactly the appended rows, late ones included 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup a refresh after retention folds nothing, refuses nothing, and keeps every bucket 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup not dropping groups retention retired +rollup rollup a refresh after the source was rewritten is refused, naming rebase_rollup 15;16;17;18;19 2026-10-13 refresh_rollup without the source identity check +rollup rollup a refresh at REPEATABLE READ is refused 15;16;17;18;19 2026-10-13 refresh_rollup at any isolation level +rollup rollup a refresh folds a bulk append 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup a refresh with nothing new folds nothing 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group +rollup rollup a role that cannot read the source cannot create a rollup of it 15;16;17;18;19 2026-10-13 get_storage_id without its SELECT check, so create_rollup no longer refuses a caller that cannot read the source +rollup rollup a rollup kept in a heap table refreshes and equals a full recompute 15;16;17;18;19 2026-10-13 create_rollup ignoring access_method;refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup a rollup with NULL keys and NULL timestamps equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup matching keys with = instead of IS NOT DISTINCT FROM;row_tid mapping 290 rows a block instead of the AM's +rollup rollup a table's owner, not a superuser, creates a rollup of it 15;16;17;18;19 2026-10-13 the rollup catalogs without their grants +rollup rollup a width with months is refused 15;16;17;18;19 2026-10-13 create_rollup without the months refusal +rollup rollup after a rebase the next refresh folds exactly the appended rows 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;row_tid mapping 290 rows a block instead of the AM's +rollup rollup aggregate functions and columns of different lengths are refused 15;16;17;18;19 2026-10-13 create_rollup without the equal-length check +rollup rollup an aggregate column the source does not have is refused 15;16;17;18;19 2026-10-13 create_rollup without the column-exists check +rollup rollup an aggregate that does not combine is refused 15;16;17;18;19 2026-10-13 create_rollup without the five-function allow-list +rollup rollup and after retention an append still folds 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup not dropping groups retention retired;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and after the bulk append the rollup equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup recording the folded groups from a fresh read after the fold;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and after the first refresh the rollup equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup folding only the first new group +rollup rollup and after the rebase the rollup equals a full recompute 15;16;17;18;19 2026-10-13 rebase_rollup clearing the rollup's rows;refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup recording the folded groups from a fresh read after the fold;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and after the refresh that ran beside an insert the rollup equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup recording the folded groups from a fresh read after the fold;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and after the small append the rollup equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup recording the folded groups from a fresh read after the fold;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and after two refreshes started together the rollup equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup recording the folded groups from a fresh read after the fold;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and cannot refresh it 15;16;17;18;19 2026-10-13 the rollup catalogs without row-level security;the rollup catalogs without their grants +rollup rollup and leaves the rollup as it was 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group +rollup rollup and leaves the rollup's rows alone 15;16;17;18;19 2026-10-13 rebase_rollup clearing the rollup's rows +rollup rollup and no refused rollup left a table behind 15;16;17;18;19 2026-10-13 create_rollup allowing * with any function;create_rollup without the equal-length check;create_rollup without the months refusal +rollup rollup and reads what a recompute gives 15;16;17;18;19 2026-10-13 refresh_rollup bucketing with date_trunc, which follows the session's zone;rollup_source_groups without SECURITY DEFINER;the rollup catalogs without their grants +rollup rollup and refreshes it 15;16;17;18;19 2026-10-13 rollup_source_groups without SECURITY DEFINER;the rollup catalogs without their grants +rollup rollup and that is not zero 15;16;17;18;19 never - +rollup rollup and the recent buckets equal a recompute, while the expired ones keep their history 15;16;17;18;19 2026-10-13 refresh_rollup bucketing with date_trunc, which follows the session's zone;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup not dropping groups retention retired;row_tid mapping 290 rows a block instead of the AM's +rollup rollup and the refused refresh changed nothing 15;16;17;18;19 never - +rollup rollup and the rollup still equals a full recompute 15;16;17;18;19 2026-10-13 refresh_rollup combining min with greatest;refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup another role sees none of its rollup catalog rows 15;16;17;18;19 2026-10-13 the rollup catalogs without row-level security;the rollup catalogs without their grants +rollup rollup each row group's TID range holds exactly that group's rows 15;16;17;18;19 2026-10-13 row_tid mapping 290 rows a block instead of the AM's +rollup rollup premise: and the fixture has NULL-key and NULL-bucket groups 15;16;17;18;19 never - +rollup rollup premise: expire retired the old row groups 15;16;17;18;19 never - +rollup rollup premise: pgcolumnar.vacuum gave the source's row groups new identities 15;16;17;18;19 never - +rollup rollup premise: the rollup functions exist 15;16;17;18;19 never - +rollup rollup premise: the source holds every row 15;16;17;18;19 never - +rollup rollup rebase_rollup records the source's current row groups 15;16;17;18;19 2026-10-13 rebase_rollup returning 0 +rollup rollup refreshes from sessions in different time zones build the same rollup 15;16;17;18;19 2026-10-13 refresh_rollup bucketing with date_trunc, which follows the session's zone +rollup rollup the first refresh folds every row 15;16;17;18;19 2026-10-13 refresh_rollup folding only the first new group +rollup rollup the history rollup folded every row 15;16;17;18;19 2026-10-13 refresh_rollup folding only the first new group +rollup rollup the history rollup holds old buckets 15;16;17;18;19 never - +rollup rollup the next refresh folds the rows committed during the last one, once 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup recording the folded groups from a fresh read after the fold;row_tid mapping 290 rows a block instead of the AM's +rollup rollup the paused refresh folded what was there before it began 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup the refresh after a small append read the source at all 15;16;17;18;19 never - +rollup rollup the rollup holds rows before the delete 15;16;17;18;19 never - +rollup rollup the rollup holds rows before the rewrite 15;16;17;18;19 never - +rollup rollup the rollup table exists with the documented columns 15;16;17;18;19 never - +rollup rollup the second refresh of the NULL-key rollup folded the append 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup the zone rollups hold a row per key of the recompute 15;16;17;18;19 2026-10-13 refresh_rollup folding only the first new group;row_tid mapping 290 rows a block instead of the AM's +rollup rollup two refreshes started together fold the new rows once between them 15;16;17;18;19 2026-10-13 refresh_rollup folding every group, ignoring the recorded snapshot;refresh_rollup folding only the first new group;refresh_rollup without the self-conflicting lock;row_tid mapping 290 rows a block instead of the AM's storage_row_repair storage_row_repair REFRESH MATERIALIZED VIEW repairs the view's row 15;16;17;18;19 2026-10-12 PgColumnarRetargetStorageRelation returns at once, so no repair repoints a storage row storage_row_repair storage_row_repair a same-type ALTER COLUMN ... USING repairs the table's row 15;16;17;18;19 2026-10-12 PgColumnarRetargetStorageRelation returns at once, so no repair repoints a storage row storage_row_repair storage_row_repair after both repairs the notes' query finds nothing 15;16;17;18;19 2026-10-12 PgColumnarRetargetStorageRelation returns at once, so no repair repoints a storage row;the release notes' query is edited from NOT EXISTS to EXISTS diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index 2c77ef3e..24265816 100644 --- a/test/check_ledger_budget.txt +++ b/test/check_ledger_budget.txt @@ -1410,4 +1410,22 @@ suites_not_covered 247 # never 1683 + 3 = 1686 # dated 301 + 16 = 317 # 1686 + 317 == 2003 -checks_never_observed_red 1686 +# +# #1294 phase 1, rollups, PR #1319 (OffgridwithJD): 1686 -> 1698. rollup enters +# the ledger with 56 checks from a five-major run on this tree. Forty-four are +# dated by twenty-five mutations (the rollup functions, row_tid, and +# get_storage_id's SELECT check, which is what refuses create_rollup), each merged +# one log at a time with --mutation and --target; no premise is among them (ten +# checks that read the code's output were renamed from premises first, #1236). +# Twelve arrive `never`: five premises (the functions exist, the source's rows, +# vacuum's renumbering, the NULL fixture, expire's retirement) and seven arms -- +# 'and the refused refresh changed nothing' holds because a refused refresh +# rolls back; the rest read the rollup or the refresh in ways no mutant of this +# change reaches (the documented columns, old buckets present, a read at all, +# rows present before the delete and the rewrite, a non-empty zone rollup). +# +# ledger rows 2003 + 56 = 2059 +# never 1686 + 12 = 1698 +# dated 317 + 44 = 361 +# 1698 + 361 == 2059 +checks_never_observed_red 1698 diff --git a/test/entry_point_privilege.sh b/test/entry_point_privilege.sh index 7c89c58e..4342e4d6 100755 --- a/test/entry_point_privilege.sh +++ b/test/entry_point_privilege.sh @@ -64,13 +64,16 @@ SQLFILE="$_epp_root/pgcolumnar--$_epp_ver.sql" # iceberg_rest_scan reads a REST-cataloged table's FILES, server-file role (#388 phase 7) # iceberg_rest_namespaces / iceberg_rest_tables list a REST catalog, server-file role (#388 phase 7) # read_manifest_list reads an Avro FILE, gated by the server-file role (#388) +# row_tid maps a row number to a TID by arithmetic (#1294): it +# takes no relation and reads nothing. A version that +# took a relation would need the SELECT check the others have. # # Kept on ONE line and matched after whitespace normalisation. A multi-line list # here is matched by `case " $EXEMPT " in *" $fn "*`, and the name sitting at a # line end is followed by a newline rather than a space, so it silently fails to # match and reads as an unbucketed function. That is not hypothetical: it is what # the first run of this suite reported for parquet_fdw_validator. -EXEMPT="columnar_handler parquet_fdw_handler parquet_fdw_validator iceberg_fdw_handler iceberg_fdw_validator iceberg_catalog_fdw_validator parquet_schema read_parquet file_split_offsets iceberg_current_snapshot iceberg_data_files iceberg_scan iceberg_rest_table_location iceberg_rest_scan iceberg_rest_namespaces iceberg_rest_tables read_avro_manifest read_manifest_list" +EXEMPT="columnar_handler parquet_fdw_handler parquet_fdw_validator iceberg_fdw_handler iceberg_fdw_validator iceberg_catalog_fdw_validator parquet_schema read_parquet file_split_offsets iceberg_current_snapshot iceberg_data_files iceberg_scan iceberg_rest_table_location iceberg_rest_scan iceberg_rest_namespaces iceberg_rest_tables read_avro_manifest read_manifest_list row_tid" EXEMPT="$(echo $EXEMPT)" # Relation-taking entry points with NO privilege check of any kind on main. @@ -106,9 +109,16 @@ KNOWN_UNGUARDED="" # whatever the per-table path it calls enforces, not a bar of its own. It # also carries the same phantom `stripe_count` argument as # pgcolumnar.vacuum (#560). +# create_rollup(rollup name, source regclass, ...) +# plpgsql; its relation is the SECOND argument (#1294). It runs as the +# caller and reads the source's storage id through get_storage_id, which +# is guarded (below), so a caller that cannot SELECT the source is refused +# with 42501 at creation. That is asserted in rollup.sh and +# test_rollup.py, not here. # -# Both are recorded here rather than resolved. Neither is claimed guarded. -RELATION_BY_OTHER_MEANS="alter_table_set_access_method vacuum_full" +# The first two are recorded here rather than resolved; neither is claimed +# guarded. +RELATION_BY_OTHER_MEANS="alter_table_set_access_method vacuum_full create_rollup" RELATION_BY_OTHER_MEANS="$(echo $RELATION_BY_OTHER_MEANS)" # Relation-taking entry points that are NOT C functions. Their bar is not a @@ -205,7 +215,7 @@ n_all="$(printf '%s\n' $allfns | grep -c .)" check_num "coverage: inputs == sum(buckets)" \ "$n_all" "$(( exempt_seen + rel_seen + other_seen + $(printf '%s\n' $unbucketed | grep -c .) ))" -check_num "the relation-by-other-means class is the two that are known" "$other_seen" "2" +check_num "the relation-by-other-means class is the three that are known" "$other_seen" "3" check "coverage: no entry point is unbucketed" \ "$([ -z "${unbucketed// /}" ] && echo none || echo "$unbucketed")" "none" diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index 1149c537..40e76ede 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -142,6 +142,7 @@ behaviour, the source of that number is named. - [94. test_approx_aggregates.py: hll and tdigest over columnar tables, the compatibility statement](#94-test_approx_aggregatespy-hll-and-tdigest-over-columnar-tables-the-compatibility-statement) - [95. test_docs_how_to_sections.py: the pg_partman how-to keeps its own heading](#95-test_docs_how_to_sectionspy-the-pg_partman-how-to-keeps-its-own-heading) - [96. test_autovacuum_expire.py: the daemon expires retention tables only when told to](#96-test_autovacuum_expirepy-the-daemon-expires-retention-tables-only-when-told-to) +- [97. test_rollup.py: incremental rollups fold appended row groups, and equal a full recompute](#97-test_rolluppy-incremental-rollups-fold-appended-row-groups-and-equal-a-full-recompute) ## 1. How to read a test in here @@ -6594,3 +6595,27 @@ Written red first: on main the fixture raises `unrecognized configuration parame | `test_a_failing_expire_does_not_undo_the_compaction` | `expire` on a table whose retention column was dropped fails with `42703`, and the compaction due on that table still lands | | `test_the_daemon_yields_the_lock_it_holds_inside_expire` | the worker holds ShareUpdateExclusiveLock inside `expire`; a session's ACCESS EXCLUSIVE is granted within 8 s of a 30 s hold, and the log shows the cancel | | `test_a_table_dropped_mid_pass_is_skipped_and_the_pass_goes_on` | while the worker holds `ex_hold`, `ex_gone` is dropped; after the cancel the same pass logs `ex_gone` as skipped and still expires `ex_after` | + +## 97. test_rollup.py: incremental rollups fold appended row groups, and equal a full recompute + +Port of `rollup.sh` (#1294, phase 1: append-only). `create_rollup` makes a view over a partials table; `refresh_rollup` folds the rows of the row groups appended since the last refresh, in ONE statement, so the group list, the folded rows and the recorded groups share a snapshot. The design, its twelve points and the owner's decisions are on #1294. + +THE ORACLE is a full recompute of the same GROUP BY, every time. Exactness was fixed by the design: count, integer sum, min, max and the avg of an integer column exact; the avg of `float8` within a relative 1e-12. This file uses an integer avg; the shell twin a float one. + +What the arms found while the code was written: +- A lateral per-group subquery was pulled up into a join, and a TID range scan is built only from a relation's own restrictions, so the refresh read every row group: 114 blocks of a 103-block table for one new group. `OFFSET 0` keeps it a scan: 15 blocks. +- `pgcolumnar.vacuum` gives a table a new storage id and may keep its group numbers, so the source's identity is the storage id. +- Every arm run as a superuser bypassed the row-level security on the rollup catalogs. An owner and a stranger are now ordinary roles; PUBLIC has no USAGE on schema `pgcolumnar` (by design, docs/administration.md), so the test grants it as an administrator would. +- The first draft's source load overflowed `int` and left the table empty, and three arms passed on emptiness; each has a premise that requires a real measure now. + +From @jdatcmd's review of #1319: the refresh restated 291 rows a block in SQL, which holds only at the default block size, so the TID range of each group now comes from `pgcolumnar.row_tid` (the access method's own mapping), and an arm requires every group's range to hold exactly its rows. NULL group keys and NULL timestamps have an arm in each twin, and writing it showed both twins' digests dropped any row with a NULL (`||` then `string_agg`); they use `concat_ws` over `coalesce` now. Retention keeps history: a group `expire` retired leaves the snapshot and its buckets keep their rows (#1314's `retired_group`). + +A role that cannot SELECT the source cannot create a rollup of it (42501). `create_rollup` has no check of its own for this: it reads the source's storage id through `get_storage_id`, which refuses such a caller, and the arm was red only when that check was removed. `entry_point_privilege.sh` buckets `create_rollup` as relation-taking by its second argument and `row_tid` as taking no relation. + +Removal proofs, each red on its arm: no `OFFSET 0` (the cost arm), no isolation check, no lock (two refreshes double-fold), no identity check, no changed-group check, the folded groups recorded from a fresh read (rows committed mid-refresh lost), min combined as max, no row-level security, a rebase that clears the rollup, `=` in place of `IS NOT DISTINCT FROM` (only the NULL arm), retired groups kept in the snapshot, and `row_tid` with 290 rows a block. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_rollup_phase_one` | the rollup equals a full recompute after the first refresh, appends with late rows, a bulk append, and a rebase; a refresh with nothing new folds nothing; rows committed during a refresh are folded by the next, once; two concurrent refreshes fold each row once; zones agree; a refresh after a small append reads under a fifth of the source; a rewritten source is refused until `rebase_rollup`; a delete is refused (phase 1); a heap rollup behaves the same; REPEATABLE READ is refused; an owner who is not a superuser creates and refreshes a rollup, and another role sees none of it; bad definitions are refused with `22023` | diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index ff723a7a..fde61ab8 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -571,4 +571,7 @@ guard_tests 415 # # cluster_tests 509 -> 510 for #1290's test_autovacuum_expire.py, by collection # on this branch rebased onto c0ee868. -cluster_tests 510 +# +# cluster_tests 510 -> 511 for #1294's test_rollup.py, one test, collected over the +# cluster files on this branch rebased onto b6656ac (#1307). +cluster_tests 511 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index fe21b028..eeb2fa59 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -129,6 +129,7 @@ "projection_update", "projections", "rewrite_storage_oid", + "rollup", "scan_decode_cost", "sorted_pathkeys", "stats_privilege", diff --git a/test/pytest/test_rollup.py b/test/pytest/test_rollup.py new file mode 100644 index 00000000..84143bbf --- /dev/null +++ b/test/pytest/test_rollup.py @@ -0,0 +1,355 @@ +"""Incremental rollups, phase 1: append-only (#1294). + +A rollup is a view over a partials table; refresh_rollup folds the rows of the +row groups appended since the last refresh, in ONE statement, so the group list, +the folded rows and the recorded groups share a snapshot. The design, its eight +points and the owner's decisions are on #1294. + +THE ORACLE is a full recompute of the same GROUP BY, every time. Exactness was +fixed by the design so it cannot be loosened here: count, integer sum, min and +max exact; avg of an integer column exact too (numeric partials); avg of +float8 within a relative 1e-12. This file uses an integer avg, so it is exact. + +Two things the shell twin found while the code was written, held here too: a +lateral per-group subquery pulled up into a join read every row group (a TID +range is built only from a relation's own restrictions), and every arm run as a +superuser bypassed the row-level security on the rollup catalogs, so an owner +and a stranger are tested as ordinary roles. + +ON A PRIVATE DATABASE (`pgc_own_db`): the roles and pg_stat_reset. Independent +of `test/rollup.sh`: its own table, 15-minute buckets, aggregates (count, sum, +max, avg of an integer), zones, row counts and roles. +""" + +import threading +import time + +import psycopg + +ORIGIN = "2025-12-29 06:00:00+00" +AGG_FUNCS = "{count,sum,max,avg}" +AGG_COLS = "{*,reading,temp,reading}" + + +def _one(conn, sql): + with conn.cursor() as cur: + cur.execute(sql) + row = cur.fetchone() + return row[0] if row else None + + +def _state(conn, sql): + try: + return str(_one(conn, sql)) + except psycopg.Error as e: + return e.sqlstate or "?" + + +def _load(conn, lo, hi, start): + with conn.cursor() as cur: + cur.execute(f"INSERT INTO feed SELECT '{start}'::timestamptz + (g - {lo}) * interval '23 s'," + f" g % 11, (g * 31337) % 65521, ((g * 2654435761) % 4093) / 7.0" + f" FROM generate_series({lo}::bigint, {hi}) g") + + +RECOMPUTE = (f"SELECT date_bin('15 minutes', at, '{ORIGIN}') AS bucket, sensor, count(*) AS count," + " sum(reading)::numeric AS sum_reading, max(temp) AS max_temp, avg(reading) AS avg_reading" + " FROM feed GROUP BY 1, 2") + + +def _digest(conn, rel_sql): + # concat_ws over coalesce: a plain || turns a row with any NULL into NULL, and + # string_agg drops it, so NULL keys would be compared by their absence. + return _one(conn, "SELECT count(*) || '|' || md5(string_agg(concat_ws(',', coalesce(bucket::text, '~')," + " coalesce(sensor::text, '~'), count, coalesce(sum_reading::text, '~')," + " coalesce(max_temp::text, '~'), coalesce(avg_reading::text, '~')), ';'" + f" ORDER BY bucket NULLS FIRST, sensor NULLS FIRST)) FROM ({rel_sql}) x") + + +def _matches(conn, rollup="gauge_15m"): + return _digest(conn, f"SELECT * FROM {rollup}") + + +def _oracle(conn): + return _digest(conn, RECOMPUTE) + + +def _refresh(conn, rollup="gauge_15m"): + return _state(conn, f"SELECT pgcolumnar.refresh_rollup('{rollup}')") + + +def test_rollup_phase_one(pgc_own_db, pgc_cluster, expect): + conn = pgc_own_db + dsn = pgc_cluster.dsn(dbname=_one(conn, "SELECT current_database()")) + expect.num(_one(conn, "SELECT count(*) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace" + " WHERE n.nspname = 'pgcolumnar'" + " AND p.proname IN ('create_rollup', 'refresh_rollup', 'rebase_rollup')"), 3, + "premise: the rollup functions exist") + with conn.cursor() as cur: + cur.execute("CREATE TABLE feed (at timestamptz, sensor int, reading bigint, temp float8) USING pgcolumnar") + cur.execute("SELECT pgcolumnar.set_options('feed', stripe_row_limit => 3000)") + _load(conn, 1, 24000, "2026-02-01 00:00:00+00") + expect.num(_one(conn, "SELECT count(*) FROM feed"), 24000, "premise: the source holds every row") + + with conn.cursor() as cur: + cur.execute(f"SELECT pgcolumnar.create_rollup('gauge_15m', 'feed', 'at', '15 minutes', '{ORIGIN}'," + f" '{{sensor}}', '{AGG_FUNCS}', '{AGG_COLS}')") + expect.text(_one(conn, "SELECT string_agg(attname, ',' ORDER BY attnum) FROM pg_attribute" + " WHERE attrelid = to_regclass('gauge_15m') AND attnum > 0 AND NOT attisdropped"), + "bucket,sensor,count,sum_reading,max_temp,avg_reading", + "the rollup table exists with the documented columns") + + expect.text(_refresh(conn), "24000", "the first refresh folds every row") + expect.text(_matches(conn), _oracle(conn), "and after the first refresh the rollup equals a full recompute") + expect.text(_refresh(conn), "0", "a refresh with nothing new folds nothing") + expect.text(_matches(conn), _oracle(conn), "and leaves the rollup as it was") + + _load(conn, 24001, 27000, "2026-02-08 00:00:00+00") + _load(conn, 27001, 27900, "2026-02-01 03:00:00+00") # late: old buckets + expect.text(_refresh(conn), "3900", + "a refresh after appends folds exactly the appended rows, late ones included") + expect.text(_matches(conn), _oracle(conn), "and the rollup still equals a full recompute") + + # Rows committed while a refresh runs are the next refresh's. + _load(conn, 27901, 28300, "2026-02-09 00:00:00+00") + out = {} + + def paused(): + with psycopg.connect(dsn, autocommit=True) as c: + c.execute("SET pgcolumnar.rollup_pause = on") + out["folded"] = c.execute("SELECT pgcolumnar.refresh_rollup('gauge_15m')").fetchone()[0] + th = threading.Thread(target=paused) + th.start() + time.sleep(0.8) + _load(conn, 28301, 28520, "2026-02-09 05:00:00+00") + th.join(60) + expect.num(out.get("folded"), 400, "the paused refresh folded what was there before it began") + expect.text(_refresh(conn), "220", "the next refresh folds the rows committed during the last one, once") + expect.text(_matches(conn), _oracle(conn), "and after the refresh that ran beside an insert the rollup equals a full recompute") + + # Two refreshes at once. + _load(conn, 28521, 30020, "2026-02-10 00:00:00+00") + got = [] + + def refresher(): + with psycopg.connect(dsn, autocommit=True) as c: + got.append(c.execute("SELECT pgcolumnar.refresh_rollup('gauge_15m')").fetchone()[0]) + ts = [threading.Thread(target=refresher) for _ in range(2)] + for t in ts: + t.start() + for t in ts: + t.join(60) + expect.text(" ".join(str(x) for x in sorted(got)), "0 1500", + "two refreshes started together fold the new rows once between them") + expect.text(_matches(conn), _oracle(conn), "and after two refreshes started together the rollup equals a full recompute") + + # Zones. + for zone, name in (("UTC", "z_utc"), ("America/St_Johns", "z_stjohns")): + with conn.cursor() as cur: + cur.execute(f"SET TimeZone = '{zone}'") + cur.execute(f"SELECT pgcolumnar.create_rollup('{name}', 'feed', 'at', '15 minutes', '{ORIGIN}'," + f" '{{sensor}}', '{AGG_FUNCS}', '{AGG_COLS}')") + cur.execute(f"SELECT pgcolumnar.refresh_rollup('{name}')") + cur.execute("SET TimeZone = 'UTC'") + expect.text(_matches(conn, "z_utc"), _matches(conn, "z_stjohns"), + "refreshes from sessions in different time zones build the same rollup") + expect.num(_one(conn, "SELECT count(*) FROM z_utc"), _one(conn, f"SELECT count(*) FROM ({RECOMPUTE}) z"), + "the zone rollups hold a row per key of the recompute") + expect.at_least(_one(conn, "SELECT count(*) FROM z_utc"), 1, "and that is not zero") + + # Cost, on a source of real size. + _load(conn, 100001, 260000, "2026-03-01 00:00:00+00") + expect.text(_refresh(conn), "160000", "a refresh folds a bulk append") + expect.text(_matches(conn), _oracle(conn), "and after the bulk append the rollup equals a full recompute") + _load(conn, 30021, 30820, "2026-02-11 00:00:00+00") + with conn.cursor() as cur: + cur.execute("SELECT pg_stat_force_next_flush()") + cur.execute("SELECT pg_stat_reset()") + cur.execute("SELECT pgcolumnar.refresh_rollup('gauge_15m')") + cur.execute("SELECT pg_stat_force_next_flush()") + read = _one(conn, "SELECT coalesce(heap_blks_read, 0) + coalesce(heap_blks_hit, 0)" + " FROM pg_statio_user_tables WHERE relname = 'feed'") + pages = _one(conn, "SELECT pg_relation_size('feed') / 8192") + print(f"-- refresh after an 800-row append read {read} of {pages} blocks") + expect.at_least(read, 1, "the refresh after a small append read the source at all") + expect.at_least(pages - 5 * read if read else 0, 1, + "a refresh after a small append reads under a fifth of the source") + expect.text(_matches(conn), _oracle(conn), "and after the small append the rollup equals a full recompute") + + # A rewrite of the source, and rebase. + def identity(): + return _one(conn, "SELECT pgcolumnar.get_storage_id('feed') || ':' || string_agg(group_number::text, ','" + " ORDER BY group_number) FROM pgcolumnar.row_group" + " WHERE storage_id = pgcolumnar.get_storage_id('feed')") + before = identity() + with conn.cursor() as cur: + cur.execute("SELECT pgcolumnar.vacuum('feed')") + expect.text("renumbered" if identity() != before else "same", "renumbered", + "premise: pgcolumnar.vacuum gave the source's row groups new identities") + vb = _matches(conn) + expect.at_least(_one(conn, "SELECT count(*) FROM gauge_15m"), 1, "the rollup holds rows before the rewrite") + expect.text(f"{_refresh(conn)}|{_matches(conn)}", f"55000|{vb}", + "a refresh after the source was rewritten is refused, naming rebase_rollup") + # Not through _state: a SQLSTATE such as 42883 is all digits and would read as a count. + try: + rebased = _one(conn, "SELECT pgcolumnar.rebase_rollup('gauge_15m')") + except psycopg.Error: + rebased = -1 + expect.at_least(rebased, 1, "rebase_rollup records the source's current row groups") + expect.text(_matches(conn), vb, "and leaves the rollup's rows alone") + _load(conn, 30821, 31420, "2026-02-12 00:00:00+00") + expect.text(_refresh(conn), "600", "after a rebase the next refresh folds exactly the appended rows") + expect.text(_matches(conn), _oracle(conn), "and after the rebase the rollup equals a full recompute") + + # Phase 1 refuses anything but appends. + before = _matches(conn) + expect.at_least(_one(conn, "SELECT count(*) FROM gauge_15m"), 1, "the rollup holds rows before the delete") + with conn.cursor() as cur: + cur.execute("DELETE FROM feed WHERE sensor = 4 AND at < '2026-02-01 01:00:00+00'") + expect.text(_refresh(conn), "0A000", "a refresh after a delete is refused until phase 2") + expect.text(_matches(conn), before, "and the refused refresh changed nothing") + + # The TID range of each row group comes from the access method's own mapping + # (row_tid), so it holds exactly the group's rows at any block size. + bad, groups = conn.execute( + "SELECT count(*) FILTER (WHERE got <> want), count(*) FROM (" + " SELECT g.row_count - g.deleted_count AS want," + " (SELECT count(*) FROM feed WHERE ctid >= pgcolumnar.row_tid(g.first_row_number)" + " AND ctid <= pgcolumnar.row_tid(g.first_row_number + g.row_count - 1)) AS got" + " FROM pgcolumnar.rollup_source_groups('feed') g) x").fetchone() + expect.text(f"{bad}|{groups}", + f"0|{_one(conn, 'SELECT count(*) FROM pgcolumnar.row_group WHERE storage_id = pgcolumnar.get_storage_id(' + chr(39) + 'feed' + chr(39) + ')')}", + "each row group's TID range holds exactly that group's rows") + + # NULL keys and NULL timestamps (#1319 review): keys match with IS NOT + # DISTINCT FROM, and a NULL ts is a NULL bucket. + with conn.cursor() as cur: + cur.execute("CREATE TABLE holes (at timestamptz, sensor int, reading bigint, temp float8) USING pgcolumnar") + cur.execute("SELECT pgcolumnar.set_options('holes', stripe_row_limit => 1000)") + cur.execute("INSERT INTO holes SELECT CASE WHEN g % 13 = 0 THEN NULL ELSE '2026-07-01'::timestamptz" + " + g * interval '47 s' END, CASE WHEN g % 6 = 0 THEN NULL ELSE g % 4 END, g, g / 4.0" + " FROM generate_series(1, 2600) g") + cur.execute(f"SELECT pgcolumnar.create_rollup('holes_15m', 'holes', 'at', '15 minutes', '{ORIGIN}'," + f" '{{sensor}}', '{AGG_FUNCS}', '{AGG_COLS}')") + cur.execute("SELECT pgcolumnar.refresh_rollup('holes_15m')") + cur.execute("INSERT INTO holes SELECT CASE WHEN g % 13 = 0 THEN NULL ELSE '2026-07-01'::timestamptz" + " + g * interval '47 s' END, CASE WHEN g % 6 = 0 THEN NULL ELSE g % 4 END, g, g / 4.0" + " FROM generate_series(2601, 3900) g") + expect.text(_refresh(conn, "holes_15m"), "1300", "the second refresh of the NULL-key rollup folded the append") + holes_recompute = RECOMPUTE.replace("FROM feed", "FROM holes") + nulls = f"SELECT count(*) FILTER (WHERE sensor IS NULL) || ' ' || count(*) FILTER (WHERE bucket IS NULL) FROM ({{}}) z" + expect.text(f"{_digest(conn, 'SELECT * FROM holes_15m')}|{_one(conn, nulls.format('SELECT * FROM holes_15m'))}", + f"{_digest(conn, holes_recompute)}|{_one(conn, nulls.format(holes_recompute))}", + "a rollup with NULL keys and NULL timestamps equals a full recompute") + expect.at_least(_one(conn, f"SELECT count(*) FROM ({holes_recompute}) z WHERE sensor IS NULL OR bucket IS NULL"), 2, + "premise: and the fixture has NULL-key and NULL-bucket groups") + + # Retention keeps history: groups expire retires (pgcolumnar.retired_group, + # #1314) leave the snapshot, and their buckets keep what they folded. + with conn.cursor() as cur: + cur.execute("CREATE TABLE aged (at timestamptz, reading int) USING pgcolumnar") + cur.execute("SELECT pgcolumnar.set_options('aged', stripe_row_limit => 1000)") + cur.execute("INSERT INTO aged SELECT now() - interval '90 days' + g * interval '45 s', g % 97" + " FROM generate_series(1, 3200) g") + cur.execute("INSERT INTO aged SELECT now() - interval '3 days' + g * interval '20 s', g % 89" + " FROM generate_series(1, 1600) g") + cur.execute(f"SELECT pgcolumnar.create_rollup('aged_hourly', 'aged', 'at', '1 hour', '{ORIGIN}', '{{}}'," + " '{count,max}', '{*,reading}')") + expect.text(_refresh(conn, "aged_hourly"), "4800", "the history rollup folded every row") + hist = "SELECT md5(string_agg(bucket || ':' || count || ':' || max_reading, ',' ORDER BY bucket)) FROM aged_hourly" + before_hist = _one(conn, hist) + old_buckets = _one(conn, "SELECT count(*) FROM aged_hourly WHERE bucket < now() - interval '60 days'") + expect.at_least(old_buckets, 1, "the history rollup holds old buckets") + with conn.cursor() as cur: + cur.execute("SELECT pgcolumnar.set_options('aged', ttl_column => 'at', ttl_interval => '14 days')") + expect.at_least(_one(conn, "SELECT pgcolumnar.expire('aged')"), 1, "premise: expire retired the old row groups") + expect.text(f"{_refresh(conn, 'aged_hourly')}|{_one(conn, hist)}", f"0|{before_hist}", + "a refresh after retention folds nothing, refuses nothing, and keeps every bucket") + with conn.cursor() as cur: + cur.execute("INSERT INTO aged SELECT now() - interval '5 hours' + g * interval '3 s', g % 83" + " FROM generate_series(1, 700) g") + expect.text(_refresh(conn, "aged_hourly"), "700", "and after retention an append still folds") + recent = _one(conn, "SELECT md5(string_agg(bucket || ':' || count || ':' || max_reading, ',' ORDER BY bucket))" + " FROM aged_hourly WHERE bucket >= now() - interval '14 days'") + recomputed = _one(conn, f"SELECT md5(string_agg(b || ':' || n || ':' || m, ',' ORDER BY b)) FROM" + f" (SELECT date_bin('1 hour', at, '{ORIGIN}') b, count(*) n, max(reading) m FROM aged" + " WHERE at >= now() - interval '14 days' GROUP BY 1) x WHERE b >= now() - interval '14 days'") + still_old = _one(conn, "SELECT count(*) FROM aged_hourly WHERE bucket < now() - interval '60 days'") + expect.text(f"{recent}|{still_old}", f"{recomputed}|{old_buckets}", + "and the recent buckets equal a recompute, while the expired ones keep their history") + + # Ordinary roles: row-level security on the rollup catalogs is what is tested. + with conn.cursor() as cur: + for r in ("rp_owner", "rp_other"): + cur.execute(f"DO $$ BEGIN CREATE ROLE {r}; EXCEPTION WHEN duplicate_object THEN NULL; END $$") + cur.execute(f"GRANT USAGE, CREATE ON SCHEMA public TO {r}") + cur.execute(f"GRANT USAGE ON SCHEMA pgcolumnar TO {r}") + + def as_role(role, sql): + with psycopg.connect(dsn, autocommit=True) as c: + c.execute(f"SET ROLE {role}") + try: + row = c.execute(sql).fetchone() + return str(row[0]) if row else "" + except psycopg.Error as e: + return e.sqlstate or "?" + as_role("rp_owner", "CREATE TABLE meter (at timestamptz, kwh int) USING pgcolumnar") + as_role("rp_owner", "INSERT INTO meter SELECT '2026-04-01'::timestamptz + g * interval '7 min', g % 50" + " FROM generate_series(1, 3000) g") + expect.text(as_role("rp_owner", f"SELECT pgcolumnar.create_rollup('meter_daily', 'meter', 'at', '1 day'," + f" '{ORIGIN}', '{{}}', '{{count,max}}', '{{*,kwh}}')"), + "meter_daily", "a table's owner, not a superuser, creates a rollup of it") + expect.text(as_role("rp_owner", "SELECT pgcolumnar.refresh_rollup('meter_daily')"), "3000", "and refreshes it") + expect.text(as_role("rp_owner", "SELECT string_agg(bucket || ':' || count || ':' || max_kwh, ' ' ORDER BY bucket)" + " FROM meter_daily"), + _one(conn, f"SELECT string_agg(b || ':' || n || ':' || m, ' ' ORDER BY b) FROM" + f" (SELECT date_bin('1 day', at, '{ORIGIN}') b, count(*) n, max(kwh) m FROM meter GROUP BY 1) x"), + "and reads what a recompute gives") + expect.text(f"{as_role('rp_other', 'SELECT count(*) FROM pgcolumnar.rollup')} " + f"{as_role('rp_other', 'SELECT count(*) FROM pgcolumnar.rollup_group')}", "0 0", + "another role sees none of its rollup catalog rows") + expect.text(as_role("rp_other", "SELECT pgcolumnar.refresh_rollup('public.meter_daily')"), "42704", + "and cannot refresh it") + # create_rollup reads the source's storage id through get_storage_id, which + # refuses a caller that cannot SELECT the table: refused at creation. + expect.text(as_role("rp_other", f"SELECT pgcolumnar.create_rollup('other_daily', 'public.meter', 'at', '1 day'," + f" '{ORIGIN}', '{{}}', '{{count}}', '{{*}}')"), + "42501", "a role that cannot read the source cannot create a rollup of it") + + # access_method: a heap rollup folds and matches the same way. + with conn.cursor() as cur: + cur.execute(f"SELECT pgcolumnar.create_rollup('heap_15m', 'feed', 'at', '15 minutes', '{ORIGIN}'," + f" '{{sensor}}', '{AGG_FUNCS}', '{AGG_COLS}', 'heap')") + am = _one(conn, "SELECT a.amname FROM pg_class c JOIN pg_am a ON a.oid = c.relam" + " WHERE c.oid = 'heap_15m__partials'::regclass") + folded = _refresh(conn, "heap_15m") + expect.text(f"{am}|{folded.isdigit() and int(folded) > 0}|{_matches(conn, 'heap_15m')}", + f"heap|True|{_oracle(conn)}", + "a rollup kept in a heap table refreshes and equals a full recompute") + + # REPEATABLE READ is refused: its snapshot would predate the refresh's lock. + with psycopg.connect(dsn, autocommit=True) as c: + c.execute("BEGIN ISOLATION LEVEL REPEATABLE READ") + try: + c.execute("SELECT pgcolumnar.refresh_rollup('z_utc')") + rr = "ok" + except psycopg.Error as e: + rr = e.sqlstate + c.execute("ROLLBACK") + expect.text(rr, "25000", "a refresh at REPEATABLE READ is refused") + + # Refused at creation. + def refused(name, width, funcs, cols): + return _state(conn, f"SELECT pgcolumnar.create_rollup('{name}', 'feed', 'at', '{width}', '{ORIGIN}'," + f" '{{}}', '{funcs}', '{cols}')") + expect.text(refused("q_m", "2 months", "{count}", "{*}"), "22023", "a width with months is refused") + expect.text(refused("q_p", "1 hour", "{stddev}", "{temp}"), "22023", "an aggregate that does not combine is refused") + # More columns than functions, so only the length check can catch it. + expect.text(refused("q_l", "1 hour", "{count}", "{*,temp}"), "22023", + "aggregate functions and columns of different lengths are refused") + # Under min, which has no "cannot be folded" check to catch it first. + expect.text(refused("q_c", "1 hour", "{min}", "{humidity}"), "22023", + "an aggregate column the source does not have is refused") + expect.text(refused("q_s", "1 hour", "{max}", "{*}"), "22023", "* with anything but count is refused") + expect.num(_one(conn, "SELECT count(*) FROM pg_class WHERE relname IN ('q_m', 'q_p', 'q_l', 'q_c', 'q_s')"), 0, + "and no refused rollup left a table behind") diff --git a/test/rollup.sh b/test/rollup.sh new file mode 100755 index 00000000..5d8cb08a --- /dev/null +++ b/test/rollup.sh @@ -0,0 +1,313 @@ +#!/usr/bin/env bash +# +# Incremental rollups, phase 1: append-only (#1294). +# +# WRITTEN BEFORE THE CODE, against the API proposed on #1294: +# +# pgcolumnar.create_rollup(rollup, source, ts_column, width, origin, group_by name[], +# agg_functions text[], agg_columns name[], access_method) +# pgcolumnar.refresh_rollup(rollup regclass) -> rows folded +# pgcolumnar.rebase_rollup(rollup regclass) -> row groups recorded +# +# The rollup is a table, one row per (bucket, group_by...) key, with columns +# `bucket`, the group_by columns, then one per aggregate named `count`, or +# _ (sum_v, min_v, max_v, avg_f). +# +# THE ORACLE is a full recompute of the same GROUP BY over the source, every +# time. Exactness is fixed by the design on #1294 so it cannot be loosened here: +# count, integer sum, min and max exact; avg of float8 within a relative 1e-12. +# +# A refresh is one statement and so one snapshot: rows committed while it runs +# are neither folded nor recorded, and the next refresh takes them, exactly once +# (pgcolumnar.rollup_pause, a developer setting, holds it open for that arm). +# Phase 1 refuses any change but appends, until phase 2's dirty buckets. +# +# Usage: test/rollup.sh [PG_CONFIG] + +set -uo pipefail +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" + +at_least() { [ -z "$1" ] && return; [ "$1" -lt "$2" ] && echo "$1" || echo "$2"; } +sqlstate() { # sqlstate SQL -> the SQLSTATE it fails with, or ok + local out + out=$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \ + -X -At -v ON_ERROR_STOP=1 -v VERBOSITY=sqlstate -c "$1" 2>&1 >/dev/null) + [ -z "$out" ] && { echo ok; return; } + echo "$out" | sed -n 's/^ERROR: *\([0-9A-Z]\{5\}\).*/\1/p' | head -1 +} + +sqlstate_or() { # sqlstate_or SQL -> the single value it returns, or the SQLSTATE it fails with + local out + out=$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \ + -X -At -v ON_ERROR_STOP=1 -v VERBOSITY=sqlstate -c "$1" 2>&1) + case "$out" in ERROR:*) sed -n 's/^ERROR: *\([0-9A-Z]\{5\}\).*/\1/p' <<<"$out" | head -1;; *) echo "$out";; esac +} +q "CREATE EXTENSION IF NOT EXISTS pgcolumnar;" >/dev/null +check_num "premise: the rollup functions exist" \ + "$(q "SELECT count(*) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE n.nspname = 'pgcolumnar' AND p.proname IN ('create_rollup', 'refresh_rollup', 'rebase_rollup');")" "3" + +ORIGIN="2026-01-01 00:00:00+00" +q "CREATE TABLE src (ts timestamptz, dev int, v bigint, f float8) USING pgcolumnar; + SELECT pgcolumnar.set_options('src', stripe_row_limit => 4000);" >/dev/null +load() { # load FROM TO START-TS: rows g in [FROM, TO], 37 s apart from START-TS + # bigint: g * 104729 overflows int past g = 20,505, the INSERT then fails, and + # every arm below would measure an empty table (the first red run did). + q "INSERT INTO src SELECT '$3'::timestamptz + (g - $1) * interval '37 s', g % 7, (g * 7919) % 100003, + ((g * 104729) % 9973) / 13.0 FROM generate_series($1::bigint, $2) g;" >/dev/null +} +load 1 30000 "2026-03-01 00:00:00+00" +check_num "premise: the source holds every row" "$(q "SELECT count(*) FROM src;")" "30000" + +# The oracle, and the rollup read the same way: exact columns hashed, avg kept apart. +RECOMPUTE="SELECT date_bin('1 hour', ts, '$ORIGIN') AS bucket, dev, count(*) AS count, + sum(v)::numeric AS sum_v, min(v) AS min_v, max(v) AS max_v, avg(f) AS avg_f + FROM src GROUP BY 1, 2" +# concat_ws over coalesce: a plain || makes a row with any NULL into NULL, and +# string_agg then drops it, so NULL keys would be compared by not being there. +exact() { q "SELECT count(*) || '|' || md5(string_agg(concat_ws(',', coalesce(bucket::text, '~'), + coalesce(dev::text, '~'), count, coalesce(sum_v::text, '~'), coalesce(min_v::text, '~'), + coalesce(max_v::text, '~')), ';' ORDER BY bucket NULLS FIRST, dev NULLS FIRST)) FROM ($1) x;"; } +# the worst relative difference of avg_f between the rollup and the recompute +avg_gap() { + q "SELECT coalesce(max(abs(r.avg_f - o.avg_f) / greatest(abs(o.avg_f), 1e-300)), 0) < 1e-12 + AND count(*) = (SELECT count(*) FROM ($RECOMPUTE) z) + FROM r_hourly r JOIN ($RECOMPUTE) o USING (bucket, dev);" +} +matches() { echo "$(exact "SELECT * FROM r_hourly")|$(avg_gap)"; } +oracle() { echo "$(exact "$RECOMPUTE")|t"; } + +AGGS="'{count,sum,min,max,avg}', '{*,v,v,v,f}'" +q "SELECT pgcolumnar.create_rollup('r_hourly', 'src', 'ts', '1 hour', '$ORIGIN', '{dev}', $AGGS);" >/dev/null +check "the rollup table exists with the documented columns" \ + "$(q "SELECT string_agg(attname, ',' ORDER BY attnum) FROM pg_attribute + WHERE attrelid = to_regclass('r_hourly') AND attnum > 0 AND NOT attisdropped;")" \ + "bucket,dev,count,sum_v,min_v,max_v,avg_f" + +check_num "the first refresh folds every row" "$(q "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "30000" +check "and after the first refresh the rollup equals a full recompute" "$(matches)" "$(oracle)" +check_num "a refresh with nothing new folds nothing" "$(q "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "0" +check "and leaves the rollup as it was" "$(matches)" "$(oracle)" + +# Appends, including late data: rows whose timestamps land in buckets the rollup +# already has. A time watermark would never see them. +load 30001 33000 "2026-03-13 08:00:00+00" +load 33001 34500 "2026-03-02 05:00:00+00" +check_num "a refresh after appends folds exactly the appended rows, late ones included" \ + "$(q "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "4500" +check "and the rollup still equals a full recompute" "$(matches)" "$(oracle)" + +# Rows committed while a refresh runs: not in its snapshot, so the NEXT refresh +# folds them, once. +load 34501 35000 "2026-03-14 00:00:00+00" +env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -X -At \ + -c "SET pgcolumnar.rollup_pause = on" -c "SELECT 'folded=' || pgcolumnar.refresh_rollup('r_hourly')" \ + > "$PGC_WORKDIR/paused.out" 2>&1 & +refresher=$! +sleep 1 +load 35001 35300 "2026-03-14 06:00:00+00" +wait "$refresher" +check "the paused refresh folded what was there before it began" \ + "$(sed -n 's/^folded=//p' "$PGC_WORKDIR/paused.out")" "500" +check_num "the next refresh folds the rows committed during the last one, once" \ + "$(q "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "300" +check "and after the refresh that ran beside an insert the rollup equals a full recompute" "$(matches)" "$(oracle)" + +# Two refreshes at once: the second waits for the first, then has nothing to fold. +load 35301 37300 "2026-03-15 00:00:00+00" +for i in 1 2; do + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -X -At \ + -c "SELECT pgcolumnar.refresh_rollup('r_hourly')" > "$PGC_WORKDIR/both_$i.out" 2>&1 & +done +wait +check "two refreshes started together fold the new rows once between them" \ + "$(cat "$PGC_WORKDIR/both_1.out" "$PGC_WORKDIR/both_2.out" | sort -n | tr '\n' ' ')" "0 2000 " +check "and after two refreshes started together the rollup equals a full recompute" "$(matches)" "$(oracle)" + +# The bucket is a pure function of ts: refreshes from different zones agree. +for z in UTC Asia/Kolkata; do + r="r_$(tr '/' '_' <<<"$z" | tr 'A-Z' 'a-z')" + q "SET TimeZone = '$z'; SELECT pgcolumnar.create_rollup('$r', 'src', 'ts', '1 hour', '$ORIGIN', '{dev}', $AGGS); + SELECT pgcolumnar.refresh_rollup('$r');" >/dev/null +done +check "refreshes from sessions in different time zones build the same rollup" \ + "$(exact "SELECT * FROM r_utc")" "$(exact "SELECT * FROM r_asia_kolkata")" +# Empty equals empty: in the first red run both were, and the arm above passed. +check_num "the zone rollups hold a row per key of the recompute" \ + "$(q "SELECT count(*) FROM r_utc;")" "$(q "SELECT count(*) FROM ($RECOMPUTE) z;")" +check_num "and that is not zero" "$(at_least "$(q "SELECT count(*) FROM r_utc;")" 1)" "1" + +# Cost: after a small append to the whole table, a refresh reads the new groups. +# The source is grown first: on a 26-page table the refresh's fixed cost (the +# group list, the storage id) is itself more than a fifth (measured: 15 blocks). +load 100001 300000 "2026-04-01 00:00:00+00" +check_num "a refresh folds a bulk append" "$(q "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "200000" +check "and after the bulk append the rollup equals a full recompute" "$(matches)" "$(oracle)" +load 37301 38300 "2026-03-16 00:00:00+00" +q "SELECT pg_stat_reset();" >/dev/null +env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -At \ + -c "SELECT pgcolumnar.refresh_rollup('r_hourly')" -c "SELECT pg_stat_force_next_flush()" >/dev/null 2>&1 +read_blocks="$(q "SELECT coalesce(heap_blks_read, 0) + coalesce(heap_blks_hit, 0) FROM pg_statio_user_tables WHERE relname = 'src';")" +pages="$(q "SELECT pg_relation_size('src') / 8192;")" +echo "-- refresh after a 1000-row append read $read_blocks of $pages blocks" +check_num "the refresh after a small append read the source at all" "$(at_least "$read_blocks" 1)" "1" +# Reading nothing also reads under a fifth: the arm requires a read. +check_num "a refresh after a small append reads under a fifth of the source" \ + "$( [ -n "$read_blocks" ] && [ "$read_blocks" -ge 1 ] && at_least $((pages - 5 * read_blocks)) 1 || echo "read ${read_blocks:-nothing}")" "1" +check "and after the small append the rollup equals a full recompute" "$(matches)" "$(oracle)" + +# A rewrite renumbers every row group (#1294 point 9): the snapshot no longer names +# the source's groups, so a refresh is refused, and rebase_rollup records the +# current groups without touching the rollup. +# The identity is the storage id and the group numbers: measured, pgcolumnar.vacuum +# gives a table a new storage id and may keep its group numbers. +groups_before="$(q "SELECT pgcolumnar.get_storage_id('src') || ':' || string_agg(group_number::text, ',' ORDER BY group_number) + FROM pgcolumnar.row_group WHERE storage_id = pgcolumnar.get_storage_id('src');")" +q "SELECT pgcolumnar.vacuum('src');" >/dev/null +groups_after="$(q "SELECT pgcolumnar.get_storage_id('src') || ':' || string_agg(group_number::text, ',' ORDER BY group_number) + FROM pgcolumnar.row_group WHERE storage_id = pgcolumnar.get_storage_id('src');")" +check "premise: pgcolumnar.vacuum gave the source's row groups new identities" \ + "$([ -n "$groups_after" ] && [ "$groups_after" != "$groups_before" ] && echo renumbered || echo "same: $groups_after")" "renumbered" +vb="$(exact "SELECT * FROM r_hourly")" +check_num "the rollup holds rows before the rewrite" "$(at_least "$(q "SELECT count(*) FROM r_hourly;")" 1)" "1" +check "a refresh after the source was rewritten is refused, naming rebase_rollup" \ + "$(sqlstate "SELECT pgcolumnar.refresh_rollup('r_hourly');")|$(exact "SELECT * FROM r_hourly")" "55000|$vb" +check_num "rebase_rollup records the source's current row groups" \ + "$(at_least "$(q "SELECT pgcolumnar.rebase_rollup('r_hourly');")" 1)" "1" +check "and leaves the rollup's rows alone" "$(exact "SELECT * FROM r_hourly")" "$vb" +load 38301 39100 "2026-03-17 00:00:00+00" +check_num "after a rebase the next refresh folds exactly the appended rows" \ + "$(q "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "800" +check "and after the rebase the rollup equals a full recompute" "$(matches)" "$(oracle)" + +# Phase 1 folds appends only; anything else is refused, and the rollup is untouched. +before="$(exact "SELECT * FROM r_hourly")" +check_num "the rollup holds rows before the delete" "$(at_least "$(q "SELECT count(*) FROM r_hourly;")" 1)" "1" +q "DELETE FROM src WHERE dev = 3 AND ts < '2026-03-01 06:00:00+00';" >/dev/null +check "a refresh after a delete is refused until phase 2" \ + "$(sqlstate "SELECT pgcolumnar.refresh_rollup('r_hourly');")" "0A000" +check "and the refused refresh changed nothing" "$(exact "SELECT * FROM r_hourly")" "$before" + +# Refused at creation. +# Each new group is read through a TID range built from its row numbers. The +# mapping is the access method's (row_tid, in C), not restated in SQL: 291 rows +# per block holds only at the default block size (#1319 review). Every group's +# range must hold exactly its rows. +check "each row group's TID range holds exactly that group's rows" \ + "$(q "SELECT count(*) FILTER (WHERE got <> want) || '|' || count(*) FROM ( + SELECT g.row_count - g.deleted_count AS want, + (SELECT count(*) FROM src WHERE ctid >= pgcolumnar.row_tid(g.first_row_number) + AND ctid <= pgcolumnar.row_tid(g.first_row_number + g.row_count - 1)) AS got + FROM pgcolumnar.rollup_source_groups('src') g) x;")" \ + "0|$(q "SELECT count(*) FROM pgcolumnar.row_group WHERE storage_id = pgcolumnar.get_storage_id('src');")" + +# NULL group keys and NULL timestamps: the fold matches keys with IS NOT DISTINCT +# FROM, and a NULL ts is a NULL bucket. Measured right by @jdatcmd on #1319; an +# = in place of IS NOT DISTINCT FROM would duplicate the NULL partials. +q "CREATE TABLE nk (ts timestamptz, dev int, v bigint) USING pgcolumnar; + SELECT pgcolumnar.set_options('nk', stripe_row_limit => 1000); + INSERT INTO nk SELECT CASE WHEN g % 11 = 0 THEN NULL ELSE '2026-06-01'::timestamptz + g * interval '61 s' END, + CASE WHEN g % 7 = 0 THEN NULL ELSE g % 5 END, g FROM generate_series(1, 3000) g;" >/dev/null +q "SELECT pgcolumnar.create_rollup('r_nk', 'nk', 'ts', '1 hour', '$ORIGIN', '{dev}', '{count,sum,min,max}', '{*,v,v,v}'); + SELECT pgcolumnar.refresh_rollup('r_nk');" >/dev/null +q "INSERT INTO nk SELECT CASE WHEN g % 11 = 0 THEN NULL ELSE '2026-06-01'::timestamptz + g * interval '61 s' END, + CASE WHEN g % 7 = 0 THEN NULL ELSE g % 5 END, g FROM generate_series(3001, 4500) g;" >/dev/null +check_num "the second refresh of the NULL-key rollup folded the append" "$(q "SELECT pgcolumnar.refresh_rollup('r_nk');")" "1500" +NKR="SELECT date_bin('1 hour', ts, '$ORIGIN') AS bucket, dev, count(*) AS count, sum(v)::numeric AS sum_v, + min(v) AS min_v, max(v) AS max_v FROM nk GROUP BY 1, 2" +check "a rollup with NULL keys and NULL timestamps equals a full recompute" \ + "$(exact "SELECT * FROM r_nk")|$(q "SELECT count(*) FROM r_nk WHERE dev IS NULL;") $(q "SELECT count(*) FROM r_nk WHERE bucket IS NULL;")" \ + "$(exact "$NKR")|$(q "SELECT count(*) FROM ($NKR) z WHERE dev IS NULL;") $(q "SELECT count(*) FROM ($NKR) z WHERE bucket IS NULL;")" +check_num "premise: and the fixture has NULL-key and NULL-bucket groups" \ + "$(at_least "$(q "SELECT count(*) FROM ($NKR) z WHERE dev IS NULL OR bucket IS NULL;")" 2)" "2" + +# Retention keeps history (owner's decision on #1294 point 2): groups that expire +# retires are recorded in pgcolumnar.retired_group (#1314), and a refresh drops +# them from its snapshot without touching their buckets. Any other missing group +# is still refused (the delete arm above). +q "CREATE TABLE hist (ts timestamptz, v int) USING pgcolumnar; + SELECT pgcolumnar.set_options('hist', stripe_row_limit => 1000); + INSERT INTO hist SELECT now() - interval '200 days' + g * interval '1 min', g FROM generate_series(1, 3000) g; + INSERT INTO hist SELECT now() - interval '1 day' + g * interval '10 s', g FROM generate_series(1, 2000) g;" >/dev/null +q "SELECT pgcolumnar.create_rollup('r_hist', 'hist', 'ts', '1 day', '$ORIGIN', '{}', '{count,sum}', '{*,v}');" >/dev/null +check_num "the history rollup folded every row" "$(q "SELECT pgcolumnar.refresh_rollup('r_hist');")" "5000" +hb="$(q "SELECT md5(string_agg(bucket || ':' || count || ':' || sum_v, ',' ORDER BY bucket)) FROM r_hist;")" +ob="$(q "SELECT count(*) FROM r_hist WHERE bucket < now() - interval '100 days';")" +check_num "the history rollup holds old buckets" "$(at_least "${ob:-0}" 1)" "1" +q "SELECT pgcolumnar.set_options('hist', ttl_column => 'ts', ttl_interval => '30 days');" >/dev/null +retired="$(q "SELECT pgcolumnar.expire('hist');")" +check_num "premise: expire retired the old row groups" "$(at_least "${retired:-0}" 1)" "1" +check "a refresh after retention folds nothing, refuses nothing, and keeps every bucket" \ + "$(sqlstate_or "SELECT pgcolumnar.refresh_rollup('r_hist');")|$(q "SELECT md5(string_agg(bucket || ':' || count || ':' || sum_v, ',' ORDER BY bucket)) FROM r_hist;")" \ + "0|$hb" +q "INSERT INTO hist SELECT now() - interval '2 hours' + g * interval '1 s', g FROM generate_series(1, 500) g;" >/dev/null +check_num "and after retention an append still folds" "$(q "SELECT pgcolumnar.refresh_rollup('r_hist');")" "500" +check "and the recent buckets equal a recompute, while the expired ones keep their history" \ + "$(q "SELECT md5(string_agg(bucket || ':' || count || ':' || sum_v, ',' ORDER BY bucket)) FROM r_hist WHERE bucket >= now() - interval '30 days';")|$(q "SELECT count(*) FROM r_hist WHERE bucket < now() - interval '100 days';")" \ + "$(q "SELECT md5(string_agg(b || ':' || n || ':' || s, ',' ORDER BY b)) FROM (SELECT date_bin('1 day', ts, '$ORIGIN') b, count(*) n, sum(v)::numeric s FROM hist WHERE ts >= now() - interval '30 days' GROUP BY 1) x WHERE b >= now() - interval '30 days';")|$ob" + +# Not only for superusers: a superuser bypasses row-level security, so every arm +# above ran past the policies on pgcolumnar.rollup and rollup_group. A table's +# owner can roll it up; another role sees nothing of it and cannot refresh it. +# PUBLIC has no USAGE on schema pgcolumnar (measured: 42501 for any call), so the +# roles get it as an administrator would grant it; what is then tested is the +# row-level security on the rollup catalogs. +q "CREATE ROLE ru_a; CREATE ROLE ru_b; GRANT USAGE, CREATE ON SCHEMA public TO ru_a, ru_b; + GRANT USAGE ON SCHEMA pgcolumnar TO ru_a, ru_b;" >/dev/null +as_role() { env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -X -At \ + -v VERBOSITY=sqlstate -c "SET ROLE $1" -c "$2" 2>&1 | grep -v '^SET$' | tail -1; } +as_role ru_a "CREATE TABLE own_src (ts timestamptz, v int) USING pgcolumnar" >/dev/null +as_role ru_a "INSERT INTO own_src SELECT '2026-05-01'::timestamptz + g * interval '1 min', g FROM generate_series(1, 5000) g" >/dev/null +check "a table's owner, not a superuser, creates a rollup of it" \ + "$(as_role ru_a "SELECT pgcolumnar.create_rollup('own_r', 'own_src', 'ts', '1 day', '$ORIGIN', '{}', '{count,sum}', '{*,v}')")" "own_r" +check_num "and refreshes it" "$(as_role ru_a "SELECT pgcolumnar.refresh_rollup('own_r')")" "5000" +check "and reads what a recompute gives" \ + "$(as_role ru_a "SELECT string_agg(bucket || ':' || count || ':' || sum_v, ' ' ORDER BY bucket) FROM own_r")" \ + "$(q "SELECT string_agg(b || ':' || n || ':' || s, ' ' ORDER BY b) FROM (SELECT date_bin('1 day', ts, '$ORIGIN') b, count(*) n, sum(v)::numeric s FROM own_src GROUP BY 1) x;")" +check "another role sees none of its rollup catalog rows" \ + "$(as_role ru_b "SELECT count(*) FROM pgcolumnar.rollup") $(as_role ru_b "SELECT count(*) FROM pgcolumnar.rollup_group")" "0 0" +# Row-level security hides the catalog row, so to that role it is no rollup at +# all (42704), which also says nothing about whether one exists. +check "and cannot refresh it" "$(as_role ru_b "SELECT pgcolumnar.refresh_rollup('public.own_r')" | sed -n 's/^ERROR: *\([0-9A-Z]\{5\}\).*/\1/p')" "42704" +# create_rollup takes the source as a relation, and it reads the source's +# storage id through pgcolumnar.get_storage_id, which refuses a caller that +# cannot SELECT the table (42501; one of entry_point_privilege.sh's guarded +# entry points). So the refusal comes at creation, not at the first refresh. +check "a role that cannot read the source cannot create a rollup of it" \ + "$(as_role ru_b "SELECT pgcolumnar.create_rollup('b_r', 'public.own_src', 'ts', '1 day', '$ORIGIN', '{}', '{count}', '{*}')" | sed -n 's/^ERROR: *\([0-9A-Z]\{5\}\).*/\1/p')" "42501" + +# access_method: the partials need not be columnar (#1294 point 12 measures which +# serves a hot rollup better). A heap rollup folds and matches the same way. +q "SELECT pgcolumnar.create_rollup('r_heap', 'src', 'ts', '1 hour', '$ORIGIN', '{dev}', $AGGS, 'heap');" >/dev/null +check "a rollup kept in a heap table refreshes and equals a full recompute" \ + "$(q "SELECT a.amname FROM pg_class c JOIN pg_am a ON a.oid = c.relam WHERE c.oid = 'r_heap__partials'::regclass;")|$(q "SELECT pgcolumnar.refresh_rollup('r_heap') > 0;")|$(exact "SELECT * FROM r_heap")" \ + "heap|t|$(exact "$RECOMPUTE")" + +# REPEATABLE READ: the caller's snapshot would predate the refresh's lock, and a +# refresh that waited for another would fold the same groups again. +check "a refresh at REPEATABLE READ is refused" \ + "$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -X -At \ + -v VERBOSITY=sqlstate -c "BEGIN ISOLATION LEVEL REPEATABLE READ" \ + -c "SELECT pgcolumnar.refresh_rollup('r_utc')" -c "ROLLBACK" 2>&1 | + sed -n 's/^ERROR: *\([0-9A-Z]\{5\}\).*/\1/p' | head -1)" "25000" + +# Refused at creation, each with 22023: the aggregates are structured input +# checked against the five that combine, never SQL text (#1294 point 11). +refused_create() { sqlstate "SELECT pgcolumnar.create_rollup('$1', 'src', 'ts', '$2', '$ORIGIN', '{}', '$3', '$4');"; } +check "a width with months is refused" "$(refused_create r_m '1 month' '{count}' '{*}')" "22023" +check "an aggregate that does not combine is refused" "$(refused_create r_p '1 hour' '{median}' '{v}')" "22023" +# More columns than functions: a shorter columns array would be caught by the +# column check instead, and the arm would not see this check go. +check "aggregate functions and columns of different lengths are refused" \ + "$(refused_create r_l '1 hour' '{count}' '{*,v}')" "22023" +# Under min: under sum the "cannot be folded" check would catch it first. +check "an aggregate column the source does not have is refused" \ + "$(refused_create r_c '1 hour' '{min}' '{nosuch}')" "22023" +check "* with anything but count is refused" "$(refused_create r_s '1 hour' '{sum}' '{*}')" "22023" +# An arm, not a premise: with a refusal removed, the create succeeds and leaves a +# table (the months-check mutation did), so a repository change reddens it. +check_num "and no refused rollup left a table behind" \ + "$(q "SELECT count(*) FROM pg_class WHERE relname IN ('r_m', 'r_p', 'r_l', 'r_c', 'r_s');")" "0" + +pgc_summary diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index b81030a1..9859dd5f 100755 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -282,6 +282,7 @@ SUITES=( rewrite_group_scan rewrite_storage_oid rls_direct_storage + rollup row_triggers scan_decode_cost scan_direction