You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudberry.apache.org by ro...@apache.org on 2026/10/20 08:56:03 UTC
(cloudberry) branch refactor_ao_parallel_scan created (now 9a1512a4983)
This is an automated email from the ASF dual-hosted git repository.
roseduan pushed a change to branch refactor_ao_parallel_scan
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
at 9a1512a4983 init
This branch includes the following new commits:
new 9a1512a4983 init
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org
(cloudberry) 01/01: init
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roseduan pushed a commit to branch refactor_ao_parallel_scan
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
commit 9a1512a49837310efa05f5c5e3deed34579ef9d3
Author: roseduan <ro...@163.com>
AuthorDate: Tue Oct 20 01:33:22 2026 -0700
init
---
src/backend/access/aocs/aocsam.c | 362 ++++++++++++++++++++++++++++++-
src/backend/access/aocs/aocsam_handler.c | 41 ++--
src/backend/access/table/tableam.c | 13 +-
src/include/access/relscan.h | 74 +++++++
src/include/cdb/cdbaocsam.h | 22 ++
5 files changed, 492 insertions(+), 20 deletions(-)
diff --git a/src/backend/access/aocs/aocsam.c b/src/backend/access/aocs/aocsam.c
index 244d1da808d..e0dec910d7d 100644
--- a/src/backend/access/aocs/aocsam.c
+++ b/src/backend/access/aocs/aocsam.c
@@ -352,24 +352,338 @@ initscan_with_colinfo(AOCSScanDesc scan)
pgstat_count_heap_scan(scan->rs_base.rs_rd);
}
+/* ----------------------------------------------------------------------------
+ * POC: leader-side chunk map construction.
+ *
+ * Divides each segment file into row ranges whose boundaries come from real
+ * block directory entries of an "anchor" column. We deliberately do NOT split
+ * the row number space arithmetically: gp_fastsequence allocates row numbers in
+ * batches of NUM_FAST_SEQUENCES and burns whatever a transaction does not use,
+ * so row number is not a proxy for row ordinal. Measured density is ~99.98%
+ * for a bulk load but drops to ~10% after aborted transactions, which would
+ * give wildly uneven chunks. Taking boundaries straight from the directory is
+ * immune to that by construction.
+ *
+ * Only the anchor column's *directory* is consulted here; its data file is not
+ * opened. Other columns are positioned at scan time by probing the directory
+ * with the same row number (see position_all_columns()), which is legal because
+ * AppendOnlyBlockDirectory_GetEntry() resolves an arbitrary row number to the
+ * entry whose range covers it. Columns therefore never need aligned block
+ * boundaries -- a bigint column packs ~4090 rows/block while an int column
+ * packs ~8181, and both resolve the same row correctly.
+ *
+ * Leaves aopscan->nchunks == 0 if the relation has no block directory, in which
+ * case the scan falls back to the existing segfile-granular behaviour.
+ * ----------------------------------------------------------------------------
+ */
+void
+aocs_parallelscan_buildchunks(Relation rel,
+ Snapshot metaSnapshot,
+ ParallelAOScanDesc aopscan,
+ struct AOCSFileSegInfo **seginfo,
+ int nseg)
+{
+ AppendOnlyBlockDirectory blkdir;
+ AppendOnlyBlockDirectoryEntry entry;
+ Oid blkdirrelid = InvalidOid;
+ bool *proj;
+ int natts = RelationGetNumberOfAttributes(rel);
+ int anchor = 0;
+ int64 total_blocks = 0;
+ int blocks_per_chunk;
+ int nchunks = 0;
+ int i;
+
+ aopscan->nchunks = 0;
+ aopscan->anchor_col = anchor;
+
+ if (nseg <= 0)
+ return;
+
+ GetAppendOnlyEntryAuxOids(rel, NULL, &blkdirrelid, NULL, NULL, NULL);
+ if (!OidIsValid(blkdirrelid))
+ return; /* no block directory -> caller falls back */
+
+ /* We only need the anchor column group's minipages. */
+ proj = (bool *) palloc0(natts * sizeof(bool));
+ proj[anchor] = true;
+
+ AppendOnlyBlockDirectory_Init_forSearch(&blkdir,
+ metaSnapshot,
+ (FileSegInfo **) seginfo,
+ nseg,
+ rel,
+ natts,
+ true /* isAOCol */,
+ proj);
+
+ /*
+ * Pass 1: count blocks so we can size chunks to fit the fixed DSM budget.
+ * Two passes keep leader memory bounded; materialising every boundary
+ * first would cost 16 bytes per block.
+ */
+ for (i = 0; i < nseg; i++)
+ {
+ int64 row = 1;
+
+ for (;;)
+ {
+ AOTupleId tid;
+
+ AOTupleIdInit(&tid, seginfo[i]->segno, row);
+ if (!AppendOnlyBlockDirectory_GetEntry(&blkdir, &tid, anchor, &entry))
+ break;
+
+
+ /*
+ * GetEntry() clamps to the last entry rather than failing when the
+ * probe row is past the end of the segment file, so "the range did
+ * not advance" is our end-of-data condition. It also guarantees
+ * forward progress, which this loop depends on.
+ */
+ if (entry.range.lastRowNum < row)
+ break;
+
+ total_blocks++;
+ row = entry.range.lastRowNum + 1;
+ }
+ }
+
+ if (total_blocks == 0)
+ {
+ AppendOnlyBlockDirectory_End_forSearch(&blkdir);
+ pfree(proj);
+ return;
+ }
+
+ blocks_per_chunk = (int) ((total_blocks + AO_PARALLEL_MAX_CHUNKS - 1) /
+ AO_PARALLEL_MAX_CHUNKS);
+ if (blocks_per_chunk < 1)
+ blocks_per_chunk = 1;
+
+ /* Pass 2: emit a chunk every blocks_per_chunk directory entries. */
+ for (i = 0; i < nseg && nchunks < AO_PARALLEL_MAX_CHUNKS; i++)
+ {
+ int64 row = 1;
+ int64 chunk_start = -1;
+ int64 chunk_last = -1;
+ int nblocks_in_chunk = 0;
+
+ while (nchunks < AO_PARALLEL_MAX_CHUNKS)
+ {
+ AOTupleId tid;
+
+ AOTupleIdInit(&tid, seginfo[i]->segno, row);
+ if (!AppendOnlyBlockDirectory_GetEntry(&blkdir, &tid, anchor, &entry))
+ break;
+
+ if (entry.range.lastRowNum < row)
+ break; /* past end of this segment file */
+
+ if (chunk_start < 0)
+ chunk_start = entry.range.firstRowNum;
+ chunk_last = entry.range.lastRowNum;
+ nblocks_in_chunk++;
+ row = entry.range.lastRowNum + 1;
+
+ if (nblocks_in_chunk >= blocks_per_chunk)
+ {
+ aopscan->chunks[nchunks].segno = seginfo[i]->segno;
+ aopscan->chunks[nchunks].start_row = chunk_start;
+ aopscan->chunks[nchunks].end_row = chunk_last;
+ nchunks++;
+ chunk_start = -1;
+ nblocks_in_chunk = 0;
+ }
+ }
+
+ /* flush a partial trailing chunk for this segfile */
+ if (chunk_start >= 0 && nchunks < AO_PARALLEL_MAX_CHUNKS)
+ {
+ aopscan->chunks[nchunks].segno = seginfo[i]->segno;
+ aopscan->chunks[nchunks].start_row = chunk_start;
+ aopscan->chunks[nchunks].end_row = chunk_last;
+ nchunks++;
+ }
+ }
+
+ AppendOnlyBlockDirectory_End_forSearch(&blkdir);
+ pfree(proj);
+
+ aopscan->nchunks = nchunks;
+
+ elog(DEBUG1,
+ "aocs parallel scan: rel \"%s\" %d segfiles, " INT64_FORMAT
+ " anchor blocks, %d blocks/chunk, %d chunks",
+ RelationGetRelationName(rel), nseg, total_blocks,
+ blocks_per_chunk, nchunks);
+}
+
+/* forward declaration: defined below, used by the POC chunk path */
+static void close_cur_scan_seg(AOCSScanDesc scan);
+
+/*
+ * POC: lazily set up this worker's private block directory handle, used only
+ * to translate a row number into a per-column file offset.
+ */
+static void
+init_positioning_blkdir(AOCSScanDesc scan)
+{
+ Relation rel = scan->rs_base.rs_rd;
+ int natts = RelationGetNumberOfAttributes(rel);
+ bool *proj;
+
+ if (scan->posBlkdir != NULL)
+ return;
+
+ proj = (bool *) palloc0(natts * sizeof(bool));
+ for (AttrNumber i = 0; i < scan->columnScanInfo.num_proj_atts; i++)
+ proj[scan->columnScanInfo.proj_atts[i]] = true;
+
+ scan->posBlkdir = (AppendOnlyBlockDirectory *)
+ palloc0(sizeof(AppendOnlyBlockDirectory));
+
+ AppendOnlyBlockDirectory_Init_forSearch(scan->posBlkdir,
+ scan->appendOnlyMetaDataSnapshot,
+ (FileSegInfo **) scan->seginfo,
+ scan->total_seg,
+ rel,
+ natts,
+ true /* isAOCol */,
+ proj);
+}
+
+/*
+ * POC: position every projected column so that the caller's next
+ * datumstreamread_advance() yields row `targetRow`.
+ *
+ * Each column resolves the SAME row number to its OWN byte offset. Columns do
+ * not share block boundaries -- e.g. in a table with a bigint and an int
+ * column, row 102251 is the first row of the bigint column's block 25 (offset
+ * 819000, no skip) but sits 4078 rows inside the int column's block 12 (offset
+ * 393216). That asymmetry is expected and is exactly what
+ * AppendOnlyBlockDirectory_GetEntry() + datumstreamread_find() handle.
+ */
+static bool
+position_all_columns(AOCSScanDesc scan, int32 segno, int64 targetRow)
+{
+ AppendOnlyBlockDirectoryEntry entry;
+ AOTupleId tid;
+
+ AOTupleIdInit(&tid, segno, targetRow);
+
+ for (AttrNumber i = 0; i < scan->columnScanInfo.num_proj_atts; i++)
+ {
+ AttrNumber attno = scan->columnScanInfo.proj_atts[i];
+ DatumStreamRead *ds = scan->columnScanInfo.ds[attno];
+
+ if (!AppendOnlyBlockDirectory_GetEntry(scan->posBlkdir, &tid, attno, &entry))
+ return false;
+
+ /* Seek the storage reader to the block covering targetRow. */
+ AppendOnlyStorageRead_SetTemporaryRange(&ds->ao_read,
+ entry.range.fileOffset,
+ ds->eof);
+
+ if (datumstreamread_block(ds, NULL, attno) < 0)
+ return false;
+
+ Assert(ds->blockFirstRowNum <= targetRow);
+
+ /*
+ * Skip forward inside the block, stopping one short because the
+ * caller's loop starts with datumstreamread_advance(). find() walks
+ * row by row, but advance() is an inlined pointer bump, so the cost is
+ * ~half a block per column per chunk boundary -- under 1% at the chunk
+ * sizes we generate.
+ */
+ datumstreamread_find(ds, targetRow - ds->blockFirstRowNum - 1);
+ }
+
+ return true;
+}
+
+/*
+ * POC: claim the next row-range chunk. This replaces open_next_scan_seg() on
+ * the parallel path. The atomic fetch-add is the entire load balancing
+ * strategy: because chunks are numerous and uniform, a worker that finishes
+ * early simply takes the next one, so no work stealing is needed.
+ *
+ * Returns the seginfo index positioned for scanning, or -1 when the scan is
+ * exhausted.
+ */
+static int
+claim_next_chunk(AOCSScanDesc scan)
+{
+ ParallelAOScanDesc aopscan = (ParallelAOScanDesc) scan->rs_base.rs_parallel;
+
+
+ init_positioning_blkdir(scan);
+
+ for (;;)
+ {
+ uint64 i;
+ int s;
+
+ i = pg_atomic_fetch_add_u64(&aopscan->cur_chunk, 1);
+ if (i >= (uint64) aopscan->nchunks)
+ return -1;
+
+ scan->cur_chunk = aopscan->chunks[i];
+
+ /* Find the seginfo slot for this chunk's segment file. */
+ for (s = 0; s < scan->total_seg; s++)
+ if (scan->seginfo[s]->segno == scan->cur_chunk.segno)
+ break;
+ if (s >= scan->total_seg)
+ {
+ continue; /* segfile not visible to us; skip the chunk */
+ }
+
+ if (scan->cur_seg != s)
+ {
+ close_cur_scan_seg(scan);
+ scan->cur_seg = s;
+ open_all_datumstreamread_segfiles(scan, scan->seginfo[s]);
+ }
+
+ if (!position_all_columns(scan, scan->cur_chunk.segno,
+ scan->cur_chunk.start_row))
+ {
+ continue;
+ }
+
+ scan->segrowsprocessed = 0;
+ return scan->cur_seg;
+ }
+}
+
static int
open_next_scan_seg(AOCSScanDesc scan)
{
bool isParallel = false;
- ParallelBlockTableScanDesc pbscan = NULL;
+ ParallelAOScanDesc pbscan = NULL;
+
+ /*
+ * POC: when a chunk map is available the unit of work is a row range, not
+ * a whole segment file. Everything below is the legacy segfile-granular
+ * path, kept as the fallback for relations with no block directory.
+ */
+ if (scan->chunked)
+ return claim_next_chunk(scan);
if (scan->rs_base.rs_parallel != NULL)
{
isParallel = true;
- pbscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel;
+ pbscan = (ParallelAOScanDesc) scan->rs_base.rs_parallel;
}
while (scan->cur_seg < scan->total_seg)
{
if (isParallel)
{
- scan->cur_seg = pg_atomic_fetch_add_u64(&pbscan->phs_nallocated, 1);
- if (scan->cur_seg >= pbscan->phs_nblocks)
+ scan->cur_seg = pg_atomic_fetch_add_u64(&pbscan->cur_segfile, 1);
+ if (scan->cur_seg >= pbscan->nsegfiles)
break;
}
else
@@ -600,6 +914,15 @@ aocs_beginscan_internal(Relation relation,
scan->appendOnlyMetaDataSnapshot = appendOnlyMetaDataSnapshot;
scan->seginfo = seginfo;
scan->total_seg = total_seg;
+
+ /*
+ * POC: use block-granular chunks only if the leader managed to build a
+ * chunk map (which requires a block directory). Otherwise fall through to
+ * the existing segfile-granular behaviour.
+ */
+ scan->chunked = (parallel_scan != NULL &&
+ ((ParallelAOScanDesc) parallel_scan)->nchunks > 0);
+ scan->posBlkdir = NULL;
scan->columnScanInfo.scanCtx = CurrentMemoryContext;
/* relationTupleDesc will be inited by the slot when needed */
@@ -683,6 +1006,14 @@ aocs_endscan(AOCSScanDesc scan)
{
close_cur_scan_seg(scan);
+ /* POC: release this worker's positioning block directory handle */
+ if (scan->posBlkdir != NULL)
+ {
+ AppendOnlyBlockDirectory_End_forSearch(scan->posBlkdir);
+ pfree(scan->posBlkdir);
+ scan->posBlkdir = NULL;
+ }
+
if (scan->columnScanInfo.ds)
{
Assert(scan->columnScanInfo.proj_atts);
@@ -1220,6 +1551,29 @@ ReadNext:
rowNum = scan->columnScanInfo.ds[attno]->blockFirstRowNum +
datumstreamread_nth(scan->columnScanInfo.ds[attno]);
}
+
+ /*
+ * POC: enforce this worker's chunk boundary. Setting err = -1
+ * and jumping back routes through the "open next" path, which
+ * dispatches to claim_next_chunk(). Without this the worker
+ * would run to the end of the segment file, i.e. the old
+ * behaviour.
+ */
+ if (scan->chunked &&
+ rowNum != INT64CONST(-1) &&
+ rowNum > scan->cur_chunk.end_row)
+ {
+ /*
+ * rowNum is only recomputed when it is -1, so it MUST be
+ * reset here. Otherwise the next chunk compares this stale
+ * row number against its own end_row, fires the budget
+ * immediately and yields zero rows.
+ */
+ rowNum = INT64CONST(-1);
+ err = -1;
+ goto ReadNext;
+ }
+
scan->segrowsprocessed++;
if (rowNum == INT64CONST(-1))
{
diff --git a/src/backend/access/aocs/aocsam_handler.c b/src/backend/access/aocs/aocsam_handler.c
index 4dc7d541c5f..811f01f7bc8 100644
--- a/src/backend/access/aocs/aocsam_handler.c
+++ b/src/backend/access/aocs/aocsam_handler.c
@@ -780,35 +780,48 @@ aoco_scan_flags(Relation rel)
return SCAN_SUPPORT_COLUMN_ORIENTED_SCAN | SCAN_SUPPORT_VECTORIZATION;
}
+/*
+ * POC: AOCS now uses its own ParallelAOScanDescData rather than borrowing
+ * heap's ParallelBlockTableScanDescData and reinterpreting its fields. The
+ * DSM size must be fixed here because parallelscan_estimate() has no snapshot
+ * and therefore cannot know how many chunks the table will yield.
+ */
static Size
aoco_parallelscan_estimate(Relation rel)
{
- return sizeof(ParallelBlockTableScanDescData);
+ return SizeOfParallelAOScanDesc;
}
-/*
- * AOCO only uses part fields of ParallelBlockTableScanDesc.
- */
static Size
aoco_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan)
{
- ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
+ ParallelAOScanDesc aopscan = (ParallelAOScanDesc) pscan;
+
+ aopscan->base.phs_relid = RelationGetRelid(rel);
+
+ /*
+ * nchunks and nsegfiles need a snapshot to compute (they come from the
+ * aocsseg catalog and the block directory), so they are filled in later by
+ * table_parallelscan_initialize(). nchunks == 0 means "no block directory
+ * available" and the scan falls back to segfile granularity.
+ */
+ aopscan->nchunks = 0;
+ aopscan->anchor_col = 0;
+ aopscan->nsegfiles = 0;
+
+ pg_atomic_init_u64(&aopscan->cur_chunk, 0);
+ pg_atomic_init_u64(&aopscan->cur_segfile, 0);
- bpscan->base.phs_relid = RelationGetRelid(rel);
- bpscan->phs_nblocks = 0; /* init, will be updated later by table_parallelscan_initialize */
- pg_atomic_init_u64(&bpscan->phs_nallocated, 0);
- /* we don't need phs_mutex and phs_startblock in ao, though, init them. */
- SpinLockInit(&bpscan->phs_mutex);
- bpscan->phs_startblock = InvalidBlockNumber;
- return sizeof(ParallelBlockTableScanDescData);
+ return SizeOfParallelAOScanDesc;
}
static void
aoco_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)
{
- ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
+ ParallelAOScanDesc aopscan = (ParallelAOScanDesc) pscan;
- pg_atomic_write_u64(&bpscan->phs_nallocated, 0);
+ pg_atomic_write_u64(&aopscan->cur_chunk, 0);
+ pg_atomic_write_u64(&aopscan->cur_segfile, 0);
}
static IndexFetchTableData *
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index 70cb8cc9cc7..5c20147899a 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -32,6 +32,7 @@
#include "access/aosegfiles.h"
#include "access/aocssegfiles.h"
+#include "cdb/cdbaocsam.h" /* POC: aocs_parallelscan_buildchunks */
/*
* Constants to control the behavior of block allocation to parallel workers
@@ -206,7 +207,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan,
else if (RelationIsAoCols(rel))
{
Snapshot appendOnlyMetaDataSnapshot = snapshot;
- ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
+ ParallelAOScanDesc aopscan = (ParallelAOScanDesc) pscan;
int segfile_count;
AOCSFileSegInfo **seginfo;
if (snapshot == SnapshotAny)
@@ -218,7 +219,15 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan,
appendOnlyMetaDataSnapshot = GetTransactionSnapshot();
}
seginfo = GetAllAOCSFileSegInfo(rel, appendOnlyMetaDataSnapshot, &segfile_count, NULL);
- bpscan->phs_nblocks = segfile_count;
+ aopscan->nsegfiles = segfile_count;
+
+ /*
+ * POC: try to build a block-granular chunk map. If the relation has
+ * no block directory this leaves nchunks == 0 and the scan falls back
+ * to claiming whole segment files, i.e. today's behaviour.
+ */
+ aocs_parallelscan_buildchunks(rel, appendOnlyMetaDataSnapshot,
+ aopscan, seginfo, segfile_count);
if (seginfo)
{
FreeAllAOCSSegFileInfo(seginfo, segfile_count);
diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h
index 2f27e9fc6a2..1c1c29aaebe 100644
--- a/src/include/access/relscan.h
+++ b/src/include/access/relscan.h
@@ -95,6 +95,80 @@ typedef struct ParallelBlockTableScanDescData *ParallelBlockTableScanDesc;
/*
* Per backend state for parallel table scan, for block-oriented storage.
*/
+/* ----------------------------------------------------------------------------
+ * POC: block-granular parallel scan for AO/AOCS.
+ *
+ * Today AO/AOCS reuse ParallelBlockTableScanDescData and reinterpret its
+ * fields so that the unit of parallel work is an entire segment file. That
+ * caps parallelism at the segfile count (and at *zero* when there is only one
+ * segfile) and leaves segfile size skew completely unmitigated.
+ *
+ * This descriptor instead hands out *row ranges*. A row range is the only
+ * workable unit for AOCS: blocks are variable length, so there is no
+ * offset = f(block number), and the N column files of one segfile must stay
+ * row-aligned with each other. Boundaries are derived from the block
+ * directory rather than computed arithmetically, because row numbers are
+ * ordered but NOT contiguous -- gp_fastsequence hands them out in batches of
+ * NUM_FAST_SEQUENCES and burns whatever a transaction does not use.
+ * ----------------------------------------------------------------------------
+ */
+
+/*
+ * One unit of parallel work: rows [start_row, end_row] of segment file segno.
+ * Both bounds are inclusive and are real row numbers taken from block
+ * directory entries, so gaps in the row number space cannot produce an
+ * empty or oversized chunk.
+ */
+typedef struct AOParallelScanChunk
+{
+ int32 segno;
+ int64 start_row;
+ int64 end_row;
+} AOParallelScanChunk;
+
+/*
+ * parallelscan_estimate() is called without a snapshot, so we cannot know the
+ * real chunk count when the DSM segment is sized. Reserve a fixed ceiling
+ * instead; at 24 bytes per chunk this costs ~48kB, which is noise next to the
+ * tuple queues. Chunks are sized to fit this budget (see
+ * aoco_parallelscan_buildchunks).
+ */
+#define AO_PARALLEL_MAX_CHUNKS 2048
+
+typedef struct ParallelAOScanDescData
+{
+ ParallelTableScanDescData base;
+
+ /*
+ * Number of chunks actually populated by the leader. Zero means "no block
+ * directory available" and the scan silently falls back to today's
+ * segfile-granular behaviour, which is what makes this safe to ship
+ * incrementally.
+ */
+ int32 nchunks;
+
+ /*
+ * Column group whose block directory entries defined the chunk boundaries.
+ * Only its *directory* is read; its data file is not touched unless the
+ * query happens to project it.
+ */
+ int32 anchor_col;
+
+ /* Chunk cursor. This is the entire load balancing strategy. */
+ pg_atomic_uint64 cur_chunk;
+
+ /* --- fallback path: segfile-granular, mirrors the current behaviour --- */
+ int32 nsegfiles;
+ pg_atomic_uint64 cur_segfile;
+
+ AOParallelScanChunk chunks[FLEXIBLE_ARRAY_MEMBER];
+} ParallelAOScanDescData;
+typedef struct ParallelAOScanDescData *ParallelAOScanDesc;
+
+#define SizeOfParallelAOScanDesc \
+ (offsetof(ParallelAOScanDescData, chunks) + \
+ AO_PARALLEL_MAX_CHUNKS * sizeof(AOParallelScanChunk))
+
typedef struct ParallelBlockTableScanWorkerData
{
uint64 phsw_nallocated; /* Current # of blocks into the scan */
diff --git a/src/include/cdb/cdbaocsam.h b/src/include/cdb/cdbaocsam.h
index 68c62b55b99..2c3f1c702b4 100644
--- a/src/include/cdb/cdbaocsam.h
+++ b/src/include/cdb/cdbaocsam.h
@@ -252,6 +252,19 @@ typedef struct AOCSScanDescData
int32 total_seg;
int32 cur_seg;
+ /*
+ * POC: block-granular parallel scan state.
+ *
+ * When "chunked" is set this worker claims row ranges out of the shared
+ * ParallelAOScanDesc instead of whole segment files. cur_chunk holds the
+ * range currently being scanned; posBlkdir is this worker's private block
+ * directory handle, used only to translate a row number into a per-column
+ * file offset at chunk boundaries.
+ */
+ bool chunked;
+ AOParallelScanChunk cur_chunk;
+ struct AppendOnlyBlockDirectory *posBlkdir;
+
/*
* The only relation wide Storage Option, the rest are aquired in a per
* column basis and there is no need to keep track of.
@@ -366,6 +379,15 @@ typedef AOCSAddColumnDescData *AOCSAddColumnDesc;
* ----------------
*/
+struct ParallelAOScanDescData;
+
+/* POC: build the parallel scan chunk map from the block directory */
+extern void aocs_parallelscan_buildchunks(Relation rel,
+ Snapshot metaSnapshot,
+ struct ParallelAOScanDescData *aopscan,
+ struct AOCSFileSegInfo **seginfo,
+ int nseg);
+
extern AOCSScanDesc aocs_beginscan(Relation relation, Snapshot snapshot, ParallelTableScanDesc parallel_scan,
bool *proj, uint32 flags);
extern AOCSScanDesc aocs_beginrangescan(Relation relation, Snapshot snapshot,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org