Hi,
On Tue, 2 Apr 2024 at 10:23, Nazir Bilal Yavuz <byavu...@gmail.com> wrote:
>
> v4 is rebased on top of v14 streaming read API changes.
Streaming API has been committed but the committed version has a minor
change, the read_stream_begin_relation function takes Relation instead
of BufferManagerRelation now. So, here is a v5 which addresses this
change.
--
Regards,
Nazir Bilal Yavuz
Microsoft
From 43e2f2b32e2fdb7e1fd787b1d8595768741f4792 Mon Sep 17 00:00:00 2001
From: Nazir Bilal Yavuz <byavu...@gmail.com>
Date: Wed, 3 Apr 2024 01:22:59 +0300
Subject: [PATCH v5] Use streaming read API in ANALYZE
ANALYZE command gets random tuples using BlockSampler algorithm. Use
streaming reads to get these tuples by using BlockSampler algorithm in
streaming read API prefetch logic.
---
src/include/access/heapam.h | 4 +-
src/backend/access/heap/heapam_handler.c | 16 ++---
src/backend/commands/analyze.c | 85 ++++++++----------------
3 files changed, 36 insertions(+), 69 deletions(-)
diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index b632fe953c4..4e35caeb42e 100644
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -25,6 +25,7 @@
#include "storage/bufpage.h"
#include "storage/dsm.h"
#include "storage/lockdefs.h"
+#include "storage/read_stream.h"
#include "storage/shm_toc.h"
#include "utils/relcache.h"
#include "utils/snapshot.h"
@@ -374,8 +375,7 @@ extern bool HeapTupleIsSurelyDead(HeapTuple htup,
/* in heap/heapam_handler.c*/
extern void heapam_scan_analyze_next_block(TableScanDesc scan,
- BlockNumber blockno,
- BufferAccessStrategy bstrategy);
+ ReadStream *stream);
extern bool heapam_scan_analyze_next_tuple(TableScanDesc scan,
TransactionId OldestXmin,
double *liverows, double *deadrows,
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
index c86000d245b..0533d9660c4 100644
--- a/src/backend/access/heap/heapam_handler.c
+++ b/src/backend/access/heap/heapam_handler.c
@@ -1054,16 +1054,15 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap,
}
/*
- * Prepare to analyze block `blockno` of `scan`. The scan has been started
- * with SO_TYPE_ANALYZE option.
+ * Prepare to analyze block returned from streaming object. The scan has been
+ * started with SO_TYPE_ANALYZE option.
*
* This routine holds a buffer pin and lock on the heap page. They are held
* until heapam_scan_analyze_next_tuple() returns false. That is until all the
* items of the heap page are analyzed.
*/
void
-heapam_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
- BufferAccessStrategy bstrategy)
+heapam_scan_analyze_next_block(TableScanDesc scan, ReadStream *stream)
{
HeapScanDesc hscan = (HeapScanDesc) scan;
@@ -1076,11 +1075,12 @@ heapam_scan_analyze_next_block(TableScanDesc scan, BlockNumber blockno,
* doing much work per tuple, the extra lock traffic is probably better
* avoided.
*/
- hscan->rs_cblock = blockno;
- hscan->rs_cindex = FirstOffsetNumber;
- hscan->rs_cbuf = ReadBufferExtended(scan->rs_rd, MAIN_FORKNUM,
- blockno, RBM_NORMAL, bstrategy);
+ hscan->rs_cbuf = read_stream_next_buffer(stream, NULL);
+ Assert(BufferIsValid(hscan->rs_cbuf));
LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_SHARE);
+
+ hscan->rs_cblock = BufferGetBlockNumber(hscan->rs_cbuf);
+ hscan->rs_cindex = FirstOffsetNumber;
}
/*
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
index 2fb39f3ede1..105285c3ea2 100644
--- a/src/backend/commands/analyze.c
+++ b/src/backend/commands/analyze.c
@@ -1102,6 +1102,20 @@ examine_attribute(Relation onerel, int attnum, Node *index_expr)
return stats;
}
+/*
+ * Prefetch callback function to get next block number while using
+ * BlockSampling algorithm
+ */
+static BlockNumber
+block_sampling_streaming_read_next(ReadStream *stream,
+ void *user_data,
+ void *per_buffer_data)
+{
+ BlockSamplerData *bs = user_data;
+
+ return BlockSampler_HasMore(bs) ? BlockSampler_Next(bs) : InvalidBlockNumber;
+}
+
/*
* acquire_sample_rows -- acquire a random sample of rows from the heap
*
@@ -1154,10 +1168,7 @@ acquire_sample_rows(Relation onerel, int elevel,
TableScanDesc scan;
BlockNumber nblocks;
BlockNumber blksdone = 0;
-#ifdef USE_PREFETCH
- int prefetch_maximum = 0; /* blocks to prefetch if enabled */
- BlockSamplerData prefetch_bs;
-#endif
+ ReadStream *stream;
Assert(targrows > 0);
@@ -1170,13 +1181,6 @@ acquire_sample_rows(Relation onerel, int elevel,
randseed = pg_prng_uint32(&pg_global_prng_state);
nblocks = BlockSampler_Init(&bs, totalblocks, targrows, randseed);
-#ifdef USE_PREFETCH
- prefetch_maximum = get_tablespace_maintenance_io_concurrency(onerel->rd_rel->reltablespace);
- /* Create another BlockSampler, using the same seed, for prefetching */
- if (prefetch_maximum)
- (void) BlockSampler_Init(&prefetch_bs, totalblocks, targrows, randseed);
-#endif
-
/* Report sampling block numbers */
pgstat_progress_update_param(PROGRESS_ANALYZE_BLOCKS_TOTAL,
nblocks);
@@ -1187,59 +1191,20 @@ acquire_sample_rows(Relation onerel, int elevel,
scan = heap_beginscan(onerel, NULL, 0, NULL, NULL, SO_TYPE_ANALYZE);
slot = table_slot_create(onerel, NULL);
-#ifdef USE_PREFETCH
-
- /*
- * If we are doing prefetching, then go ahead and tell the kernel about
- * the first set of pages we are going to want. This also moves our
- * iterator out ahead of the main one being used, where we will keep it so
- * that we're always pre-fetching out prefetch_maximum number of blocks
- * ahead.
- */
- if (prefetch_maximum)
- {
- for (int i = 0; i < prefetch_maximum; i++)
- {
- BlockNumber prefetch_block;
-
- if (!BlockSampler_HasMore(&prefetch_bs))
- break;
-
- prefetch_block = BlockSampler_Next(&prefetch_bs);
- PrefetchBuffer(scan->rs_rd, MAIN_FORKNUM, prefetch_block);
- }
- }
-#endif
+ stream = read_stream_begin_relation(READ_STREAM_MAINTENANCE,
+ vac_strategy,
+ scan->rs_rd,
+ MAIN_FORKNUM,
+ block_sampling_streaming_read_next,
+ &bs,
+ 0);
/* Outer loop over blocks to sample */
- while (BlockSampler_HasMore(&bs))
+ while (nblocks)
{
- BlockNumber targblock = BlockSampler_Next(&bs);
-#ifdef USE_PREFETCH
- BlockNumber prefetch_targblock = InvalidBlockNumber;
-
- /*
- * Make sure that every time the main BlockSampler is moved forward
- * that our prefetch BlockSampler also gets moved forward, so that we
- * always stay out ahead.
- */
- if (prefetch_maximum && BlockSampler_HasMore(&prefetch_bs))
- prefetch_targblock = BlockSampler_Next(&prefetch_bs);
-#endif
-
vacuum_delay_point();
- heapam_scan_analyze_next_block(scan, targblock, vac_strategy);
-
-#ifdef USE_PREFETCH
-
- /*
- * When pre-fetching, after we get a block, tell the kernel about the
- * next one we will want, if there's any left.
- */
- if (prefetch_maximum && prefetch_targblock != InvalidBlockNumber)
- PrefetchBuffer(scan->rs_rd, MAIN_FORKNUM, prefetch_targblock);
-#endif
+ heapam_scan_analyze_next_block(scan, stream);
while (heapam_scan_analyze_next_tuple(scan, OldestXmin, &liverows, &deadrows, slot))
{
@@ -1288,7 +1253,9 @@ acquire_sample_rows(Relation onerel, int elevel,
pgstat_progress_update_param(PROGRESS_ANALYZE_BLOCKS_DONE,
++blksdone);
+ nblocks--;
}
+ read_stream_end(stream);
ExecDropSingleTupleTableSlot(slot);
heap_endscan(scan);
--
2.43.0
