From 5f68103bf91881389314605b886eb9bc62177bc9 Mon Sep 17 00:00:00 2001 From: Mikhail Nikalayeu Date: Sat, 10 Oct 2026 14:53:20 +0200 Subject: [PATCH v7 4/7] Allow snapshot resets in concurrent BRIN builds BRIN keeps values in a range summary until the range is written out. For out-of-line values, those copies are only TOAST pointers. Resetting the snapshot can let vacuum or pruning remove the data before the next comparison, causing the build to fail. Fetch external values before adding them to the summary, and enable amcanresetsnapshot. Do this for all BRIN builds. Allocate retained values in the range's memory context so they are freed when the range is done. This also prevents minmax_multi from serializing TOAST pointers during a build; the insert path is unchanged. Add a test that deletes a summarized row between snapshot resets and checks that the build can still read the retained value. Reviewed-by: Matthias van de Meent --- src/backend/access/brin/brin.c | 95 +++++++++++++- .../expected/cic_reset_snapshots.out | 21 +++- .../sql/cic_reset_snapshots.sql | 16 +++ src/test/modules/test_misc/meson.build | 1 + .../modules/test_misc/t/016_cic_brin_toast.pl | 116 ++++++++++++++++++ 5 files changed, 242 insertions(+), 7 deletions(-) create mode 100644 src/test/modules/test_misc/t/016_cic_brin_toast.pl diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index bf6fad98f6a..eeaa788859f 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -19,6 +19,7 @@ #include "access/brin_page.h" #include "access/brin_pageops.h" #include "access/brin_xlog.h" +#include "access/detoast.h" #include "access/relation.h" #include "access/reloptions.h" #include "access/relscan.h" @@ -30,6 +31,7 @@ #include "commands/vacuum.h" #include "executor/instrument.h" #include "miscadmin.h" +#include "nodes/nodeFuncs.h" #include "pgstat.h" #include "postmaster/autovacuum.h" #include "storage/bufmgr.h" @@ -39,6 +41,7 @@ #include "tcop/tcopprot.h" #include "utils/acl.h" #include "utils/datum.h" +#include "utils/lsyscache.h" #include "utils/fmgrprotos.h" #include "utils/guc.h" #include "utils/index_selfuncs.h" @@ -169,6 +172,9 @@ typedef struct BrinBuildState BrinDesc *bs_bdesc; BrinMemTuple *bs_dtuple; + /* does each indexed column or expression return a varlena? */ + bool *bs_attvarlena; + BrinTuple *bs_emptyTuple; Size bs_emptyTupleLen; MemoryContext bs_context; @@ -274,8 +280,7 @@ brinhandler(PG_FUNCTION_ARGS) .ampredlocks = false, .amcanparallel = false, .amcanbuildparallel = true, - /* not yet: a range summary keeps values across many heap pages */ - .amcanresetsnapshot = false, + .amcanresetsnapshot = true, .amcaninclude = false, .amusemaintenanceworkmem = false, .amsummarizing = true, @@ -991,6 +996,82 @@ brinendscan(IndexScanDesc scan) pfree(opaque); } +/* + * Identify varlena inputs to the BRIN opclass. + * + * Use the heap column or index expression type. The index descriptor may + * show the opclass's STORAGE type instead of its input type. + */ +static bool * +brin_build_attvarlena(Relation heap, IndexInfo *indexInfo) +{ + int natts = indexInfo->ii_NumIndexAttrs; + bool *attvarlena = palloc_array(bool, natts); + ListCell *expr = list_head(indexInfo->ii_Expressions); + + for (int i = 0; i < natts; i++) + { + AttrNumber attnum = indexInfo->ii_IndexAttrNumbers[i]; + + if (attnum != 0) + attvarlena[i] = + (TupleDescAttr(RelationGetDescr(heap), attnum - 1)->attlen == -1); + else + { + attvarlena[i] = (get_typlen(exprType((Node *) lfirst(expr))) == -1); + expr = lnext(indexInfo->ii_Expressions, expr); + } + } + + return attvarlena; +} + +/* + * Add one heap tuple to the range summary. + * + * Fetch external values before the opclass copies them. Retaining a TOAST + * pointer until brin_form_tuple() is too late: after a snapshot reset, vacuum + * may remove the referenced data while the range is still being summarized. + */ +static void +brin_add_values(BrinBuildState *state, Relation index, Datum *values, + bool *isnull) +{ + int natts = state->bs_bdesc->bd_tupdesc->natts; + Datum fetched[INDEX_MAX_KEYS]; + int fetchedatts[INDEX_MAX_KEYS]; + int nfetched = 0; + MemoryContext oldcxt; + + for (int i = 0; i < natts; i++) + { + if (isnull[i] || !state->bs_attvarlena[i] || + !VARATT_IS_EXTERNAL(DatumGetPointer(values[i]))) + continue; + + if (nfetched == 0) + { + memcpy(fetched, values, natts * sizeof(Datum)); + values = fetched; + } + + values[i] = PointerGetDatum(detoast_external_attr(DatumGetPointer(values[i]))); + fetchedatts[nfetched++] = i; + } + + /* + * Allocate retained values in the context that brin_memtuple_initialize() + * resets for each range. + */ + oldcxt = MemoryContextSwitchTo(state->bs_dtuple->bt_context); + (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple, + values, isnull); + MemoryContextSwitchTo(oldcxt); + + for (int i = 0; i < nfetched; i++) + pfree(DatumGetPointer(values[fetchedatts[i]])); +} + /* * Per-heap-tuple callback for table_index_build_scan. * @@ -1036,8 +1117,7 @@ brinbuildCallback(Relation index, } /* Accumulate the current tuple into the running state */ - (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple, - values, isnull); + brin_add_values(state, index, values, isnull); } /* @@ -1101,8 +1181,7 @@ brinbuildCallbackParallel(Relation index, } /* Accumulate the current tuple into the running state */ - (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple, - values, isnull); + brin_add_values(state, index, values, isnull); } /* @@ -1166,6 +1245,7 @@ brinbuild(Relation heap, Relation index, IndexInfo *indexInfo) revmap = brinRevmapInitialize(index, &pagesPerRange); state = initialize_brin_buildstate(index, revmap, pagesPerRange, RelationGetNumberOfBlocks(heap)); + state->bs_attvarlena = brin_build_attvarlena(heap, indexInfo); /* * Attempt to launch parallel worker scan when required @@ -1696,6 +1776,7 @@ initialize_brin_buildstate(Relation idxRel, BrinRevmap *revmap, /* Remember the memory context to use for an empty tuple, if needed. */ state->bs_context = CurrentMemoryContext; + state->bs_attvarlena = NULL; state->bs_emptyTuple = NULL; state->bs_emptyTupleLen = 0; @@ -1952,6 +2033,7 @@ brinsummarize(Relation index, Relation heapRel, BlockNumber pageRange, pagesPerRange, InvalidBlockNumber); indexInfo = BuildIndexInfo(index); + state->bs_attvarlena = brin_build_attvarlena(heapRel, indexInfo); } summarize_range(indexInfo, state, heapRel, startBlk, heapNumBlocks); @@ -2844,6 +2926,7 @@ _brin_parallel_scan_and_build(BrinBuildState *state, /* Join parallel scan */ indexInfo = BuildIndexInfo(index); indexInfo->ii_Concurrent = brinshared->isconcurrent; + state->bs_attvarlena = brin_build_attvarlena(heap, indexInfo); scan = table_beginscan_parallel(heap, ParallelTableScanFromBrinShared(brinshared), diff --git a/src/test/modules/injection_points/expected/cic_reset_snapshots.out b/src/test/modules/injection_points/expected/cic_reset_snapshots.out index b460d45838d..972e7836fae 100644 --- a/src/test/modules/injection_points/expected/cic_reset_snapshots.out +++ b/src/test/modules/injection_points/expected/cic_reset_snapshots.out @@ -111,7 +111,25 @@ NOTICE: notice triggered for injection point heap_reset_scan_snapshot_xid_assig NOTICE: notice triggered for injection point heap_reset_scan_snapshot_xid_assigned DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING BRIN(i); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot +DROP INDEX CONCURRENTLY cic_reset_snap.idx; +-- Check TOAST handling when an opclass's STORAGE type differs from its +-- input type. The build must use the input type to identify varlena values. +CREATE OPERATOR CLASS cic_reset_snap.int4_bloom_storage_ops FOR TYPE int4 + USING brin AS + OPERATOR 1 =(int4, int4), + FUNCTION 1 brin_bloom_opcinfo(internal), + FUNCTION 2 brin_bloom_add_value(internal, internal, internal, internal), + FUNCTION 3 brin_bloom_consistent(internal, internal, internal, int4), + FUNCTION 4 brin_bloom_union(internal, internal, internal), + FUNCTION 5 brin_bloom_options(internal), + FUNCTION 11 hashint4(int4), + STORAGE pg_brin_bloom_summary; +CREATE INDEX CONCURRENTLY idx + ON cic_reset_snap.tbl USING BRIN(i cic_reset_snap.int4_bloom_storage_ops); +NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot DROP INDEX CONCURRENTLY cic_reset_snap.idx; CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING HASH(i); NOTICE: notice triggered for injection point heap_beginscan_reset_snapshot @@ -202,10 +220,11 @@ SELECT injection_points_detach('heap_reset_scan_snapshot_swap'); (1 row) DROP SCHEMA cic_reset_snap CASCADE; -NOTICE: drop cascades to 3 other objects +NOTICE: drop cascades to 4 other objects DETAIL: drop cascades to table cic_reset_snap.tbl drop cascades to function cic_reset_snap.predicate_stable(integer) drop cascades to function cic_reset_snap.predicate_stable_no_param() +drop cascades to operator family cic_reset_snap.int4_bloom_storage_ops for access method brin SELECT injection_points_detach('heap_beginscan_reset_snapshot'); injection_points_detach ------------------------- diff --git a/src/test/modules/injection_points/sql/cic_reset_snapshots.sql b/src/test/modules/injection_points/sql/cic_reset_snapshots.sql index 35a9e290823..a0da9c33e02 100644 --- a/src/test/modules/injection_points/sql/cic_reset_snapshots.sql +++ b/src/test/modules/injection_points/sql/cic_reset_snapshots.sql @@ -65,6 +65,22 @@ CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING BRIN(i); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; +-- Check TOAST handling when an opclass's STORAGE type differs from its +-- input type. The build must use the input type to identify varlena values. +CREATE OPERATOR CLASS cic_reset_snap.int4_bloom_storage_ops FOR TYPE int4 + USING brin AS + OPERATOR 1 =(int4, int4), + FUNCTION 1 brin_bloom_opcinfo(internal), + FUNCTION 2 brin_bloom_add_value(internal, internal, internal, internal), + FUNCTION 3 brin_bloom_consistent(internal, internal, internal, int4), + FUNCTION 4 brin_bloom_union(internal, internal, internal), + FUNCTION 5 brin_bloom_options(internal), + FUNCTION 11 hashint4(int4), + STORAGE pg_brin_bloom_summary; +CREATE INDEX CONCURRENTLY idx + ON cic_reset_snap.tbl USING BRIN(i cic_reset_snap.int4_bloom_storage_ops); +DROP INDEX CONCURRENTLY cic_reset_snap.idx; + CREATE INDEX CONCURRENTLY idx ON cic_reset_snap.tbl USING HASH(i); REINDEX INDEX CONCURRENTLY cic_reset_snap.idx; DROP INDEX CONCURRENTLY cic_reset_snap.idx; diff --git a/src/test/modules/test_misc/meson.build b/src/test/modules/test_misc/meson.build index 5d81f5b13be..eb1e6b3662e 100644 --- a/src/test/modules/test_misc/meson.build +++ b/src/test/modules/test_misc/meson.build @@ -24,6 +24,7 @@ tests += { 't/013_temp_obj_multisession.pl', 't/014_log_statement_max_length.pl', 't/015_temp_schema_exit_deferrable.pl', + 't/016_cic_brin_toast.pl', ], # The injection points are cluster-wide, so disable installcheck 'runningcheck': false, diff --git a/src/test/modules/test_misc/t/016_cic_brin_toast.pl b/src/test/modules/test_misc/t/016_cic_brin_toast.pl new file mode 100644 index 00000000000..2f22d2c47db --- /dev/null +++ b/src/test/modules/test_misc/t/016_cic_brin_toast.pl @@ -0,0 +1,116 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Check that BRIN fetches external values before resetting its snapshot. +# Pause a concurrent build after it summarizes page 0, then delete that +# page's row. A retained TOAST pointer would fail on the next comparison +# when the build's newer snapshot allows the TOAST data to be pruned. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +plan skip_all => 'Injection points not supported by this build' + unless $ENV{enable_injection_points} eq 'yes'; + +my $node = PostgreSQL::Test::Cluster->new('node'); +$node->init; +# Nothing else may hold back the horizon. +$node->append_conf('postgresql.conf', 'autovacuum = off'); +$node->start; + +plan skip_all => 'Extension injection_points not installed' + unless $node->check_extension('injection_points'); + +# One row per page whatever the block size, with values stored out of line. +$node->safe_psql( + 'postgres', q[ +CREATE EXTENSION injection_points; +SELECT current_setting('block_size')::int AS bs, + current_setting('block_size')::int / 2 AS pad \gset +CREATE TABLE brin_toast (i int, t text, pad char(:pad)); +ALTER TABLE brin_toast ALTER COLUMN t SET STORAGE EXTERNAL, + ALTER COLUMN pad SET STORAGE PLAIN; +INSERT INTO brin_toast + SELECT g, repeat(chr(64 + g), :bs * 4), 'x' FROM generate_series(1, 2) g; +]); + +my $cic = $node->background_psql('postgres', on_error_stop => 0); +$cic->query_safe( + q[ +SET concurrent_index_reset_snapshot_interval = 0; +SELECT injection_points_set_local(); +SELECT injection_points_attach('heap_reset_scan_snapshot_swap', 'wait'); +]); +$cic->query_until( + qr/starting/, q[ +\echo starting +CREATE INDEX CONCURRENTLY brin_toast_idx ON brin_toast USING brin (t); +]); + +# The build stops before each page's snapshot reset. Commit a transaction +# at the first stop so a changed xmin identifies the second stop. +my $xmin = wait_for_stop($node, 'NULL'); +$node->safe_psql( + 'postgres', q[ +SELECT pg_current_xact_id(); +SELECT injection_points_wakeup('heap_reset_scan_snapshot_swap'); +]); + +# Detach before waking up, so the build runs to the end. +wait_for_stop($node, "'$xmin'::xid"); +$node->safe_psql( + 'postgres', q[ +DELETE FROM brin_toast WHERE i = 1; +SELECT injection_points_detach('heap_reset_scan_snapshot_swap'); +SELECT injection_points_wakeup('heap_reset_scan_snapshot_swap'); +]); +$cic->quit; + +is( $node->safe_psql( + 'postgres', + q[SELECT indisvalid FROM pg_index WHERE indexrelid = 'brin_toast_idx'::regclass] + ), + 't', + 'build survived the delete of a summarized row'); + +# Verify that the index remains usable after the delete. +like( + $node->safe_psql( + 'postgres', q[ +SET enable_seqscan = off; +EXPLAIN (COSTS OFF) SELECT count(*) FROM brin_toast WHERE t < 'C'; +]), + qr/Bitmap Index Scan on brin_toast_idx/, + 'index is used'); +is( $node->safe_psql( + 'postgres', q[ +SET enable_seqscan = off; +SELECT count(*) FROM brin_toast WHERE t < 'C'; +]), + '1', + 'index scan finds the remaining row'); + +done_testing(); + +# Wait until the build sleeps at the reset point with an xmin other than +# the given one, and return its xmin. +sub wait_for_stop +{ + my ($node, $prev_xmin) = @_; + + $node->poll_query_until( + 'postgres', qq[ +SELECT count(*) > 0 FROM pg_stat_activity +WHERE wait_event = 'heap_reset_scan_snapshot_swap' + AND backend_xmin IS DISTINCT FROM $prev_xmin +]) or die "timed out waiting for the build to stop at a snapshot reset"; + + return $node->safe_psql( + 'postgres', q[ +SELECT backend_xmin FROM pg_stat_activity +WHERE wait_event = 'heap_reset_scan_snapshot_swap' +]); +} -- 2.43.0