diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index e423340ae70..f2d33494d8f 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -61,6 +61,7 @@ static Snapshot finalize_block_range(Relation rel_old, Relation rel_dst, MultiXactId multi_cutoff, ChangeContext *chgcxt, BlockNumber cur, BlockNumber start, BlockNumber *end_p, + BlockNumber nblocks, RewriteState *rwstate_p, BlockNumber *range_start_dst_p); static void reform_and_rewrite_tuple(TupleTableSlot *src, TupleTableSlot *reform, @@ -626,6 +627,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, BlockNumber range_end = InvalidBlockNumber; BlockNumber range_start_new = InvalidBlockNumber; BlockNumber range_end_new = InvalidBlockNumber; + BlockNumber nblocks; Relation rel_dst; /* Remember if it's a system catalog */ @@ -742,6 +744,18 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, snapshot = repack_get_snapshot(chgcxt); chgcxt->cc_last_snapshot_xmin = snapshot->xmin; + /* + * Unlike a scan during query processing, we also need to capture + * changes on pages that the scan will not reach: when processing + * particular range of pages, we always need to process changes on + * pages beyond nblocks. + * + * This had to wait until logical decoding has started (i.e. until we + * received the snapshot above), otherwise we could miss tuples + * inserted right after the block that we consider the last one. + */ + nblocks = heapScan->rs_nblocks; + /* * Since we currently do not freeze tuples when replaying data * changes, the replayed transactions should not precede the new value @@ -937,7 +951,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, OldestXmin, *xid_cutoff, *multi_cutoff, chgcxt, blkno, range_start, - &range_end, &rwstate, + &range_end, nblocks, + &rwstate, &range_start_new); /* Finally check the tuple visibility. */ @@ -1002,6 +1017,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, */ end_of_wal = GetFlushRecPtr(NULL); repack_process_concurrent_changes(chgcxt, end_of_wal, + InvalidBlockNumber, InvalidBlockNumber, InvalidBlockNumber, false, false); @@ -1170,6 +1186,7 @@ finalize_block_range(Relation rel_old, Relation rel_dst, MultiXactId multi_cutoff, ChangeContext *chgcxt, BlockNumber cur, BlockNumber start, BlockNumber *end_p, + BlockNumber nblocks, RewriteState *rwstate_p, BlockNumber *range_start_dst_p) { @@ -1216,8 +1233,8 @@ finalize_block_range(Relation rel_old, Relation rel_dst, * rid of the old tuple in the current range. */ end_of_wal = GetFlushRecPtr(NULL); - repack_process_concurrent_changes(chgcxt, end_of_wal, start, end, true, - false); + repack_process_concurrent_changes(chgcxt, end_of_wal, start, end, nblocks, + true, false); /* * A new snapshot will be pushed below. Note that it's important to not do diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 8cdf0d3ed20..eb8ceaed82e 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -208,7 +208,8 @@ static bool repack_is_permitted_for_relation(RepackCommand cmd, static void apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, - BlockNumber range_end); + BlockNumber range_end, + BlockNumber nblocks); static void apply_concurrent_insert(RepackDest *dest, TupleTableSlot *spill_tuple, TupleTableSlot *new_tuple, @@ -227,7 +228,7 @@ static void restore_tuple(BufFile *file, Relation relation, static void adjust_toast_pointers(Relation relation, TupleTableSlot *dest, TupleTableSlot *src); static bool is_block_in_range(BlockNumber blknum, BlockNumber start, - BlockNumber end); + BlockNumber end, BlockNumber nblocks); static bool find_target_tuple(RepackDest *dest, TupleTableSlot *locator, TupleTableSlot *retrieved); static bool identity_key_equal(RepackDest *dest, @@ -3174,11 +3175,12 @@ RepackCommandAsString(RepackCommand cmd) } /* - * Apply data changes that affect pages in given range. + * Apply data changes that affect pages in given range (plus changes beyond + * nblocks). */ static void apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, - BlockNumber range_end) + BlockNumber range_end, BlockNumber nblocks) { ConcurrentChangeKind kind = '\0'; RepackDest *dest; @@ -3275,7 +3277,7 @@ apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, * range does not matter). */ if (!check_range || - is_block_in_range(block, range_start, range_end)) + is_block_in_range(block, range_start, range_end, nblocks)) apply_concurrent_insert(dest, spilled_tuple, new_tuple, xid); } else if (kind == CHANGE_DELETE) @@ -3285,7 +3287,7 @@ apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, * range does not matter). */ if (!check_range || - is_block_in_range(block, range_start, range_end)) + is_block_in_range(block, range_start, range_end, nblocks)) { bool found; @@ -3311,8 +3313,8 @@ apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, * current range. */ if (!check_range || - (is_block_in_range(old_block, range_start, range_end) && - is_block_in_range(block, range_start, range_end))) + (is_block_in_range(old_block, range_start, range_end, nblocks) && + is_block_in_range(block, range_start, range_end, nblocks))) { /* Find the tuple to be updated or deleted. */ found = find_target_tuple(dest, key, ondisk_tuple); @@ -3335,7 +3337,7 @@ apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, { Assert(check_range); - if (is_block_in_range(block, range_start, range_end)) + if (is_block_in_range(block, range_start, range_end, nblocks)) { /* * The old key is in another range, so only insert the new @@ -3353,7 +3355,8 @@ apply_concurrent_changes(ChangeContext *chgcxt, BlockNumber range_start, apply_concurrent_insert(dest, spilled_tuple, new_tuple, xid); } - else if (is_block_in_range(old_block, range_start, range_end)) + else if (is_block_in_range(old_block, range_start, range_end, + nblocks)) { found = find_target_tuple(dest, key, ondisk_tuple); if (!found) @@ -3673,11 +3676,18 @@ adjust_toast_pointers(Relation relation, TupleTableSlot *dest, * copied. */ static bool -is_block_in_range(BlockNumber blknum, BlockNumber start, BlockNumber end) +is_block_in_range(BlockNumber blknum, BlockNumber start, BlockNumber end, + BlockNumber nblocks) { Assert(BlockNumberIsValid(start) && BlockNumberIsValid(end)); Assert(BlockNumberIsValid(blknum)); + /* + * Changes beyond the scan boundary are always needed. + */ + if (blknum >= nblocks) + return true; + if (start < end) return blknum >= start && blknum < end; else @@ -4285,6 +4295,7 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ repack_process_concurrent_changes(chgcxt, end_of_wal, InvalidBlockNumber, InvalidBlockNumber, + InvalidBlockNumber, false, false); /* @@ -4344,6 +4355,7 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ repack_process_concurrent_changes(chgcxt, end_of_wal, InvalidBlockNumber, InvalidBlockNumber, + InvalidBlockNumber, false, true); /* Remember info about rel before closing OldHeap */ @@ -5105,7 +5117,8 @@ repack_get_snapshot(ChangeContext *chgcxt) /* * Get concurrent changes, up to (and including) the record whose LSN is * 'end_of_wal', from the decoding worker, and apply them to the new table. If - * block range is specified, only apply changes related to that range. + * block range is specified, only apply changes related to that range, plus + * changes in blocks >= nblocks. * * If 'request_snapshot' is true, the snapshot built at LSN following the last * data change needs to be exported too. @@ -5115,6 +5128,7 @@ repack_process_concurrent_changes(ChangeContext *chgcxt, XLogRecPtr end_of_wal, BlockNumber range_start, BlockNumber range_end, + BlockNumber nblocks, bool request_snapshot, bool done) { DecodingWorkerShared *shared; @@ -5163,7 +5177,7 @@ repack_process_concurrent_changes(ChangeContext *chgcxt, #endif /* Apply the changes to the new table. */ - apply_concurrent_changes(chgcxt, range_start, range_end); + apply_concurrent_changes(chgcxt, range_start, range_end, nblocks); /* Get ready for the next set of changes. */ chgcxt->cc_file_seq_changes++; diff --git a/src/include/commands/repack.h b/src/include/commands/repack.h index 9e2f3e491c6..2f3312428ee 100644 --- a/src/include/commands/repack.h +++ b/src/include/commands/repack.h @@ -165,6 +165,7 @@ extern void repack_process_concurrent_changes(ChangeContext *chgcxt, XLogRecPtr end_of_wal, BlockNumber range_start, BlockNumber range_end, + BlockNumber nblocks, bool request_snapshot, bool done); extern void HandleRepackMessageInterrupt(void); extern void ProcessRepackMessages(void);