Add additional information in the vacuum error context.

Amit Kapila · Amit Kapila · commit 7e453634bb62 · 2020-08-26T09:40:52.000+05:30
The additional information added will be an offset number for heap operations. This information will help us in finding the exact tuple due to which the error has occurred. Author: Mahendra Singh Thalor and Amit Kapila Reviewed-by: Sawada Masahiko, Justin Pryzby and Amit Kapila Discussion: https://postgr.es/m/CAKYtNApK488TDF4bMbw+1QH8HJf9cxdNDXquhU50TK5iv_FtCQ@mail.gmail.com
diff --git a/src/backend/access/heap/pruneheap.c b/src/backend/access/heap/pruneheap.c
@@ -188,7 +188,7 @@ heap_page_prune_opt(Relation relation, Buffer buffer)
 			/* OK to prune */
 			(void) heap_page_prune(relation, buffer, vistest,
 								   limited_xmin, limited_ts,
-								   true, &ignore);
+								   true, &ignore, NULL);
 		}
 
 		/* And release buffer lock */
@@ -213,6 +213,9 @@ heap_page_prune_opt(Relation relation, Buffer buffer)
  * send its own new total to pgstats, and we don't want this delta applied
  * on top of that.)
  *
+ * off_loc is the offset location required by the caller to use in error
+ * callback.
+ *
  * Returns the number of tuples deleted from the page and sets
  * latestRemovedXid.
  */
@@ -221,7 +224,8 @@ heap_page_prune(Relation relation, Buffer buffer,
 				GlobalVisState *vistest,
 				TransactionId old_snap_xmin,
 				TimestampTz old_snap_ts,
-				bool report_stats, TransactionId *latestRemovedXid)
+				bool report_stats, TransactionId *latestRemovedXid,
+				OffsetNumber *off_loc)
 {
 	int			ndeleted = 0;
 	Page		page = BufferGetPage(buffer);
@@ -262,6 +266,13 @@ heap_page_prune(Relation relation, Buffer buffer,
 		if (prstate.marked[offnum])
 			continue;
 
+		/*
+		 * Set the offset number so that we can display it along with any
+		 * error that occurred while processing this tuple.
+		 */
+		if (off_loc)
+			*off_loc = offnum;
+
 		/* Nothing to do if slot is empty or already dead */
 		itemid = PageGetItemId(page, offnum);
 		if (!ItemIdIsUsed(itemid) || ItemIdIsDead(itemid))
@@ -271,6 +282,10 @@ heap_page_prune(Relation relation, Buffer buffer,
 		ndeleted += heap_prune_chain(buffer, offnum, &prstate);
 	}
 
+	/* Clear the offset information once we have processed the given page. */
+	if (off_loc)
+		*off_loc = InvalidOffsetNumber;
+
 	/* Any error while applying the changes is critical */
 	START_CRIT_SECTION();
 
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
@@ -316,13 +316,15 @@ typedef struct LVRelStats
 	/* Used for error callback */
 	char	   *indname;
 	BlockNumber blkno;			/* used only for heap operations */
+	OffsetNumber offnum;		/* used only for heap operations */
 	VacErrPhase phase;
 } LVRelStats;
 
 /* Struct for saving and restoring vacuum error information. */
 typedef struct LVSavedErrInfo
 {
 	BlockNumber blkno;
+	OffsetNumber offnum;
 	VacErrPhase phase;
 } LVSavedErrInfo;
 
@@ -341,7 +343,8 @@ static void lazy_scan_heap(Relation onerel, VacuumParams *params,
 						   LVRelStats *vacrelstats, Relation *Irel, int nindexes,
 						   bool aggressive);
 static void lazy_vacuum_heap(Relation onerel, LVRelStats *vacrelstats);
-static bool lazy_check_needs_freeze(Buffer buf, bool *hastup);
+static bool lazy_check_needs_freeze(Buffer buf, bool *hastup,
+									LVRelStats *vacrelstats);
 static void lazy_vacuum_all_indexes(Relation onerel, Relation *Irel,
 									IndexBulkDeleteResult **stats,
 									LVRelStats *vacrelstats, LVParallelState *lps,
@@ -364,6 +367,7 @@ static void lazy_record_dead_tuple(LVDeadTuples *dead_tuples,
 static bool lazy_tid_reaped(ItemPointer itemptr, void *state);
 static int	vac_cmp_itemptr(const void *left, const void *right);
 static bool heap_page_is_all_visible(Relation rel, Buffer buf,
+									 LVRelStats *vacrelstats,
 									 TransactionId *visibility_cutoff_xid, bool *all_frozen);
 static void lazy_parallel_vacuum_indexes(Relation *Irel, IndexBulkDeleteResult **stats,
 										 LVRelStats *vacrelstats, LVParallelState *lps,
@@ -396,7 +400,8 @@ static LVSharedIndStats *get_indstats(LVShared *lvshared, int n);
 static bool skip_parallel_vacuum_index(Relation indrel, LVShared *lvshared);
 static void vacuum_error_callback(void *arg);
 static void update_vacuum_error_info(LVRelStats *errinfo, LVSavedErrInfo *saved_err_info,
-									 int phase, BlockNumber blkno);
+									 int phase, BlockNumber blkno,
+									 OffsetNumber offnum);
 static void restore_vacuum_error_info(LVRelStats *errinfo, const LVSavedErrInfo *saved_err_info);
 
 
@@ -547,7 +552,8 @@ heap_vacuum_rel(Relation onerel, VacuumParams *params,
 		 * revert to the previous phase.
 		 */
 		update_vacuum_error_info(vacrelstats, NULL, VACUUM_ERRCB_PHASE_TRUNCATE,
-								 vacrelstats->nonempty_pages);
+								 vacrelstats->nonempty_pages,
+								 InvalidOffsetNumber);
 		lazy_truncate_heap(onerel, vacrelstats);
 	}
 
@@ -960,7 +966,7 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 		pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED, blkno);
 
 		update_vacuum_error_info(vacrelstats, NULL, VACUUM_ERRCB_PHASE_SCAN_HEAP,
-								 blkno);
+								 blkno, InvalidOffsetNumber);
 
 		if (blkno == next_unskippable_block)
 		{
@@ -1129,7 +1135,7 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 			 * to use lazy_check_needs_freeze() for both situations, though.
 			 */
 			LockBuffer(buf, BUFFER_LOCK_SHARE);
-			if (!lazy_check_needs_freeze(buf, &hastup))
+			if (!lazy_check_needs_freeze(buf, &hastup, vacrelstats))
 			{
 				UnlockReleaseBuffer(buf);
 				vacrelstats->scanned_pages++;
@@ -1244,7 +1250,8 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 		 */
 		tups_vacuumed += heap_page_prune(onerel, buf, vistest, false,
 										 InvalidTransactionId, 0,
-										 &vacrelstats->latestRemovedXid);
+										 &vacrelstats->latestRemovedXid,
+										 &vacrelstats->offnum);
 
 		/*
 		 * Now scan the page to collect vacuumable items and check for tuples
@@ -1267,6 +1274,11 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 		{
 			ItemId		itemid;
 
+			/*
+			 * Set the offset number so that we can display it along with any
+			 * error that occurred while processing this tuple.
+			 */
+			vacrelstats->offnum = offnum;
 			itemid = PageGetItemId(page, offnum);
 
 			/* Unused items require no processing, but we count 'em */
@@ -1468,6 +1480,12 @@ lazy_scan_heap(Relation onerel, VacuumParams *params, LVRelStats *vacrelstats,
 			}
 		}						/* scan along page */
 
+		/*
+		 * Clear the offset information once we have processed all the tuples
+		 * on the page.
+		 */
+		vacrelstats->offnum = InvalidOffsetNumber;
+
 		/*
 		 * If we froze any tuples, mark the buffer dirty, and write a WAL
 		 * record recording the changes.  We must log the changes to be
@@ -1845,7 +1863,7 @@ lazy_vacuum_heap(Relation onerel, LVRelStats *vacrelstats)
 
 	/* Update error traceback information */
 	update_vacuum_error_info(vacrelstats, &saved_err_info, VACUUM_ERRCB_PHASE_VACUUM_HEAP,
-							 InvalidBlockNumber);
+							 InvalidBlockNumber, InvalidOffsetNumber);
 
 	pg_rusage_init(&ru0);
 	npages = 0;
@@ -1927,7 +1945,7 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 
 	/* Update error traceback information */
 	update_vacuum_error_info(vacrelstats, &saved_err_info, VACUUM_ERRCB_PHASE_VACUUM_HEAP,
-							 blkno);
+							 blkno, InvalidOffsetNumber);
 
 	START_CRIT_SECTION();
 
@@ -1979,7 +1997,8 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
 	 * dirty, exclusively locked, and, if needed, a full page image has been
 	 * emitted in the log_heap_clean() above.
 	 */
-	if (heap_page_is_all_visible(onerel, buffer, &visibility_cutoff_xid,
+	if (heap_page_is_all_visible(onerel, buffer, vacrelstats,
+								 &visibility_cutoff_xid,
 								 &all_frozen))
 		PageSetAllVisible(page);
 
@@ -2018,7 +2037,7 @@ lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
  * Also returns a flag indicating whether page contains any tuples at all.
  */
 static bool
-lazy_check_needs_freeze(Buffer buf, bool *hastup)
+lazy_check_needs_freeze(Buffer buf, bool *hastup, LVRelStats *vacrelstats)
 {
 	Page		page = BufferGetPage(buf);
 	OffsetNumber offnum,
@@ -2043,6 +2062,11 @@ lazy_check_needs_freeze(Buffer buf, bool *hastup)
 	{
 		ItemId		itemid;
 
+		/*
+		 * Set the offset number so that we can display it along with any
+		 * error that occurred while processing this tuple.
+		 */
+		vacrelstats->offnum = offnum;
 		itemid = PageGetItemId(page, offnum);
 
 		/* this should match hastup test in count_nondeletable_pages() */
@@ -2057,10 +2081,13 @@ lazy_check_needs_freeze(Buffer buf, bool *hastup)
 
 		if (heap_tuple_needs_freeze(tupleheader, FreezeLimit,
 									MultiXactCutoff, buf))
-			return true;
+			break;
 	}							/* scan along page */
 
-	return false;
+	/* Clear the offset information once we have processed the given page. */
+	vacrelstats->offnum = InvalidOffsetNumber;
+
+	return (offnum <= maxoff);
 }
 
 /*
@@ -2438,7 +2465,7 @@ lazy_vacuum_index(Relation indrel, IndexBulkDeleteResult **stats,
 	vacrelstats->indname = pstrdup(RelationGetRelationName(indrel));
 	update_vacuum_error_info(vacrelstats, &saved_err_info,
 							 VACUUM_ERRCB_PHASE_VACUUM_INDEX,
-							 InvalidBlockNumber);
+							 InvalidBlockNumber, InvalidOffsetNumber);
 
 	/* Do bulk deletion */
 	*stats = index_bulk_delete(&ivinfo, *stats,
@@ -2498,7 +2525,7 @@ lazy_cleanup_index(Relation indrel,
 	vacrelstats->indname = pstrdup(RelationGetRelationName(indrel));
 	update_vacuum_error_info(vacrelstats, &saved_err_info,
 							 VACUUM_ERRCB_PHASE_INDEX_CLEANUP,
-							 InvalidBlockNumber);
+							 InvalidBlockNumber, InvalidOffsetNumber);
 
 	*stats = index_vacuum_cleanup(&ivinfo, *stats);
 
@@ -2522,7 +2549,7 @@ lazy_cleanup_index(Relation indrel,
 						   pg_rusage_show(&ru0))));
 	}
 
-	/* Revert back to the old phase information for error traceback */
+	/* Revert to the previous phase information for error traceback */
 	restore_vacuum_error_info(vacrelstats, &saved_err_info);
 	pfree(vacrelstats->indname);
 	vacrelstats->indname = NULL;
@@ -2964,6 +2991,7 @@ vac_cmp_itemptr(const void *left, const void *right)
  */
 static bool
 heap_page_is_all_visible(Relation rel, Buffer buf,
+						 LVRelStats *vacrelstats,
 						 TransactionId *visibility_cutoff_xid,
 						 bool *all_frozen)
 {
@@ -2988,6 +3016,11 @@ heap_page_is_all_visible(Relation rel, Buffer buf,
 		ItemId		itemid;
 		HeapTupleData tuple;
 
+		/*
+		 * Set the offset number so that we can display it along with any
+		 * error that occurred while processing this tuple.
+		 */
+		vacrelstats->offnum = offnum;
 		itemid = PageGetItemId(page, offnum);
 
 		/* Unused or redirect line pointers are of no interest */
@@ -3065,6 +3098,9 @@ heap_page_is_all_visible(Relation rel, Buffer buf,
 		}
 	}							/* scan along page */
 
+	/* Clear the offset information once we have processed the given page. */
+	vacrelstats->offnum = InvalidOffsetNumber;
+
 	return all_visible;
 }
 
@@ -3586,17 +3622,29 @@ vacuum_error_callback(void *arg)
 	{
 		case VACUUM_ERRCB_PHASE_SCAN_HEAP:
 			if (BlockNumberIsValid(errinfo->blkno))
-				errcontext("while scanning block %u of relation \"%s.%s\"",
-						   errinfo->blkno, errinfo->relnamespace, errinfo->relname);
+			{
+				if (OffsetNumberIsValid(errinfo->offnum))
+					errcontext("while scanning block %u and offset %u of relation \"%s.%s\"",
+							   errinfo->blkno, errinfo->offnum, errinfo->relnamespace, errinfo->relname);
+				else
+					errcontext("while scanning block %u of relation \"%s.%s\"",
+							   errinfo->blkno, errinfo->relnamespace, errinfo->relname);
+			}
 			else
 				errcontext("while scanning relation \"%s.%s\"",
 						   errinfo->relnamespace, errinfo->relname);
 			break;
 
 		case VACUUM_ERRCB_PHASE_VACUUM_HEAP:
 			if (BlockNumberIsValid(errinfo->blkno))
-				errcontext("while vacuuming block %u of relation \"%s.%s\"",
-						   errinfo->blkno, errinfo->relnamespace, errinfo->relname);
+			{
+				if (OffsetNumberIsValid(errinfo->offnum))
+					errcontext("while vacuuming block %u and offset %u of relation \"%s.%s\"",
+							   errinfo->blkno, errinfo->offnum, errinfo->relnamespace, errinfo->relname);
+				else
+					errcontext("while vacuuming block %u of relation \"%s.%s\"",
+							   errinfo->blkno, errinfo->relnamespace, errinfo->relname);
+			}
 			else
 				errcontext("while vacuuming relation \"%s.%s\"",
 						   errinfo->relnamespace, errinfo->relname);
@@ -3631,15 +3679,17 @@ vacuum_error_callback(void *arg)
  */
 static void
 update_vacuum_error_info(LVRelStats *errinfo, LVSavedErrInfo *saved_err_info, int phase,
-						 BlockNumber blkno)
+						 BlockNumber blkno, OffsetNumber offnum)
 {
 	if (saved_err_info)
 	{
+		saved_err_info->offnum = errinfo->offnum;
 		saved_err_info->blkno = errinfo->blkno;
 		saved_err_info->phase = errinfo->phase;
 	}
 
 	errinfo->blkno = blkno;
+	errinfo->offnum = offnum;
 	errinfo->phase = phase;
 }
 
@@ -3650,5 +3700,6 @@ static void
 restore_vacuum_error_info(LVRelStats *errinfo, const LVSavedErrInfo *saved_err_info)
 {
 	errinfo->blkno = saved_err_info->blkno;
+	errinfo->offnum = saved_err_info->offnum;
 	errinfo->phase = saved_err_info->phase;
 }
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
@@ -178,7 +178,8 @@ extern int	heap_page_prune(Relation relation, Buffer buffer,
 							struct GlobalVisState *vistest,
 							TransactionId limited_oldest_xmin,
 							TimestampTz limited_oldest_ts,
-							bool report_stats, TransactionId *latestRemovedXid);
+							bool report_stats, TransactionId *latestRemovedXid,
+							OffsetNumber *off_loc);
 extern void heap_page_prune_execute(Buffer buffer,
 									OffsetNumber *redirected, int nredirected,
 									OffsetNumber *nowdead, int ndead,
