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/23 02:40:39 UTC
(cloudberry) branch refactor_ao_parallel_scan updated (bfac733ec78 -> 2618d3dc7f7)
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
discard bfac733ec78 fix
discard aba36947a3c init
add e0289c97c8e Add Arrow and Parquet C++ to the Rocky build images (#2155)
add 707a2abdd70 datalake_fdw: build against the Arrow the Rocky images carry
add 2eb170819e9 Doc: add Backpatch-through trailer to .gitmessage
add be0ee2512c3 Require privilege to call gp_log_backend_memory_contexts()
add f3dda0ec069 Restrict the gpexpand catalog lock and version bump to superusers
add 1991c4d102a Restrict gp_add_master_standby() to superusers
add fa4fddedba1 ORCA: accept coerced join alias vars in USING joins
add 5921f135260 ORCA: project every computed USING join column
new cf9c3b7a8c0 init
new 2618d3dc7f7 fix
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (bfac733ec78)
\
N -- N -- N refs/heads/refactor_ao_parallel_scan (2618d3dc7f7)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 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.
Summary of changes:
.github/workflows/build-cloudberry.yml | 29 +----
.github/workflows/docker-cbdb-build-containers.yml | 6 +-
.gitmessage | 5 +
contrib/datalake_fdw/Makefile | 14 ++-
.../src/format/parquet/parquet_read.cpp | 17 ++-
.../src/format/parquet/parquet_write.cpp | 2 +-
.../isolation2/expected/gpexpand_catalog_lock.out | 38 ++++++
.../test/isolation2/sql/gpexpand_catalog_lock.sql | 25 ++++
devops/deploy/docker/build/rocky10/Dockerfile | 20 ++++
.../tests/testinfra/test_cloudberry_db_env.py | 10 ++
devops/deploy/docker/build/rocky8/Dockerfile | 22 ++++
.../tests/testinfra/test_cloudberry_db_env.py | 10 ++
devops/deploy/docker/build/rocky9/Dockerfile | 20 ++++
.../tests/testinfra/test_cloudberry_db_env.py | 10 ++
src/backend/catalog/system_functions.sql | 24 ++++
.../gpopt/translate/CTranslatorQueryToDXL.cpp | 18 ++-
src/backend/utils/adt/mcxtfuncs.c | 24 ++++
src/backend/utils/gp/segadmin.c | 25 ++--
src/backend/utils/misc/gpexpand.c | 18 ++-
.../isolation2/expected/gpexpand_catalog_lock.out | 38 ++++++
src/test/isolation2/sql/gpexpand_catalog_lock.sql | 25 ++++
src/test/regress/expected/gp_log_mem_dispatch.out | 53 +++++++++
src/test/regress/expected/join_gp.out | 130 +++++++++++++++++++++
src/test/regress/expected/join_gp_optimizer.out | 128 ++++++++++++++++++++
src/test/regress/expected/privileges.out | 26 +++++
src/test/regress/sql/gp_log_mem_dispatch.sql | 40 +++++++
src/test/regress/sql/join_gp.sql | 44 +++++++
src/test/regress/sql/privileges.sql | 25 ++++
.../expected/gpexpand_catalog_lock.out | 38 ++++++
.../sql/gpexpand_catalog_lock.sql | 25 ++++
30 files changed, 860 insertions(+), 49 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org
(cloudberry) 02/02: fix
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 2618d3dc7f7254b6b20b8ae0d79211187e3993fc
Author: roseduan <ro...@gmail.com>
AuthorDate: Fri Oct 23 10:36:18 2026 +0800
fix
---
src/backend/access/aocs/README.parallel_scan.md | 311 +++++++++++++
src/backend/access/aocs/aocsam.c | 558 +++++++++++-------------
src/backend/access/aocs/aocsam_handler.c | 41 +-
src/backend/access/table/tableam.c | 14 +-
src/include/access/relscan.h | 87 +---
src/include/cdb/cdbaocsam.h | 38 +-
6 files changed, 611 insertions(+), 438 deletions(-)
diff --git a/src/backend/access/aocs/README.parallel_scan.md b/src/backend/access/aocs/README.parallel_scan.md
new file mode 100644
index 00000000000..04b648b93d6
--- /dev/null
+++ b/src/backend/access/aocs/README.parallel_scan.md
@@ -0,0 +1,311 @@
+# Parallel sequential scan of AOCS tables
+
+How the parallel workers of one segment share the scan of an AOCS table.
+Segment file and block formats: `src/backend/access/appendonly/README.md`.
+
+
+## 1. Before and after
+
+```
+BEFORE: one worker per segment file
+
+ segfile 1 |==================================================| <- worker A reads it all
+ segfile 2 |==========| <- worker B
+ workers C, D: idle
+
+AFTER: segment files cut into chunks, workers take chunks from one counter
+
+ segfile 1 |-c0-|-c1-|-c2-|-c3-|-c4-|-c5-|-c6-|-c7-|-c8-|-c9-|
+ segfile 2 |-c10-|-c11-|
+ A B C D A C B D A ...
+```
+
+
+## 2. Why not split by block, like heap?
+
+```
+HEAP: one file, fixed 8 KB pages, each page holds whole rows
+
+ |pg 0|pg 1|pg 2|pg 3|pg 4| page N is at N * 8192 -> hand out page ranges
+
+
+AOCS: one file PER COLUMN, variable-length blocks, boundaries differ per column
+
+ column a (int) | blk 0: rows 1-8181 | blk 1: rows 8182-16362 | ...
+ column c (text) | blk 0: 1-5210 | blk 1: 5211-9885 | blk 2: 9886-14560 | ...
+ ^
+ "block 1" is different rows in each column -> no common block unit
+
+ offset of block N = ? (depends on compression and values)
+ row number of row N = ? (gp_fastsequence leaves gaps)
+```
+
+The only coordinate every column file agrees on is the row's position in the
+segment file:
+
+```
+ ordinal: 0 1 2 3 4 ...
+ column a file: a0 a1 a2 a3 a4 ...
+ column b file: b0 b1 b2 b3 b4 ...
+ column c file: c0 c1 c2 c3 c4 ...
+ \_____ row 3 = 3rd value of every column file
+```
+
+
+## 3. Chunk layout
+
+```
+pg_aoseg.pg_aocsseg_<oid> (one row per segment file, read with the scan's snapshot)
+
+ segno | tupcount | state
+ ------+----------+------
+ 1 | 35116 | 1
+ 2 | 66653 | 1
+ 3 | 99877 | 1 AWAITING_DROP or tupcount = 0 -> skipped
+
+rows_per_chunk = max(total_rows / 2048, 16384) = max(201646 / 2048, 16384) = 16384
+
+ ordinal 0 16384 32768 35116
+ segfile 1 |-- c0 --|-- c1 --|-c2-| 3 chunks
+
+ ordinal 0 16384 32768 49152 65536 66653
+ segfile 2 |-- c3 --|-- c4 --|-- c5 --|-- c6 --|c7| 5 chunks
+
+ ordinal 0 16384 81920 98304 99877
+ segfile 3 |-- c8 --|-- c9 --| ... |-- c13 --|c14| 7 chunks
+```
+
+- A chunk never spans two segment files.
+- The last chunk of a segment file is short.
+- `tupcount` includes deleted rows, so chunks cover every row physically stored.
+
+
+## 4. Shared and local state
+
+```
+SHARED: ParallelBlockTableScanDescData (relscan.h, the same struct heap uses)
++--------------------------------------------------------------+
+| base relid, snapshot offset, ... |
+| phs_nblocks 15 <- for AOCS: total number of chunks |
+| phs_nallocated atomic counter: chunks handed out so far |
+| phs_startblock, phs_mutex not used |
++--------------------------------------------------------------+
+
+LOCAL, in every process: AOCSScanDescData (cdbaocsam.h)
++--------------------------------------------------------------+
+| seginfo[] this process's own pg_aocsseg rows |
+| rows_per_chunk 16384 |
+| chunk_first[] 0, 3, 8 (first chunk of seginfo[0..2]) |
+| chunk_seg, chunk_end_row, segrowsprocessed |
+| chunks_claimed, rows_claimed (for the DEBUG1 summary) |
++--------------------------------------------------------------+
+```
+
+Every process reads `pg_aocsseg` with the same snapshot, and
+`GetAllAOCSFileSegInfo()` sorts the rows by `segno`, so every process computes
+the same layout. The shared state holds only the chunk count, which each
+process checks against its own.
+
+
+## 5. Who does what
+
+```
+ FIRST PROCESS (once per segment) EVERY PROCESS
+ -------------------------------- -------------
+ExecSeqScanInitializeDSM()
+ table_parallelscan_initialize()
+ aoco_parallelscan_initialize() reset descriptor
+ GetAllAOCSFileSegInfo(snapshot) read pg_aocsseg
+ phs_nblocks = aocs_parallelscan_nchunks()
+ |
+ | shared memory aocs_beginscan()
+ v GetAllAOCSFileSegInfo(snapshot)
+ +------------------+ aocs_parallelscan_init_layout()
+ | phs_nblocks | <---- check ---------- compute chunk_first[] locally
+ | phs_nallocated | <-------------+
+ +------------------+ | aocs_getnext*()
+ | open_next_scan_seg()
+ +------- claim_next_chunk()
+ fetch_add(phs_nallocated)
+ open segfile if needed
+ aocs_skip_rows()
+ scan rows until aocs_chunk_done()
+```
+
+
+## 6. Claiming a chunk
+
+```
+chunk = fetch_add(phs_nallocated, 1) = 5 (>= phs_nblocks -> scan finished)
+
+chunk_first[]: 0 3 8 (local, one entry per seginfo[])
+ | | |
+ s1 s2 s3 binary search: 3 <= 5 < 8 -> seginfo[1], segfile 2
+
+start = (5 - 3) * 16384 = 32768
+end = min(32768 + 16384, 66653) = 49152
+
+-> scan ordinals [32768, 49152) of segfile 2
+```
+
+Empty and AWAITING_DROP segment files get no chunks, so their entry in
+`chunk_first[]` equals the next one, and the search (last entry not after the
+chunk number) never picks them.
+
+
+## 7. Finding the start of a chunk in each column file
+
+Each block header is never compressed and stores the block's row count and
+length:
+
+```
+ one block
+ +----------------------------------+---------------------------------------+
+ | header: rowCount, length, | content: values, maybe compressed |
+ | compressedLength | (zstd / zlib / lz4 / rle_type) |
+ +----------------------------------+---------------------------------------+
+ read only this to skip a block decompressed only for the target block
+```
+
+Positioning every projected column at ordinal 32768 (`aocs_skip_rows()`):
+
+```
+column a (int)
+ rows: 0-8180 8181-16361 16362-24542 24543-32723 32724-40904
+ | blk 0 | blk 1 | blk 2 | blk 3 | blk 4 | ...
+ header header header header header
+ skip skip skip skip LOAD, skip 44 values inside
+ ^ 32768
+
+column c (text)
+ rows: 0-2399 2400-4799 ... 28800-31199 31200-33599
+ | b0 | b1 | ... | b12 | b13 | ...
+ skip skip ... skip LOAD, skip 1568 values inside
+ ^ 32768
+
+=> next datumstreamread_advance() on every column returns row 32768
+```
+
+```
+for each projected column:
+ left = start - rows already consumed
+ +-> is the target in the current block? -- yes --> datumstreamread_find()
+ | | no
+ | read next header (datumstreamread_block_info)
+ | target in this block? -- yes --> decompress it (datumstreamread_block_content)
+ | | no then datumstreamread_find()
+ | left -= rowCount
+ +-- jump over the content (AppendOnlyStorageRead_SkipCurrentBlock)
+```
+
+
+## 8. A worker only moves forward
+
+```
+worker A claims c0, then c4, then c9 (other workers took the ones in between)
+
+segfile 2 |-- c3 --|-- c4 --|-- c5 --|-- c6 --|c7|
+ [ scan ] skip skip
+ ^ position kept in segrowsprocessed
+
+segfile 3 |-- c8 --|-- c9 --| ...
+ skip [ scan ]
+
+- chunk numbers only increase -> the next chunk is always ahead
+- each column file's headers are walked at most once per worker
+- new segfile -> open it, start from its first block
+```
+
+
+## 9. Ending a chunk
+
+```
+top of every scan loop (aocs_getnext_noqual / _withqual / _sample):
+
+ +------------------------------------------------+
+ | cur_seg < 0 OR err < 0 OR aocs_chunk_done() | --yes--> open_next_scan_seg()
+ +------------------------------------------------+ -> claim_next_chunk()
+ | no
+ v
+ advance every column by one row
+ segrowsprocessed++
+ visible in visimap? -- no --> next row
+ | yes
+ v
+ return row
+
+aocs_chunk_done(): parallel scan AND segrowsprocessed >= chunk_end_row
+```
+
+
+## 10. Deleted rows, row numbers, compression
+
+```
+DELETE: data and tupcount unchanged, row marked in the visibility map
+
+ chunk: |------------------ c5 ------------------|
+ rows: r r x r x x x r r r x r r r x = deleted
+ scanned like any row, dropped by the visimap check
+
+ROW NUMBERS: used after positioning, never for positioning
+
+ blk 0: firstRowNum 1, rowCount 100 -> ordinals 0-99, rownums 1-100
+ blk 1: firstRowNum 201, rowCount 100 -> ordinals 100-199, rownums 201-300
+ (gap 101-200: aborted insert)
+ ctid / visimap lookup use rownum = blk.firstRowNum + position in block
+
+COMPRESSION: headers are plain, content is compressed
+
+ skipped block -> header only, never decompressed
+ target block -> decompressed once
+ rle_type -> more rows per block, longer skip inside the target block
+ large value -> spans several blocks, counts as one row, skipped as a unit
+```
+
+
+## 11. Limitations
+
+```
+Skipping still reads the bytes (no seek):
+
+ worker A |>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>| reads ~ whole column file
+ worker B |>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>|
+ worker C |>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>| => ~ W x file bytes in total
+ worker D |>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>| (page cache if it fits in RAM)
+
+With a block directory (table has an index), a worker could seek instead:
+
+ pg_aoblkdir_<oid>: (segno, column, first_row_no) -> minipage
+ minipage entry {firstRowNum, fileOffset, rowCount}
+
+ GetEntry(segno, rownum, col) -> fileOffset -> seek -> load one block
+ (keyed by row NUMBER, so chunk boundaries would come from directory entries)
+```
+
+- One chunk per claim; no smaller chunks near the end of the scan (heap does both).
+- AO row tables still split by whole segment file.
+- Building a block directory during a scan (`scan->blockDirectory`) is not supported in parallel.
+
+
+## 12. Code map
+
+```
+src/include/access/relscan.h ParallelBlockTableScanDescData: AOCS meaning of
+ phs_nblocks and phs_nallocated (comment only)
+src/include/cdb/cdbaocsam.h AOCSScanDescData: rows_per_chunk, chunk_first,
+ chunk_seg, chunk_end_row, chunks_claimed, rows_claimed
+ aocs_parallelscan_nchunks() declaration
+
+src/backend/access/table/tableam.c
+ table_parallelscan_initialize(): phs_nblocks = aocs_parallelscan_nchunks()
+
+src/backend/access/aocs/aocsam.c
+ AOCS_PARALLEL_SCAN_NCHUNKS (2048), AOCS_PARALLEL_SCAN_MIN_CHUNK_ROWS (16384)
+ aocs_parallelscan_layout() chunk layout from seginfo[] (used by all)
+ aocs_parallelscan_nchunks() first process: number of chunks
+ aocs_parallelscan_init_layout() every process: local layout + check
+ claim_next_chunk() take a chunk, open segfile
+ aocs_skip_rows() position every column
+ aocs_block_rows_consumed()
+ aocs_chunk_done() checked in the three aocs_getnext* loops
+```
diff --git a/src/backend/access/aocs/aocsam.c b/src/backend/access/aocs/aocsam.c
index e0dec910d7d..230a145ed61 100644
--- a/src/backend/access/aocs/aocsam.c
+++ b/src/backend/access/aocs/aocsam.c
@@ -343,6 +343,7 @@ initscan_with_colinfo(AOCSScanDesc scan)
MemoryContextSwitchTo(oldCtx);
scan->cur_seg = -1;
+ scan->chunk_seg = -1;
scan->segrowsprocessed = 0;
ItemPointerSet(&scan->cdb_fake_ctid, 0, 0);
@@ -353,345 +354,320 @@ initscan_with_colinfo(AOCSScanDesc scan)
}
/* ----------------------------------------------------------------------------
- * POC: leader-side chunk map construction.
+ * Parallel scan support.
*
- * 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.
+ * Work is handed out in chunks. A chunk is a range of row ordinals within one
+ * segment file, where the ordinal of a row is its position among the rows
+ * physically stored in that file. Ordinals are the only coordinate that all
+ * column files of a segment file agree on: blocks have different boundaries
+ * in each column file, and row numbers have gaps.
*
- * 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.
+ * The chunk layout follows from the segment file tuple counts alone. Every
+ * process of the scan reads the same segment files with the same snapshot,
+ * so every process computes the same layout from its own seginfo[], and the
+ * shared ParallelBlockTableScanDesc only holds the number of chunks
+ * (phs_nblocks) and the number handed out so far (phs_nallocated). Each
+ * process positions its column readers at the first row of a chunk by
+ * reading block headers, so no block directory is needed.
* ----------------------------------------------------------------------------
*/
-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;
+/*
+ * Target number of chunks per scan, and the smallest chunk worth handing out.
+ * As with PARALLEL_SEQSCAN_NCHUNKS for heap, there should be enough chunks for
+ * the workers to finish at about the same time, but few enough that the cost
+ * of positioning at the first row of each chunk stays small.
+ */
+#define AOCS_PARALLEL_SCAN_NCHUNKS 2048
+#define AOCS_PARALLEL_SCAN_MIN_CHUNK_ROWS 16384
- 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);
+static void close_cur_scan_seg(AOCSScanDesc scan);
+static inline bool
+aocs_segfile_has_rows(AOCSFileSegInfo *seginfo)
+{
/*
- * 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.
+ * subtle: we must check for AWAITING_DROP before looking at vpinfo, see
+ * open_next_scan_seg().
*/
- 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;
+ return seginfo->total_tupcount > 0 &&
+ seginfo->state != AOSEG_STATE_AWAITING_DROP;
+}
- total_blocks++;
- row = entry.range.lastRowNum + 1;
- }
- }
+/*
+ * Compute the chunk layout of a parallel scan, and return the number of
+ * chunks.
+ *
+ * *rows_per_chunk and *total_rows are set. If first_chunk is not NULL,
+ * first_chunk[i] is set to the number of the first chunk of seginfo[i].
+ * Empty and AWAITING_DROP segment files get no chunks, so their first chunk
+ * equals that of the next segment file. claim_next_chunk() picks the last
+ * segment file whose first chunk is not after the chunk number, so it never
+ * picks them.
+ */
+static uint64
+aocs_parallelscan_layout(AOCSFileSegInfo **seginfo, int total_seg,
+ uint64 *first_chunk, int64 *rows_per_chunk,
+ int64 *total_rows)
+{
+ int64 rows = 0;
+ int64 chunk_rows;
+ uint64 nchunks = 0;
- if (total_blocks == 0)
+ for (int i = 0; i < total_seg; i++)
{
- AppendOnlyBlockDirectory_End_forSearch(&blkdir);
- pfree(proj);
- return;
+ if (aocs_segfile_has_rows(seginfo[i]))
+ rows += seginfo[i]->total_tupcount;
}
- blocks_per_chunk = (int) ((total_blocks + AO_PARALLEL_MAX_CHUNKS - 1) /
- AO_PARALLEL_MAX_CHUNKS);
- if (blocks_per_chunk < 1)
- blocks_per_chunk = 1;
+ chunk_rows = Max(rows / AOCS_PARALLEL_SCAN_NCHUNKS,
+ AOCS_PARALLEL_SCAN_MIN_CHUNK_ROWS);
- /* Pass 2: emit a chunk every blocks_per_chunk directory entries. */
- for (i = 0; i < nseg && nchunks < AO_PARALLEL_MAX_CHUNKS; i++)
+ for (int i = 0; i < total_seg; 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;
+ int64 tupcount;
- AOTupleIdInit(&tid, seginfo[i]->segno, row);
- if (!AppendOnlyBlockDirectory_GetEntry(&blkdir, &tid, anchor, &entry))
- break;
+ tupcount = aocs_segfile_has_rows(seginfo[i]) ? seginfo[i]->total_tupcount : 0;
- if (entry.range.lastRowNum < row)
- break; /* past end of this segment file */
+ if (first_chunk != NULL)
+ first_chunk[i] = nchunks;
+ nchunks += (tupcount + chunk_rows - 1) / chunk_rows;
+ }
- if (chunk_start < 0)
- chunk_start = entry.range.firstRowNum;
- chunk_last = entry.range.lastRowNum;
- nblocks_in_chunk++;
- row = entry.range.lastRowNum + 1;
+ *rows_per_chunk = chunk_rows;
+ *total_rows = rows;
+ return nchunks;
+}
- 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;
- }
- }
+/*
+ * Return the number of chunks of a parallel scan.
+ *
+ * Called from table_parallelscan_initialize() by the process that initializes
+ * the shared state, which stores the result in phs_nblocks.
+ */
+BlockNumber
+aocs_parallelscan_nchunks(Relation rel, AOCSFileSegInfo **seginfo,
+ int total_seg)
+{
+ int64 rows_per_chunk;
+ int64 total_rows;
+ uint64 nchunks;
- /* 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++;
- }
- }
+ nchunks = aocs_parallelscan_layout(seginfo, total_seg, NULL,
+ &rows_per_chunk, &total_rows);
- AppendOnlyBlockDirectory_End_forSearch(&blkdir);
- pfree(proj);
+ if (nchunks > MaxBlockNumber)
+ elog(ERROR, "too many chunks for parallel scan of relation \"%s\"",
+ RelationGetRelationName(rel));
- aopscan->nchunks = nchunks;
+ elog(DEBUG1, "parallel scan of AOCS relation \"%s\": %d segment files, "
+ INT64_FORMAT " rows, " INT64_FORMAT " rows per chunk, " UINT64_FORMAT " chunks",
+ RelationGetRelationName(rel), total_seg, total_rows, rows_per_chunk,
+ 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);
+ return (BlockNumber) 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.
+ * Compute this process's copy of the chunk layout, and check that it agrees
+ * with the number of chunks in the shared state.
*/
static void
-init_positioning_blkdir(AOCSScanDesc scan)
+aocs_parallelscan_init_layout(AOCSScanDesc scan)
{
- Relation rel = scan->rs_base.rs_rd;
- int natts = RelationGetNumberOfAttributes(rel);
- bool *proj;
+ ParallelBlockTableScanDesc bpscan =
+ (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel;
+ int64 total_rows;
+ uint64 nchunks;
- if (scan->posBlkdir != NULL)
- return;
+ scan->chunk_first = (uint64 *) palloc(Max(scan->total_seg, 1) * sizeof(uint64));
+ nchunks = aocs_parallelscan_layout(scan->seginfo, scan->total_seg,
+ scan->chunk_first, &scan->rows_per_chunk,
+ &total_rows);
- 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);
+ /*
+ * All processes of the scan use the same snapshot, so they must see the
+ * same segment files. Otherwise rows would be skipped or returned twice.
+ */
+ if (nchunks != bpscan->phs_nblocks)
+ elog(ERROR, "parallel scan of relation \"%s\" has " UINT64_FORMAT " chunks in this process, but %u in the shared state",
+ RelationGetRelationName(scan->rs_base.rs_rd), nchunks,
+ bpscan->phs_nblocks);
}
/*
- * 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.
+ * Has the current chunk of a parallel scan been read completely?
*/
-static bool
-position_all_columns(AOCSScanDesc scan, int32 segno, int64 targetRow)
+static inline bool
+aocs_chunk_done(AOCSScanDesc scan)
{
- AppendOnlyBlockDirectoryEntry entry;
- AOTupleId tid;
+ return scan->rs_base.rs_parallel != NULL &&
+ scan->segrowsprocessed >= scan->chunk_end_row;
+}
- AOTupleIdInit(&tid, segno, targetRow);
+/*
+ * Return the number of rows of the current block of a datum stream that have
+ * already been consumed.
+ */
+static inline int64
+aocs_block_rows_consumed(DatumStreamRead *ds)
+{
+ switch (ds->largeObjectState)
+ {
+ case DatumStreamLargeObjectState_None:
+ return DatumStreamBlockRead_Nth(&ds->blockRead) + 1;
+ case DatumStreamLargeObjectState_HaveAoContent:
+ return 0;
+ default:
+ /* a large object block holds exactly one row */
+ return 1;
+ }
+}
+
+/*
+ * Move every projected column forward by 'nrows' rows, so that the next
+ * datumstreamread_advance() on each of them returns the same row.
+ *
+ * Whole blocks are skipped by reading only their headers, as aocs_gettuple()
+ * does, so skipped blocks are never decompressed. Each column file has its
+ * own block boundaries; an int column holds about twice as many rows per
+ * block as a bigint column.
+ */
+static void
+aocs_skip_rows(AOCSScanDesc scan, int64 nrows)
+{
+ if (nrows == 0)
+ return;
for (AttrNumber i = 0; i < scan->columnScanInfo.num_proj_atts; i++)
{
AttrNumber attno = scan->columnScanInfo.proj_atts[i];
DatumStreamRead *ds = scan->columnScanInfo.ds[attno];
+ int64 consumed = aocs_block_rows_consumed(ds);
+ int64 left = nrows;
- if (!AppendOnlyBlockDirectory_GetEntry(scan->posBlkdir, &tid, attno, &entry))
- return false;
+ if (left >= ds->blockRowCount - consumed)
+ {
+ /* the target row is past the current block */
+ left -= ds->blockRowCount - consumed;
- /* Seek the storage reader to the block covering targetRow. */
- AppendOnlyStorageRead_SetTemporaryRange(&ds->ao_read,
- entry.range.fileOffset,
- ds->eof);
+ for (;;)
+ {
+ if (!datumstreamread_block_info(ds))
+ elog(ERROR, "unexpected end of segment file %d, column %d, of relation \"%s\" in parallel scan",
+ scan->seginfo[scan->cur_seg]->segno, attno + 1,
+ RelationGetRelationName(scan->rs_base.rs_rd));
- if (datumstreamread_block(ds, NULL, attno) < 0)
- return false;
+ if (left < ds->blockRowCount)
+ break;
- Assert(ds->blockFirstRowNum <= targetRow);
+ left -= ds->blockRowCount;
+ AppendOnlyStorageRead_SkipCurrentBlock(&ds->ao_read);
+ }
+
+ datumstreamread_block_content(ds);
+ AOCSScanDesc_UpdateTotalBytesRead(scan, attno);
+ consumed = 0;
+ }
/*
- * 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.
+ * Skip forward inside the block. Stop one row short, because the
+ * caller's next datumstreamread_advance() returns the target row.
*/
- datumstreamread_find(ds, targetRow - ds->blockFirstRowNum - 1);
+ if (left > 0)
+ {
+ Assert(ds->largeObjectState == DatumStreamLargeObjectState_None);
+ datumstreamread_find(ds, consumed + left - 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.
+ * Claim the next chunk of a parallel scan and position the column readers at
+ * its first row. open_next_scan_seg() calls this for parallel scans.
*
- * Returns the seginfo index positioned for scanning, or -1 when the scan is
- * exhausted.
+ * Returns the index in seginfo[] of the chunk's segment file, or -1 if all
+ * chunks have been handed out.
+ *
+ * Chunks are numbered in segment file order and phs_nallocated only grows, so
+ * a process's next chunk in the same segment file is always after the point
+ * where it stopped. A process therefore only moves forward, and reads the
+ * block headers of each column file at most once.
*/
static int
claim_next_chunk(AOCSScanDesc scan)
{
- ParallelAOScanDesc aopscan = (ParallelAOScanDesc) scan->rs_base.rs_parallel;
-
+ ParallelBlockTableScanDesc bpscan =
+ (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel;
+ uint64 chunk;
+ int64 startrow;
+ int seg;
+ int lo;
+ int hi;
- init_positioning_blkdir(scan);
+ /* building a block directory needs to see every block, in order */
+ Assert(scan->blockDirectory == NULL);
- for (;;)
+ chunk = pg_atomic_fetch_add_u64(&bpscan->phs_nallocated, 1);
+ if (chunk >= bpscan->phs_nblocks)
{
- uint64 i;
- int s;
+ close_cur_scan_seg(scan);
+ return -1;
+ }
- i = pg_atomic_fetch_add_u64(&aopscan->cur_chunk, 1);
- if (i >= (uint64) aopscan->nchunks)
- return -1;
+ /* find the last segment file whose first chunk is not after this one */
+ lo = 0;
+ hi = scan->total_seg - 1;
+ while (lo < hi)
+ {
+ int mid = (lo + hi + 1) / 2;
- scan->cur_chunk = aopscan->chunks[i];
+ if (scan->chunk_first[mid] <= chunk)
+ lo = mid;
+ else
+ hi = mid - 1;
+ }
+ seg = lo;
- /* 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 */
- }
+ startrow = (int64) (chunk - scan->chunk_first[seg]) * scan->rows_per_chunk;
+ scan->chunk_end_row = Min(startrow + scan->rows_per_chunk,
+ scan->seginfo[seg]->total_tupcount);
+ scan->chunks_claimed++;
+ scan->rows_claimed += scan->chunk_end_row - startrow;
- if (scan->cur_seg != s)
- {
- close_cur_scan_seg(scan);
- scan->cur_seg = s;
- open_all_datumstreamread_segfiles(scan, scan->seginfo[s]);
- }
+ if (scan->chunk_seg != seg || startrow < scan->segrowsprocessed)
+ {
+ close_cur_scan_seg(scan);
- if (!position_all_columns(scan, scan->cur_chunk.segno,
- scan->cur_chunk.start_row))
- {
- continue;
- }
+ if (getAOCSVPEntry(scan->seginfo[seg], scan->columnScanInfo.proj_atts[0])->eof == 0)
+ elog(ERROR, "inconsistent segment state for relation %s, segment %d, tuple count " INT64_FORMAT,
+ RelationGetRelationName(scan->rs_base.rs_rd),
+ scan->seginfo[seg]->segno,
+ scan->seginfo[seg]->total_tupcount);
+ scan->cur_seg = seg;
+ open_all_datumstreamread_segfiles(scan, scan->seginfo[seg]);
+ scan->chunk_seg = seg;
scan->segrowsprocessed = 0;
- return scan->cur_seg;
}
+
+ aocs_skip_rows(scan, startrow - scan->segrowsprocessed);
+ scan->segrowsprocessed = startrow;
+
+ return seg;
}
static int
open_next_scan_seg(AOCSScanDesc scan)
{
- bool isParallel = false;
- 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 = (ParallelAOScanDesc) scan->rs_base.rs_parallel;
- }
+ return claim_next_chunk(scan);
while (scan->cur_seg < scan->total_seg)
{
- if (isParallel)
- {
- scan->cur_seg = pg_atomic_fetch_add_u64(&pbscan->cur_segfile, 1);
- if (scan->cur_seg >= pbscan->nsegfiles)
- break;
- }
- else
- {
- scan->cur_seg = scan->cur_seg + 1;
- if (scan->cur_seg >= scan->total_seg)
- break;
- }
+ scan->cur_seg = scan->cur_seg + 1;
+ if (scan->cur_seg >= scan->total_seg)
+ break;
AOCSFileSegInfo *curSegInfo = scan->seginfo[scan->cur_seg];
@@ -745,6 +721,7 @@ open_next_scan_seg(AOCSScanDesc scan)
}
open_all_datumstreamread_segfiles(scan, curSegInfo);
+ scan->segrowsprocessed = 0;
return scan->cur_seg;
}
@@ -777,6 +754,8 @@ close_cur_scan_seg(AOCSScanDesc scan)
if (scan->blockDirectory)
AppendOnlyBlockDirectory_End_forInsert(scan->blockDirectory);
+
+ scan->chunk_seg = -1;
}
static void
@@ -914,15 +893,9 @@ 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->chunk_seg = -1;
+ if (parallel_scan != NULL)
+ aocs_parallelscan_init_layout(scan);
scan->columnScanInfo.scanCtx = CurrentMemoryContext;
/* relationTupleDesc will be inited by the slot when needed */
@@ -1004,15 +977,16 @@ aocs_rescan(AOCSScanDesc scan)
void
aocs_endscan(AOCSScanDesc scan)
{
+ if (scan->rs_base.rs_parallel != NULL)
+ elog(DEBUG1, "parallel scan of AOCS relation \"%s\": this process claimed "
+ INT64_FORMAT " chunks, " INT64_FORMAT " rows",
+ RelationGetRelationName(scan->rs_base.rs_rd),
+ scan->chunks_claimed, scan->rows_claimed);
+
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->chunk_first)
+ pfree(scan->chunk_first);
if (scan->columnScanInfo.ds)
{
@@ -1492,7 +1466,7 @@ aocs_getnext_noqual(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot *
ReadNext:
/* If necessary, open next seg */
- if (scan->cur_seg < 0 || err < 0)
+ if (scan->cur_seg < 0 || err < 0 || aocs_chunk_done(scan))
{
err = open_next_scan_seg(scan);
if (err < 0)
@@ -1502,7 +1476,6 @@ ReadNext:
scan->cur_seg = -1;
return false;
}
- scan->segrowsprocessed = 0;
}
Assert(scan->cur_seg >= 0);
@@ -1551,29 +1524,6 @@ 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))
{
@@ -1662,7 +1612,7 @@ aocs_getnext_withqual(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot
ReadNext:
/* If necessary, open next seg */
- if (scan->cur_seg < 0 || err < 0)
+ if (scan->cur_seg < 0 || err < 0 || aocs_chunk_done(scan))
{
err = open_next_scan_seg(scan);
if (err < 0)
@@ -1672,7 +1622,6 @@ ReadNext:
scan->cur_seg = -1;
return false;
}
- scan->segrowsprocessed = 0;
}
Assert(scan->cur_seg >= 0);
@@ -3399,7 +3348,7 @@ aocs_getnext_sample(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot *
ReadNext:
/* If necessary, open next seg */
- if (scan->cur_seg < 0 || err < 0)
+ if (scan->cur_seg < 0 || err < 0 || aocs_chunk_done(scan))
{
err = open_next_scan_seg(scan);
if (err < 0)
@@ -3409,7 +3358,6 @@ ReadNext:
scan->cur_seg = -1;
return false;
}
- scan->segrowsprocessed = 0;
}
Assert(scan->cur_seg >= 0);
diff --git a/src/backend/access/aocs/aocsam_handler.c b/src/backend/access/aocs/aocsam_handler.c
index 811f01f7bc8..4dc7d541c5f 100644
--- a/src/backend/access/aocs/aocsam_handler.c
+++ b/src/backend/access/aocs/aocsam_handler.c
@@ -780,48 +780,35 @@ 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 SizeOfParallelAOScanDesc;
+ return sizeof(ParallelBlockTableScanDescData);
}
+/*
+ * AOCO only uses part fields of ParallelBlockTableScanDesc.
+ */
static Size
aoco_parallelscan_initialize(Relation rel, ParallelTableScanDesc 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);
+ ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
- return SizeOfParallelAOScanDesc;
+ 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);
}
static void
aoco_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)
{
- ParallelAOScanDesc aopscan = (ParallelAOScanDesc) pscan;
+ ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
- pg_atomic_write_u64(&aopscan->cur_chunk, 0);
- pg_atomic_write_u64(&aopscan->cur_segfile, 0);
+ pg_atomic_write_u64(&bpscan->phs_nallocated, 0);
}
static IndexFetchTableData *
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
index 5c20147899a..0c0441c36a0 100644
--- a/src/backend/access/table/tableam.c
+++ b/src/backend/access/table/tableam.c
@@ -32,7 +32,7 @@
#include "access/aosegfiles.h"
#include "access/aocssegfiles.h"
-#include "cdb/cdbaocsam.h" /* POC: aocs_parallelscan_buildchunks */
+#include "cdb/cdbaocsam.h"
/*
* Constants to control the behavior of block allocation to parallel workers
@@ -207,7 +207,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan,
else if (RelationIsAoCols(rel))
{
Snapshot appendOnlyMetaDataSnapshot = snapshot;
- ParallelAOScanDesc aopscan = (ParallelAOScanDesc) pscan;
+ ParallelBlockTableScanDesc bpscan = (ParallelBlockTableScanDesc) pscan;
int segfile_count;
AOCSFileSegInfo **seginfo;
if (snapshot == SnapshotAny)
@@ -219,15 +219,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan,
appendOnlyMetaDataSnapshot = GetTransactionSnapshot();
}
seginfo = GetAllAOCSFileSegInfo(rel, appendOnlyMetaDataSnapshot, &segfile_count, NULL);
- 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);
+ bpscan->phs_nblocks = aocs_parallelscan_nchunks(rel, seginfo, segfile_count);
if (seginfo)
{
FreeAllAOCSSegFileInfo(seginfo, segfile_count);
diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h
index 1c1c29aaebe..6ddbf1d6f6e 100644
--- a/src/include/access/relscan.h
+++ b/src/include/access/relscan.h
@@ -84,10 +84,15 @@ typedef struct ParallelBlockTableScanDescData
pg_atomic_uint64 phs_nallocated; /* number of blocks allocated to
* workers so far. */
/*
- * The Set of fields has different meaning for AO/AOCO tables:
- * phs_nblocks: total # of segment files
- * phs_nallocated: # of processed segment file
- * phs_startblock: not used
+ * These fields have a different meaning for AO and AOCS tables:
+ *
+ * AO: phs_nblocks is the number of segment files, and phs_nallocated
+ * the number of segment files handed out to workers.
+ * AOCS: phs_nblocks is the number of chunks, and phs_nallocated the
+ * number of chunks handed out to workers. A chunk is a range of
+ * rows of one segment file; see aocs_parallelscan_nchunks().
+ *
+ * phs_startblock and phs_mutex are not used.
*/
} ParallelBlockTableScanDescData;
typedef struct ParallelBlockTableScanDescData *ParallelBlockTableScanDesc;
@@ -95,80 +100,6 @@ 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 2c3f1c702b4..f1ed9429256 100644
--- a/src/include/cdb/cdbaocsam.h
+++ b/src/include/cdb/cdbaocsam.h
@@ -253,17 +253,26 @@ typedef struct AOCSScanDescData
int32 cur_seg;
/*
- * POC: block-granular parallel scan state.
+ * Parallel scan state, used only when rs_base.rs_parallel is set.
*
- * 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.
+ * Every process computes the chunk layout from its own seginfo[]:
+ * rows_per_chunk, and chunk_first[i], the number of the first chunk of
+ * seginfo[i]. The shared ParallelBlockTableScanDesc holds only the total
+ * number of chunks and the number handed out so far.
+ *
+ * chunk_seg is the index in seginfo[] of the segment file whose column
+ * files are open, or -1 if none are. chunk_end_row is the ordinal just
+ * past the current chunk; it is compared with segrowsprocessed, the number
+ * of rows consumed in the segment file so far. chunks_claimed and
+ * rows_claimed count this process's share of the scan, for the DEBUG1
+ * message in aocs_endscan().
*/
- bool chunked;
- AOParallelScanChunk cur_chunk;
- struct AppendOnlyBlockDirectory *posBlkdir;
+ int64 rows_per_chunk;
+ uint64 *chunk_first;
+ int32 chunk_seg;
+ int64 chunk_end_row;
+ int64 chunks_claimed;
+ int64 rows_claimed;
/*
* The only relation wide Storage Option, the rest are aquired in a per
@@ -379,14 +388,9 @@ 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 BlockNumber aocs_parallelscan_nchunks(Relation rel,
+ AOCSFileSegInfo **seginfo,
+ int total_seg);
extern AOCSScanDesc aocs_beginscan(Relation relation, Snapshot snapshot, ParallelTableScanDesc parallel_scan,
bool *proj, uint32 flags);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org
(cloudberry) 01/02: 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 cf9c3b7a8c03e46e0d182e0b7f5516747585d565
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