Antonin Houska <[email protected]> wrote:
> shihao zhong <[email protected]> wrote:
>
> > 2.
> > Looks like some rows are lost when the table grows/gap fill.
> >
> > heapScan->rs_nblocks is fixed when the scan starts. At a range boundary,
> > changes to blocks at or beyond rs_nblocks are not in [range_start,
> > range_end),
> > so they are skipped. The next snapshot can see those tuples, but the scan
> > never
> > reaches those blocks.
> >
> > The attached extend.spec has 5 blocks. It uses
> > repack_snapshot_after = 2, pauses at the first boundary, and inserts 100
> > rows. 70 of them go to blocks 5 to 7 and are missing after REPACK.
>
> Interestiong. I think we need to get the new rows from new blocks each time we
> process the concurrent changes.
I could spend a little while on this item. Attached is what fixes your
extend.spec.text.
--
Antonin Houska
Web: https://www.cybertec-postgresql.com
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);
