fast-analyze: implement fast ANALYZE for append-optimized tables

Prior to this patch, GPDB ANALYZE on large AO/CO tables is a time-consuming process.
This is because PostgreSQL's two-stage sampling method didn't work well on AO/CO tables.
GPDB had to unpack all varblocks till to the target tuples, which could easily result in
almost full table scanning if sampling tuples fall around the end of the table.

Denis Smirnov <sd@picodata.io> 's PR greenplum-db#11190 introduced
a `logical` block concept containing fixed number of tuples to support PG's two-stage sampling
mechanism, also it sped up fetching target tuples by skipping uncompression of varblock content.
Thanks for Denis Smirnov's great contribution!

Also, thanks for Ashwin Agrawal <aashwin@vmware.com> 's advice on leveraging AO Block Directory
to locate the target sample row without scanning unnecessary varblocks, which brings another
significant performance improvement with caching warmed up.

In addition,
- GPDB has AO/CO specific feature that storing total tuple count in an auxiliary table
  which could be easily obtained without too much overhead.
- GPDB has `fetch` facilities support finding varblock based on AOTupleId without
  uncompressing unnecessary varblocks.

Based on above works and properties, we re-implemented AO/CO ANALYZE sampling by combining
Knuth's Algorithm S and varblock skipping in this patch, to address the time-consuming problem.

We didn't impelment two-stage sampling for AO/CO as the total size of data set (total tuple count)
could be known in advance hence Algorithm S is sufficient to satisfy the sampling requirement.

Special thanks Zhenghua Lyu (https://kainwen.com/) for detail analysis of Algorithm S:
[Analysis of Algorithm S](https://kainwen.com/2022/11/06/analysis-of-algorithm-s) and
follow up [discussion](https://stackoverflow.com/questions/74345921/performance-comparsion-algorithm-s-and-algorithm-z?noredirect=1#comment131292564_74345921)

Here is a simple example to show the optimization effect:

[AO with compression, with Fast Analyze enabled]

create table ao (a int, b inet, c inet) with (appendonly=true, orientation=row, compresstype=zlib, compresslevel=3);
insert into ao select i, (select ((i%255)::text || '.' || (i%255)::text || '.' || (i%255)::text || '.' ||
(i%255)::text))::inet, (select ((i%255)::text || '.' || (i%255)::text || '.' || (i%255)::text || '.' ||
(i%255)::text))::inet from generate_series(1,10000000)i;

insert into ao select * from ao;
insert into ao select * from ao;
insert into ao select * from ao;
insert into ao select * from ao;
insert into ao select * from ao;
insert into ao select * from ao;
insert into ao select * from ao;

select count(*) from ao;
   count
------------
 1280000000
(1 row)

gpadmin=# analyze ao;
ANALYZE
Time: 2814.939 ms (00:02.815)
gpadmin=#

[with block directory and caching warmed]

gpadmin=# analyze ao;
ANALYZE
Time: 1605.342 ms (00:01.605)
gpadmin=#

[Legacy Analyze]

gpadmin=# analyze ao;
ANALYZE
Time: 59711.905 ms (00:59.712)
gpadmin=#

[Heap without compression]

create table heap (a int, b inet, c inet);
insert same data set

gpadmin=# analyze heap;
ANALYZE
Time: 2087.694 ms (00:02.088)
gpadmin=#

Co-authored-by: Soumyadeep Chakraborty <soumyadeep2007@gmail.com>
Reviewed by: Ashwin Agrawal, Soumyadeep Chakraborty, Zhenglong Li, Qing Ma

Author: 2 people

src/backend/access/aocs/aocsam.c

@@ -47,6 +47,7 @@
 #include "utils/lsyscache.h"
 #include "utils/pg_rusage.h"
 #include "utils/guc.h"
+#include "utils/sampling.h"
 
 #define IS_BTREE(r) ((r)->rd_rel->relam == BTREE_AM_OID)
 
@@ -1623,39 +1624,92 @@ static bool
 aoco_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
                                    BufferAccessStrategy bstrategy)
 {
-	AOCSScanDesc aoscan = (AOCSScanDesc) scan;
-	aoscan->targetTupleId = blockno;
-
-	return true;
+	/*
+	 * For append-optimized relations, we use a separate sampling
+	 * method. See table_relation_acquire_sample_rows().
+	 */
+	ereport(ERROR,
+			(errcode(ERRCODE_INTERNAL_ERROR),
+			 errmsg("API not supported for appendoptimized relations")));
 }
 
 static bool
 aoco_scan_analyze_next_tuple(TableScanDesc scan, TransactionId OldestXmin,
                                    double *liverows, double *deadrows,
                                    TupleTableSlot *slot)
 {
-	AOCSScanDesc aoscan = (AOCSScanDesc) scan;
-	bool		ret = false;
+	/*
+	 * For append-optimized relations, we use a separate sampling
+	 * method. See table_relation_acquire_sample_rows().
+	 */
+	ereport(ERROR,
+			(errcode(ERRCODE_INTERNAL_ERROR),
+			 errmsg("API not supported for appendoptimized relations")));
+}
 
-	/* skip several tuples if they are not sampling target */
-	while (aoscan->targetTupleId > aoscan->nextTupleId)
-	{
-		aoco_getnextslot(scan, ForwardScanDirection, slot);
-		aoscan->nextTupleId++;
-	}
+static int
+aoco_acquire_sample_rows(Relation onerel, int elevel, HeapTuple *rows,
+						 int targrows, double *totalrows, double *totaldeadrows)
+{
+	int		numrows = 0;	/* # rows now in reservoir */
+	double	liverows = 0;	/* # live rows seen */
+	double	deadrows = 0;	/* # dead rows seen */
 
-	if (aoscan->targetTupleId == aoscan->nextTupleId)
+	Assert(targrows > 0);
+
+	TableScanDesc scan = table_beginscan_analyze(onerel);
+	TupleTableSlot *slot = table_slot_create(onerel, NULL);
+	AOCSScanDesc aocoscan = (AOCSScanDesc) scan;
+
+	int64 totaltupcount = AOCSScanDesc_TotalTupCount(aocoscan);
+	int64 totaldeadtupcount = 0;
+	if (aocoscan->total_seg > 0 )
+		totaldeadtupcount = AppendOnlyVisimap_GetRelationHiddenTupleCount(&aocoscan->visibilityMap);
+	/*
+     * The conversion from int64 to double (53 significant bits) is safe as the
+	 * AOTupleId is 48bits, the max value of totalrows is never greater than
+	 * AOTupleId_MaxSegmentFileNum * AOTupleId_MaxRowNum (< 48 significant bits).
+	 */
+	*totalrows = (double) (totaltupcount - totaldeadtupcount);
+	*totaldeadrows = (double) totaldeadtupcount;
+
+	/* Prepare for sampling tuple numbers */
+	RowSamplerData rs;
+	RowSampler_Init(&rs, *totalrows, targrows, random());
+
+	while (RowSampler_HasMore(&rs))
 	{
-		ret = aoco_getnextslot(scan, ForwardScanDirection, slot);
-		aoscan->nextTupleId++;
+		aocoscan->targrow = RowSampler_Next(&rs);
+
+		vacuum_delay_point();
 
-		if (ret)
-			*liverows += 1;
+		if (aocs_get_target_tuple(aocoscan, aocoscan->targrow, slot))
+		{
+			rows[numrows++] = ExecCopySlotHeapTuple(slot);
+			liverows++;
+		}
 		else
-			*deadrows += 1; /* if return an invisible tuple */
+			deadrows++;
+		
+		ExecClearTuple(slot);
 	}
 
-	return ret;
+	ExecDropSingleTupleTableSlot(slot);
+	table_endscan(scan);
+
+	/*
+	 * Emit some interesting relation info
+	 */
+	ereport(elevel,
+			(errmsg("\"%s\": scanned " INT64_FORMAT " rows, "
+					"containing %.0f live rows and %.0f dead rows; "
+					"%d rows in sample, %.0f accurate total live rows, "
+					"%.f accurate total dead rows",
+					RelationGetRelationName(onerel),
+					rs.m, liverows, deadrows, numrows,
+					*totalrows, *totaldeadrows)));
+
+	return numrows;
 }
 
 static double
@@ -2588,6 +2642,7 @@ static TableAmRoutine ao_column_methods = {
 	.relation_vacuum = aoco_vacuum_rel,
 	.scan_analyze_next_block = aoco_scan_analyze_next_block,
 	.scan_analyze_next_tuple = aoco_scan_analyze_next_tuple,
+	.relation_acquire_sample_rows = aoco_acquire_sample_rows,
 	.index_build_range_scan = aoco_index_build_range_scan,
 	.index_validate_scan = aoco_index_validate_scan,
 
@@ -2602,7 +2657,6 @@ static TableAmRoutine ao_column_methods = {
 	.scan_bitmap_next_tuple = aoco_scan_bitmap_next_tuple,
 	.scan_sample_next_block = aoco_scan_sample_next_block,
 	.scan_sample_next_tuple = aoco_scan_sample_next_tuple,
-	.acquire_sample_rows = acquire_sample_rows,
 
 	.amoptions = ao_amoptions,
 	.swap_relation_files = aoco_swap_relation_files,