[#26283] YSQL: Support yb_index_check() execution using multiple snapshots

Summary:
===Problem===
Index consistency checker (`yb_index_check()`) currently employs a BNL LEFT join (to join the index and the base relation) followed by a `count(*)` on the base relation to detect any index consistency issue. The details of these operations and checks is mentioned in the summary of D41376 / 10de037.

yb_index_check() is a slow operation. Performance experiments show that `yb_index_check()` on an index with `pg_table_size()` of 3GB took 600 seconds in a single region, multi-AZ 3-node cluster. Any operation whose execution time is longer than `timestamp_history_retention_interval_sec` (default value is 15 mins) is susceptible to `Snapshot too old` error. Assuming `timestamp_history_retention_interval_sec` is set to the default, yb_index_check() is susceptible to this error on indexes with `pg_table_size()` >= 4.5 GB.

===Solution===
To resolve this shortcoming, this revision adds support for 'multi-snapshot' execution mode for yb_index_check(). In this mode, the left subplan of the JOIN is divided into small batches. A new (latest) snapshot is picked for processing each batch. The batch size is such that its processing is guaranteed to complete within `timestamp_history_retention_interval_sec` (details below). Hence, in this execution mode, yb_index_check() is guaranteed not to run into the `Snapshot too old` error.

===Implementation details===

===== Batching =====
**Terminology**
There are two batches involved
    ## BNL batch: batch inside the BNL join
    ## Checker batch: batch of rows of the left subplan of the JOIN that will be processed using the same snapshot during yb_index_check().

**Checker batch size**
To ensure all the rows are processed at least once, the checker batch size should be a multiple of the BNL batch size. This is because the BNL output is ordered by left relation ybctid across the batches, but not within a batch. For instance, consider the following scenario:
BNL batch size = 3
| BNL batch | output | max ybctid encountered
| 1 | ybctid1, ... | ybctid1
| 1 | ybctid3, ... | ybctid3
| 1 | ybctid2, ... | ybctid3
| 2 | ybctid6, ... | ybctid6
| 2 | ybctid4, ... | ybctid6
| 2 | ybctid5, ... | ybctid6
Note that the ybctids within a batch are not ordered, but across the batch, they are ordered. Now, say if the checker batch size = 4. The checker's first batch will finish at the 4th row, with max ybctid encountered == ybctid6. Consequently, rows corresponding to ybctid4 and ybctid5 will never be processed.

We keep processing rows in multiples of yb_bnl_batch_size within a checker batch as long as the elapsed time > 70% of timestamp_history_retention_interval_sec. This threshold of 70% is based on a heuristic. The idea is to keep it closer to 100% so that as many rows as possible are processed within a single batch, to avoid the overhead of creating too many batches. At the same time, keeping it too close to 100% risks running into the Snapshot too old error in scenarios when elapsed time is marginally less than the threshold, and hence the next set of rows are processed in the same batch, but that pushes the elapsed time beyond the timestamp_history_retention_interval_sec.

**Batch processing**
While processing a checker batch, we keep a track of the maximum processed ybctid (of left relation) and pass it as a lower bound when initializing the next checker batch.

**Controlling execution mode**
yb_index_check() now takes an optional bool argument `single_snapshot_mode` to control the execution mode. As the name suggests, if it is true, the execution mode is 'single snapshot' (all the rows are processed using a single snapshot) and vice-versa. The default value of this argument is false, meaning yb_index_check() by default executes in multi-snapshot mode.

===== Operations =====
In the multi-snapshot mode, the count(*) on the base relation is replaced by another LEFT join between the base rel and the index rel. Both these operations serve the same purpose - to detect missing rows from the index. The details of the JOIN are as follows:

Left relation: base relation
Right relation: index relation
Join condition: baserel.computed_index_row_ybctid = indexrel.t_ybindexrowybctid.
Join type: Batched Nested Loop join
Check condition: indexrel.t_ybindexrowybctid is not null (a null value would indicate missing index rows)
Jira: DB-15629

Test Plan:
   ./yb_build.sh --java-test 'org.yb.pgsql.TestPgRegressYbIndexCheck'
   ./yb_build.sh --java-test 'org.yb.pgsql.TestPgRegressYbIndexCheckSingleSnapshot'
   ./yb_build.sh --gtest_filter PgYbIndexCheckTest.YbIndexCheckRepeatableRead
   ./yb_build.sh --gtest_filter PgYbIndexCheckTest.YbIndexCheckSnapshotTooOld

Reviewers: amartsinchyk, kramanathan

Reviewed By: amartsinchyk

Subscribers: smishra, jason, svc_phabricator, yql

Differential Revision: https://phorge.dev.yugabyte.com/D42311

Author: arpang

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgRegressYbIndexCheck.java

@@ -12,6 +12,7 @@
 //
 package org.yb.pgsql;
 
+import java.util.Map;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.yb.YBTestRunner;
@@ -27,7 +28,14 @@ public int getTestMethodTimeoutSec() {
   }
 
   @Test
-  public void testPgRegressFeature() throws Exception {
+  public void testPgRegressYbIndexCheck() throws Exception {
     runPgRegressTest("yb_index_check_schedule");
   }
+
+  @Override
+  protected Map<String, String> getTServerFlags() {
+    Map<String, String> flagMap = super.getTServerFlags();
+    appendToYsqlPgConf(flagMap, "yb_test_index_check_num_batches_per_snapshot=1");
+    return flagMap;
+  }
 }

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgRegressYbIndexCheckSingleSnapshot.java

@@ -0,0 +1,42 @@
+// Copyright (c) YugabyteDB, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+// in compliance with the License.  You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software distributed under the License
+// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied.  See the License for the specific language governing permissions and limitations
+// under the License.
+//
+package org.yb.pgsql;
+
+import java.util.Map;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.yb.YBTestRunner;
+/**
+ * Runs the pg_regress test suite on YB code.
+ */
+@RunWith(value=YBTestRunner.class)
+public class TestPgRegressYbIndexCheckSingleSnapshot extends BasePgRegressTest {
+
+  @Override
+  public int getTestMethodTimeoutSec() {
+    return 1800;
+  }
+
+  @Test
+  public void testPgRegressYbIndexCheckSingleSnapshot() throws Exception {
+    runPgRegressTest("yb_index_check_schedule");
+  }
+
+  @Override
+  protected Map<String, String> getTServerFlags() {
+    Map<String, String> flagMap = super.getTServerFlags();
+    appendToYsqlPgConf(flagMap, "yb_test_index_check_num_batches_per_snapshot=0");
+    return flagMap;
+  }
+
+}

src/postgres/src/backend/access/yb_access/yb_scan.c

@@ -256,6 +256,17 @@ YbBindColumnCondBetween(YbScanDesc ybScan,
 						bool start_valid, bool start_inclusive, Datum value,
 						bool end_valid, bool end_inclusive, Datum value_end)
 {
+	/* Special handling of quals on ybctid column. */
+	if (attnum == YBTupleIdAttributeNumber)
+	{
+		HandleYBStatus(YBCPgDmlBindBounds(ybScan->handle,
+										  start_valid ? value : 0,
+										  start_inclusive,
+										  end_valid ? value_end : 0,
+										  end_inclusive));
+		return;
+	}
+
 	Oid			atttypid = ybc_get_atttypid(bind_desc, attnum);
 	Oid			attcollation = YBEncodingCollation(ybScan->handle, attnum,
 												   ybc_get_attcollation(bind_desc,
@@ -730,21 +741,25 @@ ybcFetchNextIndexTuple(YbScanDesc ybScan, ScanDirection dir)
 				tuple = index_form_tuple(RelationGetDescr(index), ivalues, inulls);
 				if (syscols.ybctid != NULL)
 				{
-					INDEXTUPLE_YBCTID(tuple) = PointerGetDatum(syscols.ybctid);
-					ybcUpdateFKCache(ybScan, INDEXTUPLE_YBCTID(tuple));
+					INDEXTUPLE_BASECTID(tuple) = PointerGetDatum(syscols.ybctid);
+					ybcUpdateFKCache(ybScan, INDEXTUPLE_BASECTID(tuple));
 				}
 			}
 			else
 			{
 				tuple = index_form_tuple(tupdesc, values, nulls);
 				if (syscols.ybbasectid != NULL)
 				{
-					INDEXTUPLE_YBCTID(tuple) = PointerGetDatum(syscols.ybbasectid);
-					ybcUpdateFKCache(ybScan, INDEXTUPLE_YBCTID(tuple));
+					INDEXTUPLE_BASECTID(tuple) = PointerGetDatum(syscols.ybbasectid);
+					ybcUpdateFKCache(ybScan, INDEXTUPLE_BASECTID(tuple));
 				}
+
+				/* Fields used by yb_index_check() */
 				if (syscols.ybuniqueidxkeysuffix != NULL)
 					tuple->t_ybuniqueidxkeysuffix =
 						PointerGetDatum(syscols.ybuniqueidxkeysuffix);
+				if (syscols.ybctid != NULL)
+					tuple->t_ybindexrowybctid = PointerGetDatum(syscols.ybctid);
 			}
 			break;
 		}

src/postgres/src/backend/catalog/yb_system_functions.sql

@@ -115,6 +115,20 @@ LANGUAGE INTERNAL
 VOLATILE STRICT PARALLEL SAFE
 AS 'yb_cancel_query_diagnostics';
 
+CREATE OR REPLACE FUNCTION
+  yb_index_check(indexrelid oid, single_snapshot_mode bool DEFAULT false)
+RETURNS void
+LANGUAGE INTERNAL
+VOLATILE PARALLEL SAFE
+AS 'yb_index_check';
+
+CREATE OR REPLACE FUNCTION
+  yb_compute_row_ybctid(relid oid, key_atts record, ybidxbasectid bytea DEFAULT NULL)
+RETURNS bytea
+LANGUAGE INTERNAL
+IMMUTABLE PARALLEL SAFE
+AS 'yb_compute_row_ybctid';
+
 --
 -- Grant and revoke statements on YB objects.
 --

src/postgres/src/backend/executor/nodeIndexonlyscan.c

@@ -389,8 +389,10 @@ StoreIndexTuple(IndexOnlyScanState *node, TupleTableSlot *slot,
 
 	ExecStoreVirtualTuple(slot);
 
-	TABLETUPLE_YBCTID(slot) = INDEXTUPLE_YBCTID(itup);	/* ybidxbasectid */
-	slot->ts_ybuniqueidxkeysuffix = itup->t_ybuniqueidxkeysuffix;	/* ybuniqueidxkeysuffix */
+	/* Fields used by yb_index_check() */
+	slot->tts_ybidxbasectid = INDEXTUPLE_BASECTID(itup);	/* ybidxbasectid */
+	slot->tts_ybuniqueidxkeysuffix = itup->t_ybuniqueidxkeysuffix;	/* ybuniqueidxkeysuffix */
+	slot->tts_ybctid = itup->t_ybindexrowybctid;	/* index row's ybctid */
 }
 
 /*

src/postgres/src/backend/executor/nodeIndexscan.c

@@ -1409,14 +1409,25 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
 			else
 			{
 				varattno = ((Var *) leftop)->varattno;
-				if (varattno < 1 || varattno > indnkeyatts)
-					elog(ERROR, "bogus index qualification");
 
 				/*
-				 * We have to look up the operator's strategy number.  This
-				 * provides a cross-check that the operator does match the index.
+				 * Special handling for ybctid column. This is currenly used
+				 * only by yb_index_check() which executes indexqual of the
+				 * form 'ybctid > lower_bound'.
 				 */
-				opfamily = index->rd_opfamily[varattno - 1];
+				if (varattno == YBTupleIdAttributeNumber)
+					opfamily = BYTEA_LSM_FAM_OID;
+				else
+				{
+					if (varattno < 1 || varattno > indnkeyatts)
+						elog(ERROR, "bogus index qualification");
+
+					/*
+					 * We have to look up the operator's strategy number.  This
+					 * provides a cross-check that the operator does match the index.
+					 */
+					opfamily = index->rd_opfamily[varattno - 1];
+				}
 			}
 
 			get_op_opfamily_properties(opno, opfamily, isorderby,
@@ -1674,14 +1685,25 @@ ExecIndexBuildScanKeys(PlanState *planstate, Relation index,
 				elog(ERROR, "indexqual doesn't have key on left side");
 
 			varattno = ((Var *) leftop)->varattno;
-			if (varattno < 1 || varattno > indnkeyatts)
-				elog(ERROR, "bogus index qualification");
 
 			/*
-			 * We have to look up the operator's strategy number.  This
-			 * provides a cross-check that the operator does match the index.
+			 * Special handling for ybctid column. This is currenly used only by
+			 * yb_index_check() which executes indexqual of the form:
+			 * 	'ybctid IN (array-expression)'
 			 */
-			opfamily = index->rd_opfamily[varattno - 1];
+			if (varattno == YBTupleIdAttributeNumber)
+				opfamily = BYTEA_LSM_FAM_OID;
+			else
+			{
+				if (varattno < 1 || varattno > indnkeyatts)
+					elog(ERROR, "bogus index qualification");
+
+				/*
+				 * We have to look up the operator's strategy number.  This
+				 * provides a cross-check that the operator does match the index.
+				 */
+				opfamily = index->rd_opfamily[varattno - 1];
+			}
 
 			get_op_opfamily_properties(opno, opfamily, isorderby,
 									   &op_strategy,

src/postgres/src/backend/executor/ybModifyTable.c

@@ -30,6 +30,7 @@
 #include "access/xact.h"
 #include "access/yb_scan.h"
 #include "catalog/catalog.h"
+#include "catalog/heap.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_attribute.h"
 #include "catalog/pg_auth_members_d.h"
@@ -174,24 +175,30 @@ YBCComputeYBTupleIdFromSlot(Relation rel, TupleTableSlot *slot)
 		 * Don't need to fill in for the DocDB RowId column, however we still
 		 * need to add the column to the statement to construct the ybctid.
 		 */
-		if (attnum != YBRowIdAttributeNumber)
+		if (attnum > 0)
 		{
-			Oid			type_id = ((attnum > 0) ?
-								   TupleDescAttr(slot->tts_tupleDescriptor,
-												 attnum - 1)->atttypid :
-								   InvalidOid);
+			Oid			type_id = TupleDescAttr(slot->tts_tupleDescriptor,
+												attnum - 1)->atttypid;
 
 			next_attr->type_entity = YbDataTypeFromOidMod(attnum, type_id);
 			next_attr->collation_id = ybc_get_attcollation(RelationGetDescr(rel), attnum);
 			next_attr->datum = slot_getattr(slot, attnum, &next_attr->is_null);
 		}
-		else
+		else if (attnum == YBRowIdAttributeNumber)
 		{
 			next_attr->datum = 0;
 			next_attr->is_null = false;
 			next_attr->type_entity = NULL;
 			next_attr->collation_id = InvalidOid;
 		}
+		else
+		{
+			Oid			type_id = SystemAttributeDefinition(attnum)->atttypid;
+
+			next_attr->type_entity = YbDataTypeFromOidMod(attnum, type_id);
+			next_attr->collation_id = ybc_get_attcollation(RelationGetDescr(rel), attnum);
+			next_attr->datum = slot_getsysattr(slot, attnum, &next_attr->is_null);
+		}
 		YbcPgColumnInfo column_info = {0};
 
 		HandleYBTableDescStatus(YBCPgGetColumnInfo(ybc_table_desc,

src/postgres/src/backend/utils/misc/guc.c

@@ -3536,6 +3536,18 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"yb_test_slowdown_index_check", PGC_SUSET, DEVELOPER_OPTIONS,
+			gettext_noop("Slows down yb_index_check() by sleeping for 1s after processing "
+						 "every row. Used in tests to simulate long running yb_index_check()."),
+			NULL,
+			GUC_NOT_IN_SAMPLE
+		},
+		&yb_test_slowdown_index_check,
+		false,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"yb_allow_dockey_bounds", PGC_SUSET, CUSTOM_OPTIONS,
 			gettext_noop("If true, allow lower_bound/upper_bound fields of PgsqlReadRequestPB "
@@ -5488,6 +5500,20 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"yb_test_index_check_num_batches_per_snapshot", PGC_USERSET, DEVELOPER_OPTIONS,
+			gettext_noop("Used to test yb_index_check()"),
+			gettext_noop("If set to > 0, number of index rows processed per snapshot "
+						 "is equal to  yb_test_index_check_num_batches_per_snapshot*yb_bnl_batch_size "
+						 "If set to 0, yb_index_check() will execute in single snapshot mode."),
+			GUC_NOT_IN_SAMPLE
+		},
+		&yb_test_index_check_num_batches_per_snapshot,
+		-1,
+		-1,
+		INT_MAX,
+		NULL, NULL, NULL
+	},
 	{
 		{"yb_fk_references_cache_limit", PGC_USERSET, CLIENT_CONN_STATEMENT,
 			gettext_noop("Sets the maximum size for the FK reference cache filled by the INSERT, SELECT ... FOR KEY SHARE or similar statmements"),