From 01dd6155276ed97adcad3e5c55c064b7b8116bfd Mon Sep 17 00:00:00 2001
From: "Andrey M. Borodin" <x4mmm@night.local>
Date: Tue, 23 Jan 2024 23:03:28 +0500
Subject: [PATCH v3 3/3] amcheck: avoid failing on oversized tuples

Due to changes in toast policies, some heap tuples might become too
big index tuples. This commit prevents ERRORs in this case, because
this anomaly does not create dangerous conditions to DMS. However,
the database cannot be dumped-restored, so we emit a NOTICE.

Reported-by: Alexander Lakhin
---
 contrib/amcheck/expected/check_btree.out | 14 ++++++
 contrib/amcheck/sql/check_btree.sql      | 11 +++++
 contrib/amcheck/verify_nbtree.c          | 54 +++++++++++++++++++++---
 3 files changed, 73 insertions(+), 6 deletions(-)

diff --git a/contrib/amcheck/expected/check_btree.out b/contrib/amcheck/expected/check_btree.out
index 36b1ebe34c..91394bed2b 100644
--- a/contrib/amcheck/expected/check_btree.out
+++ b/contrib/amcheck/expected/check_btree.out
@@ -270,6 +270,20 @@ SELECT bt_index_check('tbl_idx', true);
  
 (1 row)
 
+DROP TABLE tbl;
+-- Check oversized datums that cannot be inserted into index
+CREATE TABLE t(f1 text);
+CREATE INDEX t_idx ON t(f1);
+INSERT INTO t VALUES(repeat('1234567890', 1000));
+ALTER TABLE t ALTER COLUMN f1 SET STORAGE plain;
+SELECT bt_index_check('t_idx', true);
+NOTICE:  Index contain tuples that cannot fit into index page, if toasted with current toast policy
+ bt_index_check 
+----------------
+ 
+(1 row)
+
+DROP TABLE t;
 -- cleanup
 DROP TABLE bttest_a;
 DROP TABLE bttest_b;
diff --git a/contrib/amcheck/sql/check_btree.sql b/contrib/amcheck/sql/check_btree.sql
index 0a827edd1a..cdfbc8a141 100644
--- a/contrib/amcheck/sql/check_btree.sql
+++ b/contrib/amcheck/sql/check_btree.sql
@@ -172,6 +172,17 @@ INSERT INTO tbl VALUES (repeat('Test', 250));
 ALTER TABLE tbl ALTER COLUMN t SET STORAGE extended;
 
 SELECT bt_index_check('tbl_idx', true);
+DROP TABLE tbl;
+
+-- Check oversized datums that cannot be inserted into index
+
+CREATE TABLE t(f1 text);
+CREATE INDEX t_idx ON t(f1);
+INSERT INTO t VALUES(repeat('1234567890', 1000));
+ALTER TABLE t ALTER COLUMN f1 SET STORAGE plain;
+
+SELECT bt_index_check('t_idx', true);
+DROP TABLE t;
 
 -- cleanup
 DROP TABLE bttest_a;
diff --git a/contrib/amcheck/verify_nbtree.c b/contrib/amcheck/verify_nbtree.c
index 6654b5afe7..f70e01191b 100644
--- a/contrib/amcheck/verify_nbtree.c
+++ b/contrib/amcheck/verify_nbtree.c
@@ -128,6 +128,11 @@ typedef struct BtreeCheckState
 	bloom_filter *filter;
 	/* Debug counter */
 	int64		heaptuplespresent;
+	/*
+	 * During check we might find tuples that due to current TOAST policies
+	 * should not reside in index, but still are there.
+	 */
+	bool has_oversized_tuples;
 } BtreeCheckState;
 
 /*
@@ -1624,10 +1629,17 @@ bt_target_page_check(BtreeCheckState *state)
 
 					logtuple = bt_posting_plain_tuple(itup, i);
 					norm = bt_normalize_tuple(state, logtuple);
-					bloom_add_element(state->filter, (unsigned char *) norm,
+					if (norm == NULL)
+					{
+						if (!state->has_oversized_tuples)
+							elog(NOTICE, "Index contain tuples that cannot fit into index page, if toasted with current toast policy");
+						state->has_oversized_tuples = true;
+					}
+					else
+						bloom_add_element(state->filter, (unsigned char *) norm,
 									  IndexTupleSize(norm));
 					/* Be tidy */
-					if (norm != logtuple)
+					if (norm != logtuple && norm != NULL)
 						pfree(norm);
 					pfree(logtuple);
 				}
@@ -1635,10 +1647,17 @@ bt_target_page_check(BtreeCheckState *state)
 			else
 			{
 				norm = bt_normalize_tuple(state, itup);
-				bloom_add_element(state->filter, (unsigned char *) norm,
-								  IndexTupleSize(norm));
+				if (norm == NULL)
+				{
+					if (!state->has_oversized_tuples)
+						elog(NOTICE, "Index contain tuples that cannot fit into index page, if toasted with current toast policy");
+					state->has_oversized_tuples = true;
+				}
+				else
+					bloom_add_element(state->filter, (unsigned char *) norm,
+								IndexTupleSize(norm));
 				/* Be tidy */
-				if (norm != itup)
+				if (norm != itup && norm != NULL)
 					pfree(norm);
 			}
 		}
@@ -2876,6 +2895,7 @@ bt_tuple_present_callback(Relation index, ItemPointer tid, Datum *values,
 	BtreeCheckState *state = (BtreeCheckState *) checkstate;
 	IndexTuple	itup,
 				norm;
+	bool miss_oversized_tuple = false;
 
 	Assert(state->heapallindexed);
 
@@ -2883,9 +2903,19 @@ bt_tuple_present_callback(Relation index, ItemPointer tid, Datum *values,
 	itup = index_form_tuple(RelationGetDescr(index), values, isnull);
 	itup->t_tid = *tid;
 	norm = bt_normalize_tuple(state, itup);
+	if (norm == NULL)
+	{
+		if (state->has_oversized_tuples)
+		{
+			/* exempt this oversized tuple */
+			state->heaptuplespresent++;
+			pfree(itup);
+			return;
+		}
+	}
 
 	/* Probe Bloom filter -- tuple should be present */
-	if (bloom_lacks_element(state->filter, (unsigned char *) norm,
+	if ((norm == NULL) || bloom_lacks_element(state->filter, (unsigned char *) norm,
 							IndexTupleSize(norm)))
 		ereport(ERROR,
 				(errcode(ERRCODE_DATA_CORRUPTED),
@@ -2936,6 +2966,9 @@ bt_tuple_present_callback(Relation index, ItemPointer tid, Datum *values,
  * Caller does normalization for non-pivot tuples that have a posting list,
  * since dummy CREATE INDEX callback code generates new tuples with the same
  * normalized representation.
+ * 
+ * If the tuple is exampt from checking due to has_oversized_tuples this function
+ * returns NULL.
  */
 static IndexTuple
 bt_normalize_tuple(BtreeCheckState *state, IndexTuple itup)
@@ -2947,6 +2980,7 @@ bt_normalize_tuple(BtreeCheckState *state, IndexTuple itup)
 	bool		formnewtup = false;
 	IndexTuple	reformed;
 	int			i;
+	Size		data_size;
 
 	/* Caller should only pass "logical" non-pivot tuples here */
 	Assert(!BTreeTupleIsPosting(itup) && !BTreeTupleIsPivot(itup));
@@ -3026,6 +3060,14 @@ bt_normalize_tuple(BtreeCheckState *state, IndexTuple itup)
 	if (!formnewtup)
 		return itup;
 
+	data_size = MAXALIGN(heap_compute_data_size(tupleDescriptor,
+									   normalized, isnull)
+				+ MAXALIGN(sizeof(IndexTupleData) + sizeof(IndexAttributeBitMapData)));
+	if ((data_size & INDEX_SIZE_MASK) != data_size)
+	{
+		return NULL;
+	}
+
 	/*
 	 * Hard case: Tuple had compressed varlena datums that necessitate
 	 * creating normalized version of the tuple from uncompressed input datums
-- 
2.37.1 (Apple Git-137.1)

