From 5e235603b3905f816626b3ddbaaff4ad32cf0285 Mon Sep 17 00:00:00 2001 From: roseduan Date: Tue, 20 Oct 2026 01:33:22 -0700 Subject: [PATCH] Parallel scan AO row and AOCS tables by chunks of rows A parallel scan of an AO/AOCS table used to give each worker whole segment files, and the planner capped the number of workers at pg_appendonly.segfilecount. A table loaded by one transaction usually has a single segment file per segment, so it was scanned by one process. Now each segment file is split into chunks of rows, and the workers claim chunks from a shared counter, as heap does with blocks. Every process computes the same chunk layout from pg_aoseg using the scan's snapshot, so the shared state holds only the number of chunks and the counter. A scan has about 2048 chunks, bounded in bytes by the table's average row size. A block belongs to the chunk that holds its first row. To reach its chunk, a process reads only block headers and skips other chunks' blocks without decompressing them. AOCS positions each column file separately. The planner now sizes AO/AOCS workers by table size, as for heap, instead of by segment file count. --- src/backend/access/aocs/aocsam.c | 370 ++++- .../access/appendonly/README.parallel_scan.md | 177 +++ src/backend/access/appendonly/appendonlyam.c | 288 +++- src/backend/access/table/tableam.c | 6 +- src/backend/optimizer/path/allpaths.c | 31 +- src/include/cdb/cdbaocsam.h | 20 + src/include/cdb/cdbappendonlyam.h | 65 +- .../uao/parallel_scan_concurrency.source | 102 ++ src/test/isolation2/isolation2_schedule | 4 + .../uao/parallel_scan_concurrency.source | 229 +++ src/test/regress/expected/cbdb_parallel.out | 49 +- src/test/regress/expected/zlib.out | 4 +- src/test/regress/greenplum_schedule | 3 + .../input/uao_dml/ao_parallel_scan.source | 546 +++++++ .../output/uao_dml/ao_parallel_scan.source | 1372 +++++++++++++++++ src/test/regress/sql/cbdb_parallel.sql | 36 +- src/test/regress/sql/zlib.sql | 4 +- 17 files changed, 3226 insertions(+), 80 deletions(-) create mode 100644 src/backend/access/appendonly/README.parallel_scan.md create mode 100644 src/test/isolation2/input/uao/parallel_scan_concurrency.source create mode 100644 src/test/isolation2/output/uao/parallel_scan_concurrency.source create mode 100644 src/test/regress/input/uao_dml/ao_parallel_scan.source create mode 100644 src/test/regress/output/uao_dml/ao_parallel_scan.source diff --git a/src/backend/access/aocs/aocsam.c b/src/backend/access/aocs/aocsam.c index 244d1da808d..3c013d4eb1a 100644 --- a/src/backend/access/aocs/aocsam.c +++ b/src/backend/access/aocs/aocsam.c @@ -64,6 +64,9 @@ static void reorder_qual_col(AOCSScanDesc scan); static bool aocs_col_predicate_test(AOCSScanDesc scan, TupleTableSlot *slot, int i, bool sample_phase); static bool aocs_getnext_sample(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot *slot); static void aocs_insert_finish_guts(AOCSInsertDesc aoInsertDesc); +static void aocs_parallelscan_init(AOCSScanDesc scan); +static inline bool aocs_parallelscan_chunk_done(AOCSScanDesc scan); +static int aocs_parallelscan_next_chunk(AOCSScanDesc scan); /* Hook for plugins to get control in aocs_delete() */ aocs_delete_hook_type aocs_delete_hook = NULL; @@ -355,29 +358,14 @@ initscan_with_colinfo(AOCSScanDesc scan) static int open_next_scan_seg(AOCSScanDesc scan) { - bool isParallel = false; - ParallelBlockTableScanDesc pbscan = NULL; - if (scan->rs_base.rs_parallel != NULL) - { - isParallel = true; - pbscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel; - } + return aocs_parallelscan_next_chunk(scan); 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) - 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]; @@ -661,6 +649,9 @@ aocs_beginscan_internal(Relation relation, } } + if (scan->rs_base.rs_parallel != NULL) + aocs_parallelscan_init(scan); + return scan; } @@ -683,6 +674,9 @@ aocs_endscan(AOCSScanDesc scan) { close_cur_scan_seg(scan); + if (scan->seg_first_chunk) + pfree(scan->seg_first_chunk); + if (scan->columnScanInfo.ds) { Assert(scan->columnScanInfo.proj_atts); @@ -1161,7 +1155,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_parallelscan_chunk_done(scan)) { err = open_next_scan_seg(scan); if (err < 0) @@ -1171,7 +1165,9 @@ aocs_getnext_noqual(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot * scan->cur_seg = -1; return false; } - scan->segrowsprocessed = 0; + /* parallel scans set it in aocs_parallelscan_next_chunk() */ + if (scan->rs_base.rs_parallel == NULL) + scan->segrowsprocessed = 0; } Assert(scan->cur_seg >= 0); @@ -1308,7 +1304,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_parallelscan_chunk_done(scan)) { err = open_next_scan_seg(scan); if (err < 0) @@ -1318,7 +1314,9 @@ aocs_getnext_withqual(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot scan->cur_seg = -1; return false; } - scan->segrowsprocessed = 0; + /* parallel scans set it in aocs_parallelscan_next_chunk() */ + if (scan->rs_base.rs_parallel == NULL) + scan->segrowsprocessed = 0; } Assert(scan->cur_seg >= 0); @@ -3045,7 +3043,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_parallelscan_chunk_done(scan)) { err = open_next_scan_seg(scan); if (err < 0) @@ -3055,7 +3053,9 @@ aocs_getnext_sample(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot * scan->cur_seg = -1; return false; } - scan->segrowsprocessed = 0; + /* parallel scans set it in aocs_parallelscan_next_chunk() */ + if (scan->rs_base.rs_parallel == NULL) + scan->segrowsprocessed = 0; } Assert(scan->cur_seg >= 0); @@ -3163,3 +3163,323 @@ aocs_getnext_sample(AOCSScanDesc scan, ScanDirection direction, TupleTableSlot * return false; } +/* + * Return the number of rows of a segment file that a parallel scan reads. + * AWAITING_DROP segment files are skipped. + */ +static inline int64 +aocs_segfile_scan_rows(AOCSFileSegInfo *seginfo) +{ + /* + * subtle: we must check for AWAITING_DROP before looking at vpinfo, see + * open_next_scan_seg(). + */ + if (seginfo->state == AOSEG_STATE_AWAITING_DROP) + return 0; + + return seginfo->total_tupcount; +} + +/* + * Return the number of rows in each chunk of a parallel scan. The last chunk + * of a segment file may be shorter. + * + * The average row size is the uncompressed size of all live columns divided + * by the number of rows, over the segment files the scan reads. + */ +static int64 +aocs_parallelscan_rows_per_chunk(Relation rel, AOCSFileSegInfo **seginfo, + int total_seg) +{ + TupleDesc tupdesc = RelationGetDescr(rel); + int64 total_rows = 0; + int64 total_bytes = 0; + + for (int i = 0; i < total_seg; i++) + { + int64 rows = aocs_segfile_scan_rows(seginfo[i]); + + if (rows == 0) + continue; + + total_rows += rows; + for (int attno = 0; attno < tupdesc->natts; attno++) + { + if (TupleDescAttr(tupdesc, attno)->attisdropped) + continue; + total_bytes += getAOCSVPEntry(seginfo[i], attno)->eof_uncompressed; + } + } + + return AppendOnlyParallelScan_RowsPerChunk(total_rows, total_bytes); +} + +/* + * Return the number of chunks of a segment file. Empty and AWAITING_DROP + * segment files have none. + */ +static inline uint64 +aocs_segfile_chunks(AOCSFileSegInfo *seginfo, int64 rows_per_chunk) +{ + return (aocs_segfile_scan_rows(seginfo) + rows_per_chunk - 1) / rows_per_chunk; +} + +/* + * 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_compute_chunks(Relation rel, AOCSFileSegInfo **seginfo, + int total_seg) +{ + int64 rows_per_chunk; + uint64 total_chunks = 0; + + rows_per_chunk = aocs_parallelscan_rows_per_chunk(rel, seginfo, total_seg); + for (int i = 0; i < total_seg; i++) + total_chunks += aocs_segfile_chunks(seginfo[i], rows_per_chunk); + + if (total_chunks > MaxBlockNumber) + elog(ERROR, "too many chunks for parallel scan of relation \"%s\"", + RelationGetRelationName(rel)); + + elog(DEBUG1, "parallel scan of AOCS relation \"%s\": %d segment files, " + INT64_FORMAT " rows per chunk, " UINT64_FORMAT " chunks", + RelationGetRelationName(rel), total_seg, rows_per_chunk, total_chunks); + + return (BlockNumber) total_chunks; +} + +/* + * 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 +aocs_parallelscan_init(AOCSScanDesc scan) +{ + ParallelBlockTableScanDesc bpscan; + uint64 total_chunks = 0; + + bpscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel; + scan->rows_per_chunk = aocs_parallelscan_rows_per_chunk(scan->rs_base.rs_rd, + scan->seginfo, + scan->total_seg); + scan->seg_first_chunk = (uint64 *) + palloc(Max(scan->total_seg, 1) * sizeof(uint64)); + + /* + * seg_first_chunk[i] is the number of the first chunk of seginfo[i]. + * A segment file without chunks gets the same number as the next one. + * aocs_parallelscan_next_chunk() picks the last segment file whose first + * chunk is not after the chunk number, so it never picks such a segment + * file. + */ + for (int i = 0; i < scan->total_seg; i++) + { + scan->seg_first_chunk[i] = total_chunks; + total_chunks += aocs_segfile_chunks(scan->seginfo[i], scan->rows_per_chunk); + } + + /* + * 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 (total_chunks != 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), total_chunks, + bpscan->phs_nblocks); +} + +/* + * Has the current chunk of a parallel scan been read completely? + */ +static inline bool +aocs_parallelscan_chunk_done(AOCSScanDesc scan) +{ + return scan->rs_base.rs_parallel != NULL && + scan->segrowsprocessed >= scan->chunk_end_row; +} + +/* + * 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_parallelscan_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 (left >= ds->blockRowCount - consumed) + { + bool loaded = false; + + /* the target row is past the current block */ + left -= (ds->blockRowCount - consumed); + + for (;;) + { + int64 nextFirstRowNum = ds->blockFirstRowNum + ds->blockRowCount; + + CHECK_FOR_INTERRUPTS(); + + 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)); + + /* DatumStreamRead: firstRow is expected to be -1 for pre-4.0 blocks */ + loaded = false; + if (ds->getBlockInfo.firstRow < 0) + { + ds->blockFirstRowNum = nextFirstRowNum; + datumstreamread_block_content(ds); + AOCSScanDesc_UpdateTotalBytesRead(scan, attno); + loaded = true; + } + + if (left < ds->blockRowCount) + break; + + left -= ds->blockRowCount; + if (!loaded) + AppendOnlyStorageRead_SkipCurrentBlock(&ds->ao_read); + } + + if (!loaded) + { + datumstreamread_block_content(ds); + AOCSScanDesc_UpdateTotalBytesRead(scan, attno); + } + consumed = 0; + } + + /* + * Skip forward inside the block. Stop one row short, because the + * caller's next datumstreamread_advance() returns the target row. + */ + if (left > 0) + { + Assert(ds->largeObjectState == DatumStreamLargeObjectState_None); + datumstreamread_find(ds, consumed + left - 1); + } + } +} + +/* + * 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 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 +aocs_parallelscan_next_chunk(AOCSScanDesc scan) +{ + ParallelBlockTableScanDesc bpscan; + DatumStreamRead *ds; + uint64 chunk; + int64 startrow; + int seg; + int lo; + int hi; + + /* building a block directory needs to see every block, in order */ + Assert(scan->blockDirectory == NULL); + + bpscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel; + chunk = pg_atomic_fetch_add_u64(&bpscan->phs_nallocated, 1); + if (chunk >= bpscan->phs_nblocks) + { + close_cur_scan_seg(scan); + 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; + + if (scan->seg_first_chunk[mid] <= chunk) + lo = mid; + else + hi = mid - 1; + } + seg = lo; + + startrow = (int64) (chunk - scan->seg_first_chunk[seg]) * + scan->rows_per_chunk; + scan->chunk_end_row = Min(startrow + scan->rows_per_chunk, + scan->seginfo[seg]->total_tupcount); + + /* + * Open the chunk's segment file unless its column files are already open + * and the chunk is ahead of the current position. The scan loops close + * the files without resetting cur_seg when a column reaches the end of + * the segment file, so check the datum stream as well. + */ + ds = scan->columnScanInfo.ds[scan->columnScanInfo.proj_atts[0]]; + + if (scan->cur_seg != seg || !ds->need_close_file || + startrow < scan->segrowsprocessed) + { + close_cur_scan_seg(scan); + + 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->segrowsprocessed = 0; + } + + aocs_parallelscan_skip_rows(scan, startrow - scan->segrowsprocessed); + scan->segrowsprocessed = startrow; + + return seg; +} diff --git a/src/backend/access/appendonly/README.parallel_scan.md b/src/backend/access/appendonly/README.parallel_scan.md new file mode 100644 index 00000000000..27482c0f093 --- /dev/null +++ b/src/backend/access/appendonly/README.parallel_scan.md @@ -0,0 +1,177 @@ +This describes how the parallel workers of one segment share a sequential +scan of an AO row or AOCS table. A worker here is any process taking part +in the scan, including the leader. Both use the same chunk layout and the same shared +state; they differ only in how a worker reads a chunk. + +Segment file and block formats: `src/backend/access/appendonly/README.md`. + + +# Chunks + +Each segment file is split into chunks of rows, and the workers claim chunks +one at a time from a shared counter: + +AO row: one file per segment file. + +``` +segfile 1 .1 | c0 | c1 | c2 | +segfile 2 .2 | c3 | c4 | c5 | c6 | +``` + +AOCS: one file per column and segment file. A chunk covers the same rows in +every column file of its segment file. + +``` +segfile 1 column a .1 | c0 | c1 | c2 | + column b .129 | c0 | c1 | c2 | +segfile 2 column a .2 | c3 | c4 | c5 | c6 | + column b .130 | c3 | c4 | c5 | c6 | +``` + +In both cases, with three workers A, B and C, the chunks may be claimed as: + +``` +chunk c0 c1 c2 c3 c4 c5 c6 +worker A B C A C B A +``` + +A chunk is a range of row ordinals: a row's position in its segment file, +counting from 0. Ordinals are used because they can be computed without +reading the files: `tupcount` in `pg_aoseg` gives the number of rows of each +segment file. For AOCS, the ordinal is also the only position all column +files agree on, since each column file has its own block boundaries. + +Ordinals are not row numbers, which can have gaps. + + +# Chunk layout + +Every worker reads the segment files from `pg_aoseg` (AO row) or +`pg_aocsseg` (AOCS) with the scan's snapshot, and computes the same layout: + +``` +rows_per_chunk = 1000 + +segno tupcount chunks seg_first_chunk + 1 2500 c0 c1 c2 0 + 2 0 - 3 (empty: no chunks) + 3 1200 c3 c4 3 +``` + +- `rows_per_chunk` aims at about 2048 chunks per scan, each between 512kB and + 32MB of uncompressed row data (`AppendOnlyParallelScan_RowsPerChunk()`). + Small chunks keep the blocks other workers skip over in the OS cache. +- A chunk never spans two segment files; the last one of a file is short. +- Empty and AWAITING_DROP segment files get no chunks. +- `tupcount` includes deleted rows, so chunks cover every stored row. + + +# Shared and local state + +Shared, in `ParallelBlockTableScanDescData`: + +- `phs_nblocks`: the total number of chunks. +- `phs_nallocated`: an atomic counter of the chunks handed out. + +Local, in `AppendOnlyScanDescData` and `AOCSScanDescData`: + +- `rows_per_chunk`, `seg_first_chunk[]`: the chunk layout. +- `cur_seg`: the segment file being read. +- `chunk_start_row`, `chunk_end_row`: the current chunk. +- `segrowsprocessed`: how far the worker has read in the segment file. + +The first worker computes the number of chunks +(`*_parallelscan_compute_chunks()`). Every worker computes the layout again +in `*_parallelscan_init()`, and checks it has the same number of chunks. + + +# Claiming a chunk + +`*_parallelscan_next_chunk()`: + +``` +chunk = fetch_add(phs_nallocated, 1) = 4 +segfile = last with seg_first_chunk <= 4 = segfile 3 (first chunk 3) +rows = [(4 - 3) * 1000, min(2000, 1200)) = [1000, 1200) +``` + +When `chunk >= phs_nblocks`, the scan is done. + + +# Reading a chunk + +Block headers are not compressed and hold the block's row count, so a worker +can skip a block without decompressing it. The skipped block is still read +from the file (there is no seek), but only its header is looked at. + +## AO row + +A block holds whole rows and is never split. It belongs to the chunk that +holds its first row: + +``` +rows 0 1000 2000 + |---c0----|---c1----| +blocks [b0][b1][b2][b3][b4] 400 rows per block + +c0 reads b0 b1 b2 (b2 is rows 800-1199, it starts in c0) +c1 reads b3 b4 +``` + +So the rows a worker returns can differ from its chunk's range by less than +one block at each end. See `appendonly_parallelscan_next_block()`. + +## AOCS + +Each column file has its own block boundaries, so a chunk is an exact range +of rows. `aocs_parallelscan_skip_rows()` moves every scanned column to the +chunk's first row, and the scan stops at its last row: + +``` +rows 0 1000 2000 + |---c0----|---c1----| +column a [b0][b1][b2][b3][b4] 400 rows per block +column b [b0 ][b1 ][b2 ] 700 rows per block + +c1, column a: skip b0 b1, load b2, skip to row 1000 +c1, column b: skip b0, load b1, skip to row 1000 +``` + +A block that crosses a chunk boundary is loaded by both workers, and each +returns only its own rows. + + +# A worker only moves forward + +A worker's chunks have increasing numbers, so its next chunk is always after +the current one. It never reads a file backwards, and reads each block at +most once. In each segment file, it reads from the start of the file to the +end of its last chunk there: + +``` +worker A claims c0, c2, c4 +segfile 1 | c0 | c1 | c2 | read c0, skip c1, read c2 +segfile 3 | c3 | c4 | open, skip c3, read c4 +``` + + +# Deleted rows and compression + +- Deleted rows are counted in chunks like any other row; the visibility map + check drops them. +- Row numbers are computed from each block's first row number, so a skipped + block still advances it (`AppendOnlyExecutionReadBlock_FinishedScanBlock()` + for AO row). +- A large value stored over several blocks counts as one row. + + +# Limitations + +- Skipping a block does not seek past it, so a worker reads every block + before its last chunk in a file, not only the blocks of its own chunks: + the segment file (AO row), or each scanned column file (AOCS). The + workers move through a file together, a few chunks apart, so with small + chunks most of those reads come from the OS cache. +- A block directory could let a worker seek to its chunk instead, but this + is not implemented. +- Building a block directory during a scan is not supported in parallel. diff --git a/src/backend/access/appendonly/appendonlyam.c b/src/backend/access/appendonly/appendonlyam.c index aeffdc210b1..17d3157e7d6 100755 --- a/src/backend/access/appendonly/appendonlyam.c +++ b/src/backend/access/appendonly/appendonlyam.c @@ -102,11 +102,15 @@ extern void finishWriteBlock(AppendOnlyInsertDesc aoInsertDesc); extern void setupNextWriteBlock(AppendOnlyInsertDesc aoInsertDesc); static void appendonly_insert_finish_guts(AppendOnlyInsertDesc aoInsertDesc); -/* Hook for plugins to get control in appendonly_delete() */ -appendonly_delete_hook_type appendonly_delete_hook = NULL; static void AppendOnlyScanDesc_UpdateTotalBytesRead( AppendOnlyScanDesc scan); +static void appendonly_parallelscan_init(AppendOnlyScanDesc scan); +static bool appendonly_parallelscan_next_block(AppendOnlyScanDesc scan); + +/* Hook for plugins to get control in appendonly_delete() */ +appendonly_delete_hook_type appendonly_delete_hook = NULL; + /* ---------------- * initscan - scan code common to appendonly_beginscan and appendonly_rescan * ---------------- @@ -125,6 +129,7 @@ initscan(AppendOnlyScanDesc scan, ScanKey key) scan->aos_need_new_segfile = true; /* need to assign a file to be scanned */ scan->aos_done_all_segfiles = false; scan->needNextBuffer = true; + scan->cur_seg = -1; if (scan->initedStorageRoutines) AppendOnlyExecutorReadBlock_ResetCounts( @@ -147,15 +152,6 @@ SetNextFileSegForRead(AppendOnlyScanDesc scan) int formatversion = -2; /* some invalid value */ bool finished_all_files = true; /* assume */ int32 fileSegNo; - bool isParallel = false; - ParallelBlockTableScanDesc pbscan = NULL; - - if (scan->rs_base.rs_parallel != NULL) - { - isParallel = true; - pbscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel; - } - Assert(scan->aos_need_new_segfile); /* only call me when last segfile * completed */ @@ -233,16 +229,7 @@ SetNextFileSegForRead(AppendOnlyScanDesc scan) int idx; /* fetch segfile idx */ while (scan->aos_segfiles_processed < scan->aos_total_segfiles) { - if (isParallel) - { - idx = pg_atomic_fetch_add_u64(&pbscan->phs_nallocated, 1); - if (idx >= pbscan->phs_nblocks) - break; - } - else - { - idx = scan->aos_segfiles_processed; - } + idx = scan->aos_segfiles_processed; FileSegInfo *fsinfo = scan->aos_segfile_arr[idx]; @@ -1080,6 +1067,9 @@ AppendOnlyExecutorReadBlock_FetchTuple(AppendOnlyExecutorReadBlock *executorRead static bool getNextBlock(AppendOnlyScanDesc scan) { + if (scan->rs_base.rs_parallel != NULL) + return appendonly_parallelscan_next_block(scan); + if (scan->aos_need_new_segfile) { /* @@ -1796,6 +1786,9 @@ appendonly_beginrangescan_internal(Relation relation, scan->totalBytesRead = 0; + if (scan->rs_base.rs_parallel != NULL) + appendonly_parallelscan_init(scan); + return scan; } @@ -1951,6 +1944,9 @@ appendonly_endscan(TableScanDesc scan) pfree(aoscan->aos_segfile_arr); } + if (aoscan->seg_first_chunk) + pfree(aoscan->seg_first_chunk); + CloseScannedFileSeg(aoscan); AppendOnlyStorageRead_FinishSession(&aoscan->storageRead); @@ -3378,4 +3374,254 @@ appendonly_predicate_pushdown_prepare(AppendOnlyScanDesc aoscan, return NULL; } +/* + * Return the number of rows of a segment file that a parallel scan reads. + * AWAITING_DROP segment files are skipped, as in SetNextFileSegForRead(). + */ +static inline int64 +appendonly_segfile_scan_rows(FileSegInfo *seginfo) +{ + if (seginfo->state == AOSEG_STATE_AWAITING_DROP) + return 0; + + return seginfo->total_tupcount; +} + +/* + * Return the number of rows in each chunk of a parallel scan. The last chunk + * of a segment file may be shorter. + * + * The average row size is the uncompressed size of the segment files the scan + * reads, divided by their number of rows. + */ +static int64 +appendonly_parallelscan_rows_per_chunk(FileSegInfo **seginfo, int total_seg) +{ + int64 total_rows = 0; + int64 total_bytes = 0; + + for (int i = 0; i < total_seg; i++) + { + int64 rows = appendonly_segfile_scan_rows(seginfo[i]); + + if (rows == 0) + continue; + + total_rows += rows; + total_bytes += seginfo[i]->eof_uncompressed; + } + + return AppendOnlyParallelScan_RowsPerChunk(total_rows, total_bytes); +} + +/* + * Return the number of chunks of a segment file. Empty and AWAITING_DROP + * segment files have none. + */ +static inline uint64 +appendonly_segfile_chunks(FileSegInfo *seginfo, int64 rows_per_chunk) +{ + return (appendonly_segfile_scan_rows(seginfo) + rows_per_chunk - 1) / rows_per_chunk; +} + +/* + * 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 +appendonly_parallelscan_compute_chunks(Relation rel, FileSegInfo **seginfo, + int total_seg) +{ + int64 rows_per_chunk; + uint64 total_chunks = 0; + + rows_per_chunk = appendonly_parallelscan_rows_per_chunk(seginfo, total_seg); + for (int i = 0; i < total_seg; i++) + total_chunks += appendonly_segfile_chunks(seginfo[i], rows_per_chunk); + + if (total_chunks > MaxBlockNumber) + elog(ERROR, "too many chunks for parallel scan of relation \"%s\"", + RelationGetRelationName(rel)); + + elog(DEBUG1, "parallel scan of AO relation \"%s\": %d segment files, " + INT64_FORMAT " rows per chunk, " UINT64_FORMAT " chunks", + RelationGetRelationName(rel), total_seg, rows_per_chunk, total_chunks); + + return (BlockNumber) total_chunks; +} + +/* + * 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 +appendonly_parallelscan_init(AppendOnlyScanDesc scan) +{ + ParallelBlockTableScanDesc bpscan; + uint64 total_chunks = 0; + + bpscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel; + scan->rows_per_chunk = appendonly_parallelscan_rows_per_chunk(scan->aos_segfile_arr, + scan->aos_total_segfiles); + scan->seg_first_chunk = (uint64 *) + palloc(Max(scan->aos_total_segfiles, 1) * sizeof(uint64)); + + /* + * seg_first_chunk[i] is the number of the first chunk of + * aos_segfile_arr[i]. A segment file without chunks gets the same number + * as the next one. appendonly_parallelscan_next_chunk() picks the last + * segment file whose first chunk is not after the chunk number, so it + * never picks such a segment file. + */ + for (int i = 0; i < scan->aos_total_segfiles; i++) + { + scan->seg_first_chunk[i] = total_chunks; + total_chunks += appendonly_segfile_chunks(scan->aos_segfile_arr[i], + scan->rows_per_chunk); + } + + /* + * 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 (total_chunks != 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->aos_rd), total_chunks, + bpscan->phs_nblocks); +} + +/* + * Claim the next chunk of a parallel scan, and open its segment file if it is + * not the one of the previous chunk. + * + * Returns false 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 blocks + * it has already read or skipped, and the file is never read backwards. + */ +static bool +appendonly_parallelscan_next_chunk(AppendOnlyScanDesc scan) +{ + ParallelBlockTableScanDesc bpscan; + uint64 chunk; + int seg; + int lo; + int hi; + + bpscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel; + chunk = pg_atomic_fetch_add_u64(&bpscan->phs_nallocated, 1); + if (chunk >= bpscan->phs_nblocks) + { + CloseScannedFileSeg(scan); + scan->aos_need_new_segfile = false; + scan->aos_done_all_segfiles = true; + return false; + } + + /* find the last segment file whose first chunk is not after this one */ + lo = 0; + hi = scan->aos_total_segfiles - 1; + while (lo < hi) + { + int mid = (lo + hi + 1) / 2; + + if (scan->seg_first_chunk[mid] <= chunk) + lo = mid; + else + hi = mid - 1; + } + seg = lo; + + if (seg != scan->cur_seg) + { + CloseScannedFileSeg(scan); + + /* make SetNextFileSegForRead() open aos_segfile_arr[seg] */ + scan->aos_segfiles_processed = seg; + if (!SetNextFileSegForRead(scan) || scan->aos_segfiles_processed != seg + 1) + elog(ERROR, "inconsistent segment state for relation %s, segment %d, tuple count " INT64_FORMAT, + RelationGetRelationName(scan->aos_rd), + scan->aos_segfile_arr[seg]->segno, + scan->aos_segfile_arr[seg]->total_tupcount); + + scan->cur_seg = seg; + scan->segrowsprocessed = 0; + } + + scan->chunk_start_row = (int64) (chunk - scan->seg_first_chunk[seg]) * + scan->rows_per_chunk; + scan->chunk_end_row = Min(scan->chunk_start_row + scan->rows_per_chunk, + scan->aos_segfile_arr[seg]->total_tupcount); + + return true; +} + +/* + * Get the next block of a parallel scan. getNextBlock() calls this for + * parallel scans. + * + * Returns false when all chunks have been handed out. + */ +static bool +appendonly_parallelscan_next_block(AppendOnlyScanDesc scan) +{ + AppendOnlyExecutorReadBlock *varblock = &scan->executorReadBlock; + + /* building a block directory needs to see every block, in order */ + Assert(scan->blockDirectory == NULL); + + for (;;) + { + bool mine; + + CHECK_FOR_INTERRUPTS(); + + /* the current chunk is done, or there is no chunk yet */ + if (scan->cur_seg < 0 || scan->segrowsprocessed >= scan->chunk_end_row) + { + if (!appendonly_parallelscan_next_chunk(scan)) + return false; + continue; + } + + /* + * The next block starts at ordinal segrowsprocessed, which is inside + * the current chunk. The blocks of the file hold all of its rows, so + * there must be one. + */ + if (!AppendOnlyExecutorReadBlock_GetBlockInfo(&scan->storageRead, varblock)) + { + FileSegInfo *fsinfo = scan->aos_segfile_arr[scan->cur_seg]; + + elog(ERROR, "unexpected end of segment file %d of relation \"%s\" in parallel scan, " + "after " INT64_FORMAT " of " INT64_FORMAT " rows", + fsinfo->segno, RelationGetRelationName(scan->aos_rd), + scan->segrowsprocessed, fsinfo->total_tupcount); + } + + Assert(varblock->rowCount > 0); + mine = (scan->segrowsprocessed >= scan->chunk_start_row); + scan->segrowsprocessed += varblock->rowCount; + + if (mine) + { + AppendOnlyExecutorReadBlock_GetContents(varblock); + AppendOnlyScanDesc_UpdateTotalBytesRead(scan); + return true; + } + + /* + * The block belongs to an earlier chunk. Skip it without + * decompressing it, and keep the row numbers of the blocks that + * follow right; they are not stored in older block formats. + */ + AppendOnlyExecutionReadBlock_FinishedScanBlock(varblock); + AppendOnlyStorageRead_SkipCurrentBlock(&scan->storageRead); + } +} + /* end of file */ diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c index 70cb8cc9cc7..6001805291d 100644 --- a/src/backend/access/table/tableam.c +++ b/src/backend/access/table/tableam.c @@ -24,6 +24,8 @@ #include "access/syncscan.h" #include "access/tableam.h" #include "access/xact.h" +#include "cdb/cdbaocsam.h" +#include "cdb/cdbappendonlyam.h" #include "optimizer/plancat.h" #include "port/pg_bitutils.h" #include "storage/bufmgr.h" @@ -196,7 +198,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan, appendOnlyMetaDataSnapshot = GetTransactionSnapshot(); } seginfo = GetAllFileSegInfo(rel, appendOnlyMetaDataSnapshot, &segfile_count, NULL); - bpscan->phs_nblocks = segfile_count; + bpscan->phs_nblocks = appendonly_parallelscan_compute_chunks(rel, seginfo, segfile_count); if (seginfo) { FreeAllSegFileInfo(seginfo, segfile_count); @@ -218,7 +220,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan, appendOnlyMetaDataSnapshot = GetTransactionSnapshot(); } seginfo = GetAllAOCSFileSegInfo(rel, appendOnlyMetaDataSnapshot, &segfile_count, NULL); - bpscan->phs_nblocks = segfile_count; + bpscan->phs_nblocks = aocs_parallelscan_compute_chunks(rel, seginfo, segfile_count); if (seginfo) { FreeAllAOCSSegFileInfo(seginfo, segfile_count); diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c index 6bb0f9a92a5..f14ae0ff316 100644 --- a/src/backend/optimizer/path/allpaths.c +++ b/src/backend/optimizer/path/allpaths.c @@ -5188,35 +5188,12 @@ compute_parallel_worker(PlannerInfo *root, RelOptInfo *rel, double heap_pages, d else { /* - * We need to reconsider parallel workers for AO/AOCO tables - * because page number in ao is quite hard to estimate. - * The parallel for AO/AOCO tables is based on segment file count in - * pg_appendonly which will be updated by analyze/vacuum/truncate processes. + * Parallel index scans are not supported on AO/AOCO tables. */ - if (rel->reloptkind == RELOPT_BASEREL && (AMHandlerIsAO(rel->amhandler))) - { - Oid aorelid = root->simple_rte_array[rel->relid]->relid; - HeapTuple aotup; - Form_pg_appendonly aoform; - aotup = SearchSysCache1(AORELID, ObjectIdGetDatum(aorelid)); - if (!HeapTupleIsValid(aotup)) - ereport(ERROR, - (errcode(ERRCODE_UNDEFINED_OBJECT), - errmsg("appendonly table relid %u does not exist in pg_appendonly", aorelid))); - - aoform = (Form_pg_appendonly) GETSTRUCT(aotup); - Assert(aoform->segfilecount >= 0); - parallel_workers = Min(aoform->segfilecount, max_workers); - ReleaseSysCache(aotup); + if (rel->reloptkind == RELOPT_BASEREL && AMHandlerIsAO(rel->amhandler) && + index_pages >= 0) + parallel_workers = 0; - /* - * Disable parallel for AO/AOCO for: - * 1.We don't support parallel/non-parallel IndexScan/IndexOnlyScan. - * 2.If parallel_workers is 1, it is pointless in gp parallel mode. - */ - if (parallel_workers == 1 || index_pages >= 0) - parallel_workers = 0; - } else { /* diff --git a/src/include/cdb/cdbaocsam.h b/src/include/cdb/cdbaocsam.h index 68c62b55b99..85b7de9da1d 100644 --- a/src/include/cdb/cdbaocsam.h +++ b/src/include/cdb/cdbaocsam.h @@ -252,6 +252,22 @@ typedef struct AOCSScanDescData int32 total_seg; int32 cur_seg; + /* + * Parallel scan state, used only when rs_base.rs_parallel is set. + * + * Every process computes the chunk layout from its own seginfo[]: + * rows_per_chunk, and seg_first_chunk[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_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. + */ + int64 rows_per_chunk; + uint64 *seg_first_chunk; + int64 chunk_end_row; + /* * The only relation wide Storage Option, the rest are aquired in a per * column basis and there is no need to keep track of. @@ -427,6 +443,10 @@ extern ExprState * aocs_predicate_pushdown_prepare(AOCSScanDesc scan, ExprState *state, ExprContext *ecxt, PlanState *ps); + +extern BlockNumber aocs_parallelscan_compute_chunks(Relation rel, + AOCSFileSegInfo **seginfo, + int total_seg); /* * Update total bytes read for the entire scan. If the block was compressed, * update it with the compressed length. If the block was not compressed, update diff --git a/src/include/cdb/cdbappendonlyam.h b/src/include/cdb/cdbappendonlyam.h index 1d2622a8ff2..f0f3c95ac69 100644 --- a/src/include/cdb/cdbappendonlyam.h +++ b/src/include/cdb/cdbappendonlyam.h @@ -53,6 +53,18 @@ #define MAX_APPENDONLY_BLOCK_SIZE (2 * 1024 * 1024) #define DEFAULT_VARBLOCK_TEMPSPACE_LEN (4 * 1024) #define DEFAULT_FS_SAFE_WRITE_SIZE (0) +/* + * Chunk sizing for parallel scans of AO row and AOCS tables. Like heap's + * PARALLEL_SEQSCAN_NCHUNKS, a scan is split into about + * APPENDONLY_PARALLEL_SCAN_NCHUNKS chunks. Chunks are counted in rows, with + * their size in bytes bounded using the average row size. The minimum keeps + * the cost of positioning at a chunk small. The maximum is much smaller than + * heap's, since a process positions itself by reading the blocks of other + * processes' chunks; small chunks keep those pages in the OS cache. + */ +#define APPENDONLY_PARALLEL_SCAN_NCHUNKS 2048 +#define APPENDONLY_PARALLEL_SCAN_MIN_CHUNK_BYTES (16 * DEFAULT_APPENDONLY_BLOCK_SIZE) +#define APPENDONLY_PARALLEL_SCAN_MAX_CHUNK_BYTES (INT64CONST(32) * 1024 * 1024) extern AppendOnlyBlockDirectory *GetAOBlockDirectory(Relation relation); @@ -261,6 +273,26 @@ typedef struct AppendOnlyScanDescData */ int64 segrowsprocessed; + /* + * Parallel scan state, used only when rs_base.rs_parallel is set. + * + * Every process computes the chunk layout from its own aos_segfile_arr[]: + * rows_per_chunk, and seg_first_chunk[i], the number of the first chunk + * of aos_segfile_arr[i]. The shared ParallelBlockTableScanDesc holds + * only the total number of chunks and the number handed out so far. + * + * cur_seg is the index in aos_segfile_arr[] of the segment file being + * read, the one of the current chunk, or -1, as in AOCSScanDesc. + * [chunk_start_row, chunk_end_row) are the ordinals of the chunk's rows. + * A block belongs to the chunk that holds its first row; segrowsprocessed + * is the ordinal of the first row of the next block in the file. + */ + int64 rows_per_chunk; + uint64 *seg_first_chunk; + int32 cur_seg; + int64 chunk_start_row; + int64 chunk_end_row; + AOBlkDirScan blkdirscan; /* scan current state */ @@ -451,7 +483,7 @@ typedef struct IndexFetchAppendOnlyData * ---------------- */ -extern AppendOnlyScanDesc appendonly_beginrangescan(Relation relation, +extern AppendOnlyScanDesc appendonly_beginrangescan(Relation relation, Snapshot snapshot, Snapshot appendOnlyMetaDataSnapshot, int *segfile_no_arr, int segfile_count, @@ -521,6 +553,10 @@ extern bool AppendOnlyExecutorReadBlock_ScanNextTuple(AppendOnlyExecutorReadBloc TupleTableSlot *slot); extern void AppendOnlyExecutorReadBlock_GetContents(AppendOnlyExecutorReadBlock *executorReadBlock); + +extern BlockNumber appendonly_parallelscan_compute_chunks(Relation rel, + FileSegInfo **seginfo, + int total_seg); /* * Update total bytes read for the entire scan. If the block was compressed, * update it with the compressed length. If the block was not compressed, update @@ -554,4 +590,31 @@ AppendOnlyScanDesc_TotalTupCount(AppendOnlyScanDesc scan) return totalrows; } +/* + * Return the number of rows in each chunk of a parallel scan, given the number + * of rows the scan reads and their uncompressed size in bytes. The last chunk + * of a segment file may be shorter. + */ +static inline int64 +AppendOnlyParallelScan_RowsPerChunk(int64 total_rows, int64 total_bytes) +{ + int64 row_bytes; + int64 rows_per_chunk; + + /* no rows to scan, so no chunks */ + if (total_rows == 0) + return 1; + + /* an estimated bytes per row */ + row_bytes = Max(total_bytes / total_rows, 1); + + rows_per_chunk = Max(total_rows / APPENDONLY_PARALLEL_SCAN_NCHUNKS, + APPENDONLY_PARALLEL_SCAN_MIN_CHUNK_BYTES / row_bytes); + rows_per_chunk = Min(rows_per_chunk, + APPENDONLY_PARALLEL_SCAN_MAX_CHUNK_BYTES / row_bytes); + + /* a row larger than the minimum chunk size is a chunk of its own */ + return Max(rows_per_chunk, 1); +} + #endif /* CDBAPPENDONLYAM_H */ diff --git a/src/test/isolation2/input/uao/parallel_scan_concurrency.source b/src/test/isolation2/input/uao/parallel_scan_concurrency.source new file mode 100644 index 00000000000..9c885603a9f --- /dev/null +++ b/src/test/isolation2/input/uao/parallel_scan_concurrency.source @@ -0,0 +1,102 @@ +-- @Description Concurrent changes while a parallel scan of an @amname@ table +-- is in progress. +-- +-- The scan's processes are suspended at executor_pre_tuple_processed: the +-- chunk layout is in shared memory and every process has its own copy, but no +-- row has been read yet. Another session then changes the table and commits. +-- The suspended scan must return exactly the rows its snapshot sees. +-- +-- All tables are distributed by a constant key, so their rows are on one +-- segment. Each segment runs 4 parallel processes, so the fault is suspended +-- for the first 4 hits on every segment; later hits, such as the other +-- session's statements, pass. + +CREATE TABLE pscan_insert (k int, a int, c text) + USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +INSERT INTO pscan_insert SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +CREATE TABLE pscan_delete (k int, a int, c text) + USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +INSERT INTO pscan_delete SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +CREATE TABLE pscan_vacuum (k int, a int, c text) + USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +INSERT INTO pscan_vacuum SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +DELETE FROM pscan_vacuum WHERE a % 2 = 0; +CREATE TABLE pscan_snapshot (k int, a int, c text) + USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +INSERT INTO pscan_snapshot SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; + +1: SET enable_parallel = on; +1: SET optimizer = off; +1: SET max_parallel_workers_per_gather = 4; +1: EXPLAIN (COSTS OFF) SELECT count(*), sum(a) FROM pscan_insert; + +-- Concurrent insert: the new rows are appended to the segment file after +-- the scan's chunk layout was computed, so the scan does not see them. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', '', '', '', 1, 4, 0, dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +1&: SELECT count(*), sum(a) FROM pscan_insert; +SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 4, dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +2: INSERT INTO pscan_insert SELECT 0, i, md5(i::text) FROM generate_series(100001, 150000) i; +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +1<: +1: SELECT count(*), sum(a) FROM pscan_insert; + +-- Concurrent delete: the deleted rows are still visible to the scan's +-- snapshot. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', '', '', '', 1, 4, 0, dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +1&: SELECT count(*), sum(a) FROM pscan_delete; +SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 4, dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +2: DELETE FROM pscan_delete WHERE a % 2 = 0; +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +1<: +1: SELECT count(*), sum(a) FROM pscan_delete; + +-- Concurrent VACUUM: compaction moves the visible rows to a new segment file +-- and marks the old one awaiting drop. It cannot drop the old one while the +-- scan's snapshot can still see it, and the scan reads the old one. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', '', '', '', 1, 4, 0, dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +1&: SELECT count(*), sum(a) FROM pscan_vacuum; +SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 4, dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +2: VACUUM pscan_vacuum; +-- The old segment file is awaiting drop (state 2), and the 50000 visible rows +-- are in a new one. +SELECT DISTINCT segno, tupcount, state FROM gp_toolkit.__gp_@aoseg@('pscan_vacuum') + WHERE tupcount > 0 OR state = 2 ORDER BY segno; +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) + FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; +1<: +1: SELECT count(*), sum(a) FROM pscan_vacuum; +2: VACUUM pscan_vacuum; +1: SELECT count(*), sum(a) FROM pscan_vacuum; + +-- A snapshot taken before a concurrent insert: the chunk layout is computed +-- from the scan's snapshot, so the new rows are not part of it. +1: BEGIN ISOLATION LEVEL REPEATABLE READ; +1: SELECT count(*) FROM pscan_snapshot; +2: INSERT INTO pscan_snapshot SELECT 0, i, md5(i::text) FROM generate_series(100001, 150000) i; +1: SELECT count(*), sum(a) FROM pscan_snapshot; +1: COMMIT; +1: SELECT count(*), sum(a) FROM pscan_snapshot; + +-- An error in one of the parallel processes fails the query, and the next +-- parallel scan works. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'error', '', '', '', 2, 2, 0, dbid) + FROM gp_segment_configuration WHERE content = 0 AND role = 'p'; +1: SELECT count(*), sum(a) FROM pscan_insert; +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) + FROM gp_segment_configuration WHERE content = 0 AND role = 'p'; +1: SELECT count(*), sum(a) FROM pscan_insert; + +1q: +2q: +DROP TABLE pscan_insert; +DROP TABLE pscan_delete; +DROP TABLE pscan_vacuum; +DROP TABLE pscan_snapshot; diff --git a/src/test/isolation2/isolation2_schedule b/src/test/isolation2/isolation2_schedule index 82d23731fb6..0a625b2c6b5 100644 --- a/src/test/isolation2/isolation2_schedule +++ b/src/test/isolation2/isolation2_schedule @@ -326,6 +326,10 @@ test: aocs_unique_index test: uao/ao_unique_index_vacuum_row test: uao/ao_unique_index_vacuum_column +# Concurrent changes during a parallel scan of an AO/CO table (uses fault injector) +test: uao/parallel_scan_concurrency_row +test: uao/parallel_scan_concurrency_column + test: spilling_hashagg test: local_directory_table_mixed diff --git a/src/test/isolation2/output/uao/parallel_scan_concurrency.source b/src/test/isolation2/output/uao/parallel_scan_concurrency.source new file mode 100644 index 00000000000..303fe051dc7 --- /dev/null +++ b/src/test/isolation2/output/uao/parallel_scan_concurrency.source @@ -0,0 +1,229 @@ +-- @Description Concurrent changes while a parallel scan of an @amname@ table +-- is in progress. +-- +-- The scan's processes are suspended at executor_pre_tuple_processed: the +-- chunk layout is in shared memory and every process has its own copy, but no +-- row has been read yet. Another session then changes the table and commits. +-- The suspended scan must return exactly the rows its snapshot sees. +-- +-- All tables are distributed by a constant key, so their rows are on one +-- segment. Each segment runs 4 parallel processes, so the fault is suspended +-- for the first 4 hits on every segment; later hits, such as the other +-- session's statements, pass. + +CREATE TABLE pscan_insert (k int, a int, c text) USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +CREATE +INSERT INTO pscan_insert SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +INSERT 100000 +CREATE TABLE pscan_delete (k int, a int, c text) USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +CREATE +INSERT INTO pscan_delete SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +INSERT 100000 +CREATE TABLE pscan_vacuum (k int, a int, c text) USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +CREATE +INSERT INTO pscan_vacuum SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +INSERT 100000 +DELETE FROM pscan_vacuum WHERE a % 2 = 0; +DELETE 50000 +CREATE TABLE pscan_snapshot (k int, a int, c text) USING @amname@ WITH (parallel_workers = 4) DISTRIBUTED BY (k); +CREATE +INSERT INTO pscan_snapshot SELECT 0, i, md5(i::text) FROM generate_series(1, 100000) i; +INSERT 100000 + +1: SET enable_parallel = on; +SET +1: SET optimizer = off; +SET +1: SET max_parallel_workers_per_gather = 4; +SET +1: EXPLAIN (COSTS OFF) SELECT count(*), sum(a) FROM pscan_insert; + QUERY PLAN +----------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on pscan_insert + Optimizer: Postgres query optimizer +(5 rows) + +-- Concurrent insert: the new rows are appended to the segment file after +-- the scan's chunk layout was computed, so the scan does not see them. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', '', '', '', 1, 4, 0, dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: + Success: + Success: +(3 rows) +1&: SELECT count(*), sum(a) FROM pscan_insert; +SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 4, dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_wait_until_triggered_fault +------------------------------- + Success: + Success: + Success: +(3 rows) +2: INSERT INTO pscan_insert SELECT 0, i, md5(i::text) FROM generate_series(100001, 150000) i; +INSERT 50000 +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: + Success: + Success: +(3 rows) +1<: <... completed> + count | sum +--------+------------ + 100000 | 5000050000 +(1 row) +1: SELECT count(*), sum(a) FROM pscan_insert; + count | sum +--------+------------- + 150000 | 11250075000 +(1 row) + +-- Concurrent delete: the deleted rows are still visible to the scan's +-- snapshot. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', '', '', '', 1, 4, 0, dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: + Success: + Success: +(3 rows) +1&: SELECT count(*), sum(a) FROM pscan_delete; +SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 4, dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_wait_until_triggered_fault +------------------------------- + Success: + Success: + Success: +(3 rows) +2: DELETE FROM pscan_delete WHERE a % 2 = 0; +DELETE 50000 +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: + Success: + Success: +(3 rows) +1<: <... completed> + count | sum +--------+------------ + 100000 | 5000050000 +(1 row) +1: SELECT count(*), sum(a) FROM pscan_delete; + count | sum +-------+------------ + 50000 | 2500000000 +(1 row) + +-- Concurrent VACUUM: compaction moves the visible rows to a new segment file +-- and marks the old one awaiting drop. It cannot drop the old one while the +-- scan's snapshot can still see it, and the scan reads the old one. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', '', '', '', 1, 4, 0, dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: + Success: + Success: +(3 rows) +1&: SELECT count(*), sum(a) FROM pscan_vacuum; +SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 4, dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_wait_until_triggered_fault +------------------------------- + Success: + Success: + Success: +(3 rows) +2: VACUUM pscan_vacuum; +VACUUM +-- The old segment file is awaiting drop (state 2), and the 50000 visible rows +-- are in a new one. +SELECT DISTINCT segno, tupcount, state FROM gp_toolkit.__gp_@aoseg@('pscan_vacuum') WHERE tupcount > 0 OR state = 2 ORDER BY segno; + segno | tupcount | state +-------+----------+------- + 1 | 100000 | 2 + 2 | 50000 | 1 +(2 rows) +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE content >= 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: + Success: + Success: +(3 rows) +1<: <... completed> + count | sum +-------+------------ + 50000 | 2500000000 +(1 row) +1: SELECT count(*), sum(a) FROM pscan_vacuum; + count | sum +-------+------------ + 50000 | 2500000000 +(1 row) +2: VACUUM pscan_vacuum; +VACUUM +1: SELECT count(*), sum(a) FROM pscan_vacuum; + count | sum +-------+------------ + 50000 | 2500000000 +(1 row) + +-- A snapshot taken before a concurrent insert: the chunk layout is computed +-- from the scan's snapshot, so the new rows are not part of it. +1: BEGIN ISOLATION LEVEL REPEATABLE READ; +BEGIN +1: SELECT count(*) FROM pscan_snapshot; + count +-------- + 100000 +(1 row) +2: INSERT INTO pscan_snapshot SELECT 0, i, md5(i::text) FROM generate_series(100001, 150000) i; +INSERT 50000 +1: SELECT count(*), sum(a) FROM pscan_snapshot; + count | sum +--------+------------ + 100000 | 5000050000 +(1 row) +1: COMMIT; +COMMIT +1: SELECT count(*), sum(a) FROM pscan_snapshot; + count | sum +--------+------------- + 150000 | 11250075000 +(1 row) + +-- An error in one of the parallel processes fails the query, and the next +-- parallel scan works. +SELECT gp_inject_fault('executor_pre_tuple_processed', 'error', '', '', '', 2, 2, 0, dbid) FROM gp_segment_configuration WHERE content = 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: +(1 row) +1: SELECT count(*), sum(a) FROM pscan_insert; +ERROR: fault triggered, fault name:'executor_pre_tuple_processed' fault type:'error' (seg0 slice1 172.17.0.3:7002 pid=6396) +SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE content = 0 AND role = 'p'; + gp_inject_fault +----------------- + Success: +(1 row) +1: SELECT count(*), sum(a) FROM pscan_insert; + count | sum +--------+------------- + 150000 | 11250075000 +(1 row) + +1q: ... +2q: ... +DROP TABLE pscan_insert; +DROP +DROP TABLE pscan_delete; +DROP +DROP TABLE pscan_vacuum; +DROP +DROP TABLE pscan_snapshot; +DROP diff --git a/src/test/regress/expected/cbdb_parallel.out b/src/test/regress/expected/cbdb_parallel.out index eb1f3e02e8b..b5c0796d693 100644 --- a/src/test/regress/expected/cbdb_parallel.out +++ b/src/test/regress/expected/cbdb_parallel.out @@ -621,6 +621,53 @@ commit; drop table ao1; drop table ao2; drop table aocs1; +-- +-- Parallel scan of AOCS columns with RLE compression. The other cases of +-- parallel AO row and AOCS scans are in input/uao_dml/ao_parallel_scan.source; +-- rle_type is AOCS-only. A parallel process positions every column at the +-- first row of its chunk, which is inside a block; rle_type packs regular +-- values densely, so a block holds many rows. t is mostly 32 bytes, but +-- every 1000th value is larger than a block. +-- +begin; +set local enable_parallel = on; +set local max_parallel_workers_per_gather = 4; +create table aocs_rle (k int, a int, + b bigint encoding (compresstype = rle_type, compresslevel = 4), + n numeric encoding (compresstype = rle_type, compresslevel = 1), + t text encoding (compresstype = rle_type, compresslevel = 2)) + using ao_column with (parallel_workers = 4) distributed by (k); +insert into aocs_rle + select 0, i, i::bigint * 7, i / 4.0, + case when i % 1000 = 0 then repeat(md5(i::text), 2000) else md5(i::text) end + from generate_series(1, 100000) i; +-- misaligned counts rows whose columns were read from different rows +explain(costs off) +select count(*), sum(a), + sum(case when b <> a::bigint * 7 or n <> a / 4.0 or + t <> case when a % 1000 = 0 then repeat(md5(a::text), 2000) else md5(a::text) end + then 1 else 0 end) as misaligned +from aocs_rle; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on aocs_rle + Optimizer: Postgres query optimizer +(5 rows) + +select count(*), sum(a), + sum(case when b <> a::bigint * 7 or n <> a / 4.0 or + t <> case when a % 1000 = 0 then repeat(md5(a::text), 2000) else md5(a::text) end + then 1 else 0 end) as misaligned +from aocs_rle; + count | sum | misaligned +--------+------------+------------ + 100000 | 5000050000 | 0 +(1 row) + +abort; -- test Parallel Bitmap Heap Scan begin; create table t1(c1 int, c2 int) with(parallel_workers=2) distributed by (c1); @@ -2204,7 +2251,7 @@ abort; -- -- Test Parallel Hash Left Anti Semi (Not-In) Join(parallel-oblivious). -- -create table t1(c1 int, c2 int) using ao_row distributed by (c1); +create table t1(c1 int, c2 int) using ao_row with(parallel_workers=2) distributed by (c1); create table t2(c1 int, c2 int) using ao_row distributed by (c1); create table t3_null(c1 int, c2 int) using ao_row distributed by (c1); begin; diff --git a/src/test/regress/expected/zlib.out b/src/test/regress/expected/zlib.out index 8825098cc2f..632e46bc8e3 100644 --- a/src/test/regress/expected/zlib.out +++ b/src/test/regress/expected/zlib.out @@ -13,7 +13,9 @@ SET gp_workfile_compression = on; DROP TABLE IF EXISTS test_zlib_hashjoin; NOTICE: table "test_zlib_hashjoin" does not exist, skipping -CREATE TABLE test_zlib_hashjoin (i1 int, i2 int, i3 int, i4 int, i5 int, i6 int, i7 int, i8 int) WITH (APPENDONLY=true) DISTRIBUTED BY (i1) ; +-- No parallel scan: a Parallel Hash Join spills to shared files, which are +-- not created by BufFileCreateTempInSet() and do not hit the fault below. +CREATE TABLE test_zlib_hashjoin (i1 int, i2 int, i3 int, i4 int, i5 int, i6 int, i7 int, i8 int) WITH (APPENDONLY=true, parallel_workers=0) DISTRIBUTED BY (i1) ; INSERT INTO test_zlib_hashjoin SELECT i,i,i,i,i,i,i,i FROM (select generate_series(1, nsegments * 333333) as i from (select count(*) as nsegments from gp_segment_configuration where role='p' and content >= 0) foo) bar; diff --git a/src/test/regress/greenplum_schedule b/src/test/regress/greenplum_schedule index 0480c20189c..164d7269bd6 100755 --- a/src/test/regress/greenplum_schedule +++ b/src/test/regress/greenplum_schedule @@ -299,6 +299,9 @@ test: freeze_aux_tables # cbdb parallel test test: cbdb_parallel +# parallel scans of append-optimized tables +test: uao_dml/ao_parallel_scan_row uao_dml/ao_parallel_scan_column + # ORCA parallel append over partitioned tables test: orca_parallel_append diff --git a/src/test/regress/input/uao_dml/ao_parallel_scan.source b/src/test/regress/input/uao_dml/ao_parallel_scan.source new file mode 100644 index 00000000000..244cbd865b1 --- /dev/null +++ b/src/test/regress/input/uao_dml/ao_parallel_scan.source @@ -0,0 +1,546 @@ +-- +-- Parallel sequential scans of AO row and AOCS tables. +-- +-- Each segment file is split into chunks of rows, and the parallel processes +-- of a segment take chunks from a shared counter (see +-- src/backend/access/appendonly/README.parallel_scan.md). An AO row block +-- belongs to the chunk that holds its first row; an AOCS scan moves every +-- column file to exactly the first row of its chunk. A process reaches its +-- chunk by reading the block headers in front of it. +-- +-- Most cases run :check, which returns +-- rows number of rows read +-- sum_a sum of column a +-- misaligned rows whose other columns do not belong to their column a, +-- i.e. the AOCS column readers were positioned at different +-- rows; always 0 +-- sum_rownum sum of the row numbers in the rows' ctids; it changes if a +-- row is returned twice or with another row's ctid +-- Column a holds 1..n, so rows and sum_a show whether every row was read +-- exactly once. The check queries avoid aggregate FILTER and count(DISTINCT) +-- on a row value, which ORCA cannot plan. +-- +-- Tables are distributed by a constant key, so all rows are on one segment +-- and the chunk layout of that segment is known. The number of rows in a +-- chunk depends on the average row size, which is not the same for AO row +-- and AOCS tables, so chunk_rows() computes it the way the scan does. +-- +create schema ao_parallel_scan_@amname@; +set search_path = ao_parallel_scan_@amname@; +set default_table_access_method = @amname@; + +set enable_parallel = on; +set optimizer = off; +set max_parallel_workers_per_gather = 4; + +\set check 'select count(*) as rows, sum(a) as sum_a, sum(case when c <> md5(a::text) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) as sum_rownum from' + +-- For a table of :n rows with a = 1..:n stored in that order. +\set check_n 'select count(*) = :n as rows_ok, sum(a) = :n::bigint * (:n + 1) / 2 as sum_a_ok, sum(case when c <> md5(a::text) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) = :n::bigint * (:n + 1) / 2 as sum_rownum_ok from' + +-- Segment files that a scan reads on the segment that holds the table: those +-- with rows that are not awaiting drop (state 2). Whether VACUUM can drop a +-- compacted segment file right away depends on other sessions' snapshots, so +-- empty and awaiting-drop segment files are left out to keep the output +-- stable. An AOCS segment file has a row for each column. +create function segfile_rows(rel regclass) +returns table (segno int, tupcount bigint) language sql as $$ + select distinct segno, tupcount + from gp_toolkit.__gp_@aoseg@(rel) + where tupcount > 0 and state <> 2 + order by segno +$$; + +-- The number of rows in a chunk, as in AppendOnlyParallelScan_RowsPerChunk(): +-- about 1/2048 of the rows, but at least 512kB and at most 32MB of +-- uncompressed data. +create function chunk_rows(rel regclass) returns bigint language sql as $$ + with segfiles as ( + select segno, tupcount, sum(eof_uncompressed) as bytes + from gp_toolkit.__gp_@aoseg@(rel) + where tupcount > 0 and state <> 2 + group by segment_id, segno, tupcount), + row_bytes as ( + select sum(tupcount)::bigint as total_rows, + greatest(sum(bytes)::bigint / sum(tupcount)::bigint, 1) as row_bytes + from segfiles) + select greatest(least(greatest(total_rows / 2048, 524288 / row_bytes), + 33554432 / row_bytes), 1) + from row_bytes +$$; + +-- +-- Chunk layout +-- + +-- One segment file split into several chunks. +create table single_segfile (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into single_segfile select 0, i, md5(i::text) from generate_series(1, 100000) i; +select * from segfile_rows('single_segfile'); +-- the chunk size of tables with these columns +select chunk_rows('single_segfile') as chunk_rows \gset +select 100000 / :chunk_rows >= 4 as several_chunks; +explain (costs off) :check single_segfile; +:check single_segfile; + +-- Several segment files of different sizes: an insert that spreads rows over +-- three segment files writes 20000, 20000 and 10000 rows, then a plain insert +-- appends 70000 more rows to the last one. +create table multi_segfile (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +set gp_appendonly_insert_files = 3; +set gp_appendonly_insert_files_tuples_range = 20000; +insert into multi_segfile select 0, i, md5(i::text) from generate_series(1, 50000) i; +reset gp_appendonly_insert_files; +reset gp_appendonly_insert_files_tuples_range; +insert into multi_segfile select 0, i, md5(i::text) from generate_series(50001, 120000) i; +select tupcount from segfile_rows('multi_segfile') order by 1; +explain (costs off) :check multi_segfile; +:check multi_segfile; + +-- Segment files without rows to scan between, before and after the ones +-- with rows. Three segment files of 30000 rows each; deleting all rows of one +-- and vacuuming leaves an empty segment file in its place, or one awaiting +-- drop if another session's snapshot can still see its rows. The scan must +-- skip both. +create table empty_segfiles (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +set gp_appendonly_insert_files = 3; +set gp_appendonly_insert_files_tuples_range = 30000; +insert into empty_segfiles select 0, i, md5(i::text) from generate_series(1, 90000) i; +reset gp_appendonly_insert_files; +reset gp_appendonly_insert_files_tuples_range; +-- rows 30001-60000 are in the second segment file: empty it +delete from empty_segfiles where a between 30001 and 60000; +vacuum empty_segfiles; +select * from segfile_rows('empty_segfiles'); +explain (costs off) :check empty_segfiles; +:check empty_segfiles; +-- rows 1-30000 are in the first segment file: empty it too +delete from empty_segfiles where a <= 30000; +vacuum empty_segfiles; +select * from segfile_rows('empty_segfiles'); +explain (costs off) :check empty_segfiles; +:check empty_segfiles; + +-- Row counts at a chunk boundary: exactly 3 chunks, one row less, and one +-- row more (a fourth chunk of a single row). Each table must have the chunk +-- size of single_segfile. +create table boundary_eq (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into boundary_eq select 0, i, md5(i::text) from generate_series(1, 3 * :chunk_rows) i; +select chunk_rows('boundary_eq') = :chunk_rows as same_chunk_rows; +select 3 * :chunk_rows as n \gset +explain (costs off) :check_n boundary_eq; +:check_n boundary_eq; +create table boundary_minus (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into boundary_minus select 0, i, md5(i::text) from generate_series(1, 3 * :chunk_rows - 1) i; +select chunk_rows('boundary_minus') = :chunk_rows as same_chunk_rows; +select 3 * :chunk_rows - 1 as n \gset +explain (costs off) :check_n boundary_minus; +:check_n boundary_minus; +create table boundary_plus (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into boundary_plus select 0, i, md5(i::text) from generate_series(1, 3 * :chunk_rows + 1) i; +select chunk_rows('boundary_plus') = :chunk_rows as same_chunk_rows; +select 3 * :chunk_rows + 1 as n \gset +explain (costs off) :check_n boundary_plus; +:check_n boundary_plus; + +-- Fewer chunks than workers: a table of one short chunk, and an empty table. +create table tiny (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into tiny select 0, i, md5(i::text) from generate_series(1, 10) i; +explain (costs off) :check tiny; +:check tiny; +create table empty (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +explain (costs off) :check empty; +:check empty; + +-- +-- Positioning +-- +-- At the first row of a chunk a process reads the block headers in front of +-- it. For AOCS every column file is positioned separately; columns of +-- different widths hold different numbers of rows per block, so each column +-- file stops at a different block. All tables below have the same columns, +-- each derived from a, so misaligned counts the rows where any column was +-- read from another row than a. +-- + +-- b and n are fixed width, t is mostly 32 bytes, but every 1000th value is +-- 64000 bytes: larger than a block, so it is stored over several blocks. +create function t_for(a int) returns text language sql immutable parallel safe as $$ + select case when a % 1000 = 0 then repeat(md5(a::text), 2000) + else md5(a::text) end +$$; + +\set check_cols 'select count(*) as rows, sum(a) as sum_a, sum(case when b <> a::bigint * 7 or n <> a / 4.0 or t <> t_for(a) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) as sum_rownum from' + +-- Columns of different widths. +create table wide_cols (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4) distributed by (k); +insert into wide_cols select 0, i, i::bigint * 7, i / 4.0, t_for(i) from generate_series(1, 100000) i; +explain (costs off) :check_cols wide_cols; +:check_cols wide_cols; + +-- Projections reading only some of the columns. +explain (costs off) select sum(n) from wide_cols; +select sum(n) from wide_cols; +explain (costs off) select sum(b), sum(length(t)) from wide_cols; +select sum(b), sum(length(t)) from wide_cols; + +-- Small blocks: many blocks per chunk. +create table block_8k (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, blocksize = 8192) distributed by (k); +insert into block_8k select * from wide_cols; +explain (costs off) :check_cols block_8k; +:check_cols block_8k; + +-- Large blocks: a block holds more rows than a chunk, so the chunks after the +-- first one in a block have no block of their own. +create table block_2m (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, blocksize = 2097152) distributed by (k); +insert into block_2m select * from wide_cols; +explain (costs off) :check_cols block_2m; +:check_cols block_2m; + +-- Compressed blocks. Only the block holding the first row of a chunk is +-- decompressed; the blocks in front of it are skipped by their headers. +create table compress_zstd (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, compresstype = zstd, compresslevel = 1) distributed by (k); +insert into compress_zstd select * from wide_cols; +explain (costs off) :check_cols compress_zstd; +:check_cols compress_zstd; +create table compress_zlib (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, compresstype = zlib, compresslevel = 1) distributed by (k); +insert into compress_zlib select * from wide_cols; +explain (costs off) :check_cols compress_zlib; +:check_cols compress_zlib; + +-- A column added after rows were loaded: rows inserted before the ALTER get +-- the default, rows inserted after it their own value. The ALTER rewrites +-- an AO row table but not an AOCS one, so the row numbers differ; compare +-- them with those of a serial scan. +create table added_col (k int, a int, t text) + with (parallel_workers = 4) distributed by (k); +insert into added_col select 0, i, md5(i::text) from generate_series(1, 60000) i; +alter table added_col add column b bigint default -1; +insert into added_col select 0, i, md5(i::text), i::bigint * 7 from generate_series(60001, 100000) i; +set enable_parallel = off; +select sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) as serial_rownum +from added_col \gset +set enable_parallel = on; +\set check_added 'select count(*) as rows, sum(a) as sum_a, sum(b) as sum_b, sum(case when t <> md5(a::text) or b <> case when a <= 60000 then -1 else a::bigint * 7 end then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) = :serial_rownum as sum_rownum_ok from' +explain (costs off) :check_added added_col; +:check_added added_col; + +-- +-- Row visibility +-- +-- Chunks cover every row stored in a segment file, including deleted ones. +-- The scan drops deleted rows with the visibility map, which it looks up by +-- the row number of each row, so sum_rownum also checks those row numbers. +-- + +-- Every 7th row deleted: every chunk has both visible and deleted rows. +create table vis_scattered (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_scattered select 0, i, md5(i::text) from generate_series(1, 100000) i; +delete from vis_scattered where a % 7 = 0; +explain (costs off) :check vis_scattered; +:check vis_scattered; + +-- The rows of the second, third and fourth chunks deleted: those chunks have +-- no visible rows. +create table vis_chunks (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_chunks select 0, i, md5(i::text) from generate_series(1, 100000) i; +delete from vis_chunks where a between :chunk_rows + 1 and 4 * :chunk_rows; +\set check_vis_chunks 'select count(*) = 100000 - 3 * :chunk_rows as rows_ok, sum(a) = 5000050000 - (:chunk_rows + 1 + 4 * :chunk_rows)::bigint * 3 * :chunk_rows / 2 as sum_a_ok, sum(case when c <> md5(a::text) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) = sum(a) as sum_rownum_ok from' +explain (costs off) :check_vis_chunks vis_chunks; +:check_vis_chunks vis_chunks; + +-- Every 5th row updated: the old version is marked deleted and the new one is +-- appended, so the segment file holds both. +create table vis_updated (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_updated select 0, i, md5(i::text) from generate_series(1, 100000) i; +update vis_updated set a = -a, c = md5((-a)::text) where a % 5 = 0; +explain (costs off) :check vis_updated; +:check vis_updated; + +-- Gaps in row numbers: an aborted insert and many small inserts use up row +-- numbers that no stored row has, so row numbers no longer follow row +-- positions. Row numbers of the blocks a process skips must still be +-- counted. +create table vis_gaps (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_gaps select 0, i, md5(i::text) from generate_series(1, 30000) i; +begin; +insert into vis_gaps select 0, i, md5(i::text) from generate_series(1, 50000) i; +abort; +do $$ +begin + for j in 0..99 loop + insert into vis_gaps + select 0, i, md5(i::text) from generate_series(30001 + j * 500, 30500 + j * 500) i; + end loop; +end $$; +insert into vis_gaps select 0, i, md5(i::text) from generate_series(80001, 100000) i; +select count(*) as rows, + max(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) > count(*) as has_gaps +from vis_gaps; +explain (costs off) :check vis_gaps; +:check vis_gaps; + +-- +-- Filters +-- +-- A filter can be evaluated inside the scan. For AOCS that is +-- aocs_getnext_withqual(), whose first gp_predicate_pushdown_sample_rows rows +-- go through aocs_getnext_sample(). Both loops must stop at the end of a +-- chunk. +-- +create table filt (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into filt select 0, i, md5(i::text) from generate_series(1, 100000) i; + +-- One column in the filter: 10000 rows match. +explain (costs off) :check filt where a % 10 = 3; +:check filt where a % 10 = 3; + +-- Two columns in the filter. +explain (costs off) :check filt where a % 10 = 3 and c > '8'; +:check filt where a % 10 = 3 and c > '8'; + +-- No row matches, and every row matches. +explain (costs off) :check filt where a < 0; +:check filt where a < 0; +explain (costs off) :check filt where a > 0; +:check filt where a > 0; + +-- Sampling over more rows than a chunk, so the sampling loop reaches the end +-- of chunks; and no sampling at all. +set gp_predicate_pushdown_sample_rows = 50000; +explain (costs off) :check filt where a % 10 = 3; +:check filt where a % 10 = 3; +set gp_predicate_pushdown_sample_rows = 0; +explain (costs off) :check filt where a % 10 = 3; +:check filt where a % 10 = 3; +reset gp_predicate_pushdown_sample_rows; + +-- +-- Joins and aggregates +-- + +-- A parallel hash join of two parallel scans. +explain (costs off) +select count(*), sum(x.a) from filt x join filt y on x.a = y.a; +select count(*), sum(x.a) from filt x join filt y on x.a = y.a; + +-- Partial and final aggregation. +explain (costs off) +select a % 7 as g, count(*), sum(a) from filt group by 1 order by 1; +select a % 7 as g, count(*), sum(a) from filt group by 1 order by 1; + +-- +-- Partitioned tables +-- +-- Each partition is split into its own chunks. +-- +create table part (k int, a int, c text) partition by range (a) distributed by (k); +create table part_1 partition of part for values from (1) to (30001) + with (parallel_workers = 4); +create table part_2 partition of part for values from (30001) to (60001) + with (parallel_workers = 4); +create table part_3 partition of part for values from (60001) to (90001) + with (parallel_workers = 4); +insert into part select 0, i, md5(i::text) from generate_series(1, 90000) i; +analyze part; +explain (costs off) :check part; +:check part; + +-- +-- Rescans +-- +-- A parallel scan hands out each chunk once, so it cannot be scanned again. +-- Neither planner rescans one: it becomes the outer side of a nested loop, is +-- read once into a Materialize, or the plan is not parallel. These queries +-- have shapes where a rescan could happen. +-- +create table outer_rows (a int) + with (parallel_workers = 4) distributed by (a); +insert into outer_rows select i * 1000 from generate_series(1, 20) i; +create table inner_rows (k int, a int, c text) + with (parallel_workers = 4) distributed by (a); +insert into inner_rows select i % 7, i, md5(i::text) from generate_series(1, 200000) i; +analyze outer_rows; +analyze inner_rows; + +set enable_hashjoin = off; +set enable_mergejoin = off; +set enable_material = off; + +explain (costs off) +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; + +explain (costs off) +select sum(n) from + (select (select count(*) from inner_rows i where i.a < o.a and i.k = 3) as n + from outer_rows o) s; +select sum(n) from + (select (select count(*) from inner_rows i where i.a < o.a and i.k = 3) as n + from outer_rows o) s; + +explain (costs off) +select count(*) from outer_rows o + where exists (select 1 from inner_rows i where i.a = o.a + 1); +select count(*) from outer_rows o + where exists (select 1 from inner_rows i where i.a = o.a + 1); + +explain (costs off) +select count(*) from outer_rows o + where o.a in (select i.a from inner_rows i where i.k = o.a % 7); +select count(*) from outer_rows o + where o.a in (select i.a from inner_rows i where i.k = o.a % 7); + +explain (costs off) +select count(*), sum(l.n) from outer_rows o, + lateral (select count(*) as n from inner_rows i where i.a < o.a) l; +select count(*), sum(l.n) from outer_rows o, + lateral (select count(*) as n from inner_rows i where i.a < o.a) l; + +reset enable_material; +reset enable_mergejoin; +reset enable_hashjoin; + +-- The same plan executed several times. +prepare inner_rows_count(int) as select count(*), sum(a) from inner_rows where k = $1; +explain (costs off) execute inner_rows_count(3); +execute inner_rows_count(3); +execute inner_rows_count(4); +execute inner_rows_count(3); +deallocate inner_rows_count; + +-- A cursor. +begin; +declare c cursor for select count(*), sum(a) from inner_rows where k = 3; +fetch all from c; +close c; +commit; + +-- +-- Planner +-- +-- With the Postgres planner, a table without the parallel_workers option +-- gets workers according to its size, like a heap table, because its +-- segment files are split into chunks. +-- +create table plan_small (a int, c text) distributed by (a); +insert into plan_small select i, md5(i::text) from generate_series(1, 1000) i; +create table plan_mid (a int, c text) distributed by (a); +insert into plan_mid select i, md5(i::text) from generate_series(1, 100000) i; +create table plan_big (a int, c text) distributed by (a); +insert into plan_big select i, md5(i::text) from generate_series(1, 1000000) i; +create table plan_option (a int, c text) + with (parallel_workers = 3) distributed by (a); +insert into plan_option select i, md5(i::text) from generate_series(1, 1000) i; +create table plan_three (a int, c text) distributed by (a); +set gp_appendonly_insert_files = 3; +set gp_appendonly_insert_files_tuples_range = 10000; +insert into plan_three select i, md5(i::text) from generate_series(1, 1000000) i; +reset gp_appendonly_insert_files; +reset gp_appendonly_insert_files_tuples_range; +analyze plan_small; +analyze plan_mid; +analyze plan_big; +analyze plan_option; +analyze plan_three; +select c.relname, a.segfilecount +from pg_class c join pg_appendonly a on a.relid = c.oid +where c.relnamespace = 'ao_parallel_scan_@amname@'::regnamespace + and c.relname like 'plan\_%' order by 1; + +set min_parallel_table_scan_size = '8MB'; + +-- One segment file, and still a parallel plan. +explain (costs off) select count(*) from plan_big; + +-- A table smaller than min_parallel_table_scan_size is scanned serially. +explain (costs off) select count(*) from plan_small; + +-- The parallel_workers option overrides the size: 3 workers per segment. +explain (costs off) select count(*) from plan_option; + +-- A larger table gets more workers, up to max_parallel_workers_per_gather. +set min_parallel_table_scan_size = '128kB'; +set max_parallel_workers_per_gather = 8; +explain (costs off) select count(*) from plan_mid; +explain (costs off) select count(*) from plan_big; +set max_parallel_workers_per_gather = 2; +explain (costs off) select count(*) from plan_big; +set max_parallel_workers_per_gather = 8; + +-- The number of segment files does not matter: three segment files get the +-- same workers as the one of plan_big. +explain (costs off) select count(*) from plan_three; + +-- No parallel index scans. +create index plan_big_a on plan_big (a); +set enable_seqscan = off; +set min_parallel_index_scan_size = 0; +explain (costs off) select count(*) from plan_big where a < 100; +reset min_parallel_index_scan_size; +reset enable_seqscan; + +reset min_parallel_table_scan_size; +set max_parallel_workers_per_gather = 4; + +-- +-- ORCA +-- +-- ORCA plans a parallel scan only where it needs no rewinding, and puts a +-- Materialize above it on the inner side of a nested loop. +-- +set optimizer = on; +analyze empty_segfiles; + +explain (costs off) :check single_segfile; +:check single_segfile; +explain (costs off) :check empty_segfiles; +:check empty_segfiles; +explain (costs off) :check_cols compress_zstd; +:check_cols compress_zstd; +explain (costs off) :check vis_gaps; +:check vis_gaps; + +-- LIMIT: the processes stop before all chunks are handed out. The Postgres +-- planner does not choose a parallel scan for these. +explain (costs off) select count(*) from (select a from filt limit 10) s; +select count(*) from (select a from filt limit 10) s; +explain (costs off) select count(*) from (select a from filt limit 20000) s; +select count(*) from (select a from filt limit 20000) s; + +set optimizer_enable_parallel_append = on; +explain (costs off) :check part; +:check part; +reset optimizer_enable_parallel_append; + +set enable_hashjoin = off; +set enable_mergejoin = off; +explain (costs off) +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; +reset enable_mergejoin; +reset enable_hashjoin; + +reset optimizer; +reset max_parallel_workers_per_gather; +reset enable_parallel; +reset default_table_access_method; +drop schema ao_parallel_scan_@amname@ cascade; diff --git a/src/test/regress/output/uao_dml/ao_parallel_scan.source b/src/test/regress/output/uao_dml/ao_parallel_scan.source new file mode 100644 index 00000000000..03de5013583 --- /dev/null +++ b/src/test/regress/output/uao_dml/ao_parallel_scan.source @@ -0,0 +1,1372 @@ +-- +-- Parallel sequential scans of AO row and AOCS tables. +-- +-- Each segment file is split into chunks of rows, and the parallel processes +-- of a segment take chunks from a shared counter (see +-- src/backend/access/appendonly/README.parallel_scan.md). An AO row block +-- belongs to the chunk that holds its first row; an AOCS scan moves every +-- column file to exactly the first row of its chunk. A process reaches its +-- chunk by reading the block headers in front of it. +-- +-- Most cases run :check, which returns +-- rows number of rows read +-- sum_a sum of column a +-- misaligned rows whose other columns do not belong to their column a, +-- i.e. the AOCS column readers were positioned at different +-- rows; always 0 +-- sum_rownum sum of the row numbers in the rows' ctids; it changes if a +-- row is returned twice or with another row's ctid +-- Column a holds 1..n, so rows and sum_a show whether every row was read +-- exactly once. The check queries avoid aggregate FILTER and count(DISTINCT) +-- on a row value, which ORCA cannot plan. +-- +-- Tables are distributed by a constant key, so all rows are on one segment +-- and the chunk layout of that segment is known. The number of rows in a +-- chunk depends on the average row size, which is not the same for AO row +-- and AOCS tables, so chunk_rows() computes it the way the scan does. +-- +create schema ao_parallel_scan_@amname@; +set search_path = ao_parallel_scan_@amname@; +set default_table_access_method = @amname@; +set enable_parallel = on; +set optimizer = off; +set max_parallel_workers_per_gather = 4; +\set check 'select count(*) as rows, sum(a) as sum_a, sum(case when c <> md5(a::text) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) as sum_rownum from' +-- For a table of :n rows with a = 1..:n stored in that order. +\set check_n 'select count(*) = :n as rows_ok, sum(a) = :n::bigint * (:n + 1) / 2 as sum_a_ok, sum(case when c <> md5(a::text) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) = :n::bigint * (:n + 1) / 2 as sum_rownum_ok from' +-- Segment files that a scan reads on the segment that holds the table: those +-- with rows that are not awaiting drop (state 2). Whether VACUUM can drop a +-- compacted segment file right away depends on other sessions' snapshots, so +-- empty and awaiting-drop segment files are left out to keep the output +-- stable. An AOCS segment file has a row for each column. +create function segfile_rows(rel regclass) +returns table (segno int, tupcount bigint) language sql as $$ + select distinct segno, tupcount + from gp_toolkit.__gp_@aoseg@(rel) + where tupcount > 0 and state <> 2 + order by segno +$$; +-- The number of rows in a chunk, as in AppendOnlyParallelScan_RowsPerChunk(): +-- about 1/2048 of the rows, but at least 512kB and at most 32MB of +-- uncompressed data. +create function chunk_rows(rel regclass) returns bigint language sql as $$ + with segfiles as ( + select segno, tupcount, sum(eof_uncompressed) as bytes + from gp_toolkit.__gp_@aoseg@(rel) + where tupcount > 0 and state <> 2 + group by segment_id, segno, tupcount), + row_bytes as ( + select sum(tupcount)::bigint as total_rows, + greatest(sum(bytes)::bigint / sum(tupcount)::bigint, 1) as row_bytes + from segfiles) + select greatest(least(greatest(total_rows / 2048, 524288 / row_bytes), + 33554432 / row_bytes), 1) + from row_bytes +$$; +-- +-- Chunk layout +-- +-- One segment file split into several chunks. +create table single_segfile (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into single_segfile select 0, i, md5(i::text) from generate_series(1, 100000) i; +select * from segfile_rows('single_segfile'); + segno | tupcount +-------+---------- + 1 | 100000 +(1 row) + +-- the chunk size of tables with these columns +select chunk_rows('single_segfile') as chunk_rows \gset +select 100000 / :chunk_rows >= 4 as several_chunks; + several_chunks +---------------- + t +(1 row) + +explain (costs off) :check single_segfile; + QUERY PLAN +------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on single_segfile + Optimizer: Postgres query optimizer +(5 rows) + +:check single_segfile; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +-- Several segment files of different sizes: an insert that spreads rows over +-- three segment files writes 20000, 20000 and 10000 rows, then a plain insert +-- appends 70000 more rows to the last one. +create table multi_segfile (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +set gp_appendonly_insert_files = 3; +set gp_appendonly_insert_files_tuples_range = 20000; +insert into multi_segfile select 0, i, md5(i::text) from generate_series(1, 50000) i; +reset gp_appendonly_insert_files; +reset gp_appendonly_insert_files_tuples_range; +insert into multi_segfile select 0, i, md5(i::text) from generate_series(50001, 120000) i; +select tupcount from segfile_rows('multi_segfile') order by 1; + tupcount +---------- + 20000 + 20000 + 80000 +(3 rows) + +explain (costs off) :check multi_segfile; + QUERY PLAN +------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on multi_segfile + Optimizer: Postgres query optimizer +(5 rows) + +:check multi_segfile; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 120000 | 7200060000 | 0 | 3607060000 +(1 row) + +-- Segment files without rows to scan between, before and after the ones +-- with rows. Three segment files of 30000 rows each; deleting all rows of one +-- and vacuuming leaves an empty segment file in its place, or one awaiting +-- drop if another session's snapshot can still see its rows. The scan must +-- skip both. +create table empty_segfiles (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +set gp_appendonly_insert_files = 3; +set gp_appendonly_insert_files_tuples_range = 30000; +insert into empty_segfiles select 0, i, md5(i::text) from generate_series(1, 90000) i; +reset gp_appendonly_insert_files; +reset gp_appendonly_insert_files_tuples_range; +-- rows 30001-60000 are in the second segment file: empty it +delete from empty_segfiles where a between 30001 and 60000; +vacuum empty_segfiles; +select * from segfile_rows('empty_segfiles'); + segno | tupcount +-------+---------- + 1 | 30000 + 3 | 30000 +(2 rows) + +explain (costs off) :check empty_segfiles; + QUERY PLAN +------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on empty_segfiles + Optimizer: Postgres query optimizer +(5 rows) + +:check empty_segfiles; + rows | sum_a | misaligned | sum_rownum +-------+------------+------------+------------ + 60000 | 2700030000 | 0 | 900030000 +(1 row) + +-- rows 1-30000 are in the first segment file: empty it too +delete from empty_segfiles where a <= 30000; +vacuum empty_segfiles; +select * from segfile_rows('empty_segfiles'); + segno | tupcount +-------+---------- + 3 | 30000 +(1 row) + +explain (costs off) :check empty_segfiles; + QUERY PLAN +------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on empty_segfiles + Optimizer: Postgres query optimizer +(5 rows) + +:check empty_segfiles; + rows | sum_a | misaligned | sum_rownum +-------+------------+------------+------------ + 30000 | 2250015000 | 0 | 450015000 +(1 row) + +-- Row counts at a chunk boundary: exactly 3 chunks, one row less, and one +-- row more (a fourth chunk of a single row). Each table must have the chunk +-- size of single_segfile. +create table boundary_eq (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into boundary_eq select 0, i, md5(i::text) from generate_series(1, 3 * :chunk_rows) i; +select chunk_rows('boundary_eq') = :chunk_rows as same_chunk_rows; + same_chunk_rows +----------------- + t +(1 row) + +select 3 * :chunk_rows as n \gset +explain (costs off) :check_n boundary_eq; + QUERY PLAN +---------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on boundary_eq + Optimizer: Postgres query optimizer +(5 rows) + +:check_n boundary_eq; + rows_ok | sum_a_ok | misaligned | sum_rownum_ok +---------+----------+------------+--------------- + t | t | 0 | t +(1 row) + +create table boundary_minus (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into boundary_minus select 0, i, md5(i::text) from generate_series(1, 3 * :chunk_rows - 1) i; +select chunk_rows('boundary_minus') = :chunk_rows as same_chunk_rows; + same_chunk_rows +----------------- + t +(1 row) + +select 3 * :chunk_rows - 1 as n \gset +explain (costs off) :check_n boundary_minus; + QUERY PLAN +------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on boundary_minus + Optimizer: Postgres query optimizer +(5 rows) + +:check_n boundary_minus; + rows_ok | sum_a_ok | misaligned | sum_rownum_ok +---------+----------+------------+--------------- + t | t | 0 | t +(1 row) + +create table boundary_plus (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into boundary_plus select 0, i, md5(i::text) from generate_series(1, 3 * :chunk_rows + 1) i; +select chunk_rows('boundary_plus') = :chunk_rows as same_chunk_rows; + same_chunk_rows +----------------- + t +(1 row) + +select 3 * :chunk_rows + 1 as n \gset +explain (costs off) :check_n boundary_plus; + QUERY PLAN +------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on boundary_plus + Optimizer: Postgres query optimizer +(5 rows) + +:check_n boundary_plus; + rows_ok | sum_a_ok | misaligned | sum_rownum_ok +---------+----------+------------+--------------- + t | t | 0 | t +(1 row) + +-- Fewer chunks than workers: a table of one short chunk, and an empty table. +create table tiny (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into tiny select 0, i, md5(i::text) from generate_series(1, 10) i; +explain (costs off) :check tiny; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on tiny + Optimizer: Postgres query optimizer +(5 rows) + +:check tiny; + rows | sum_a | misaligned | sum_rownum +------+-------+------------+------------ + 10 | 55 | 0 | 55 +(1 row) + +create table empty (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +explain (costs off) :check empty; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on empty + Optimizer: Postgres query optimizer +(5 rows) + +:check empty; + rows | sum_a | misaligned | sum_rownum +------+-------+------------+------------ + 0 | | | +(1 row) + +-- +-- Positioning +-- +-- At the first row of a chunk a process reads the block headers in front of +-- it. For AOCS every column file is positioned separately; columns of +-- different widths hold different numbers of rows per block, so each column +-- file stops at a different block. All tables below have the same columns, +-- each derived from a, so misaligned counts the rows where any column was +-- read from another row than a. +-- +-- b and n are fixed width, t is mostly 32 bytes, but every 1000th value is +-- 64000 bytes: larger than a block, so it is stored over several blocks. +create function t_for(a int) returns text language sql immutable parallel safe as $$ + select case when a % 1000 = 0 then repeat(md5(a::text), 2000) + else md5(a::text) end +$$; +\set check_cols 'select count(*) as rows, sum(a) as sum_a, sum(case when b <> a::bigint * 7 or n <> a / 4.0 or t <> t_for(a) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) as sum_rownum from' +-- Columns of different widths. +create table wide_cols (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4) distributed by (k); +insert into wide_cols select 0, i, i::bigint * 7, i / 4.0, t_for(i) from generate_series(1, 100000) i; +explain (costs off) :check_cols wide_cols; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on wide_cols + Optimizer: Postgres query optimizer +(5 rows) + +:check_cols wide_cols; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +-- Projections reading only some of the columns. +explain (costs off) select sum(n) from wide_cols; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on wide_cols + Optimizer: Postgres query optimizer +(5 rows) + +select sum(n) from wide_cols; + sum +--------------------------------- + 1250012500.00000000000000000000 +(1 row) + +explain (costs off) select sum(b), sum(length(t)) from wide_cols; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on wide_cols + Optimizer: Postgres query optimizer +(5 rows) + +select sum(b), sum(length(t)) from wide_cols; + sum | sum +-------------+--------- + 35000350000 | 9596800 +(1 row) + +-- Small blocks: many blocks per chunk. +create table block_8k (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, blocksize = 8192) distributed by (k); +insert into block_8k select * from wide_cols; +explain (costs off) :check_cols block_8k; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on block_8k + Optimizer: Postgres query optimizer +(5 rows) + +:check_cols block_8k; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +-- Large blocks: a block holds more rows than a chunk, so the chunks after the +-- first one in a block have no block of their own. +create table block_2m (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, blocksize = 2097152) distributed by (k); +insert into block_2m select * from wide_cols; +explain (costs off) :check_cols block_2m; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on block_2m + Optimizer: Postgres query optimizer +(5 rows) + +:check_cols block_2m; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +-- Compressed blocks. Only the block holding the first row of a chunk is +-- decompressed; the blocks in front of it are skipped by their headers. +create table compress_zstd (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, compresstype = zstd, compresslevel = 1) distributed by (k); +insert into compress_zstd select * from wide_cols; +explain (costs off) :check_cols compress_zstd; + QUERY PLAN +------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on compress_zstd + Optimizer: Postgres query optimizer +(5 rows) + +:check_cols compress_zstd; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +create table compress_zlib (k int, a int, b bigint, n numeric, t text) + with (parallel_workers = 4, compresstype = zlib, compresslevel = 1) distributed by (k); +insert into compress_zlib select * from wide_cols; +explain (costs off) :check_cols compress_zlib; + QUERY PLAN +------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on compress_zlib + Optimizer: Postgres query optimizer +(5 rows) + +:check_cols compress_zlib; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +-- A column added after rows were loaded: rows inserted before the ALTER get +-- the default, rows inserted after it their own value. The ALTER rewrites +-- an AO row table but not an AOCS one, so the row numbers differ; compare +-- them with those of a serial scan. +create table added_col (k int, a int, t text) + with (parallel_workers = 4) distributed by (k); +insert into added_col select 0, i, md5(i::text) from generate_series(1, 60000) i; +alter table added_col add column b bigint default -1; +insert into added_col select 0, i, md5(i::text), i::bigint * 7 from generate_series(60001, 100000) i; +set enable_parallel = off; +select sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) as serial_rownum +from added_col \gset +set enable_parallel = on; +\set check_added 'select count(*) as rows, sum(a) as sum_a, sum(b) as sum_b, sum(case when t <> md5(a::text) or b <> case when a <= 60000 then -1 else a::bigint * 7 end then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) = :serial_rownum as sum_rownum_ok from' +explain (costs off) :check_added added_col; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on added_col + Optimizer: Postgres query optimizer +(5 rows) + +:check_added added_col; + rows | sum_a | sum_b | misaligned | sum_rownum_ok +--------+------------+-------------+------------+--------------- + 100000 | 5000050000 | 22400080000 | 0 | t +(1 row) + +-- +-- Row visibility +-- +-- Chunks cover every row stored in a segment file, including deleted ones. +-- The scan drops deleted rows with the visibility map, which it looks up by +-- the row number of each row, so sum_rownum also checks those row numbers. +-- +-- Every 7th row deleted: every chunk has both visible and deleted rows. +create table vis_scattered (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_scattered select 0, i, md5(i::text) from generate_series(1, 100000) i; +delete from vis_scattered where a % 7 = 0; +explain (costs off) :check vis_scattered; + QUERY PLAN +------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on vis_scattered + Optimizer: Postgres query optimizer +(5 rows) + +:check vis_scattered; + rows | sum_a | misaligned | sum_rownum +-------+------------+------------+------------ + 85715 | 4285785715 | 0 | 4285785715 +(1 row) + +-- The rows of the second, third and fourth chunks deleted: those chunks have +-- no visible rows. +create table vis_chunks (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_chunks select 0, i, md5(i::text) from generate_series(1, 100000) i; +delete from vis_chunks where a between :chunk_rows + 1 and 4 * :chunk_rows; +\set check_vis_chunks 'select count(*) = 100000 - 3 * :chunk_rows as rows_ok, sum(a) = 5000050000 - (:chunk_rows + 1 + 4 * :chunk_rows)::bigint * 3 * :chunk_rows / 2 as sum_a_ok, sum(case when c <> md5(a::text) then 1 else 0 end) as misaligned, sum(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) = sum(a) as sum_rownum_ok from' +explain (costs off) :check_vis_chunks vis_chunks; + QUERY PLAN +--------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on vis_chunks + Optimizer: Postgres query optimizer +(5 rows) + +:check_vis_chunks vis_chunks; + rows_ok | sum_a_ok | misaligned | sum_rownum_ok +---------+----------+------------+--------------- + t | t | 0 | t +(1 row) + +-- Every 5th row updated: the old version is marked deleted and the new one is +-- appended, so the segment file holds both. +create table vis_updated (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_updated select 0, i, md5(i::text) from generate_series(1, 100000) i; +update vis_updated set a = -a, c = md5((-a)::text) where a % 5 = 0; +explain (costs off) :check vis_updated; + QUERY PLAN +---------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on vis_updated + Optimizer: Postgres query optimizer +(5 rows) + +:check vis_updated; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 2999950000 | 0 | 6202010000 +(1 row) + +-- Gaps in row numbers: an aborted insert and many small inserts use up row +-- numbers that no stored row has, so row numbers no longer follow row +-- positions. Row numbers of the blocks a process skips must still be +-- counted. +create table vis_gaps (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into vis_gaps select 0, i, md5(i::text) from generate_series(1, 30000) i; +begin; +insert into vis_gaps select 0, i, md5(i::text) from generate_series(1, 50000) i; +abort; +do $$ +begin + for j in 0..99 loop + insert into vis_gaps + select 0, i, md5(i::text) from generate_series(30001 + j * 500, 30500 + j * 500) i; + end loop; +end $$; +insert into vis_gaps select 0, i, md5(i::text) from generate_series(80001, 100000) i; +select count(*) as rows, + max(((ctid::text::point)[0]::bigint % 33554432) * 32768 + (ctid::text::point)[1]::bigint - 1) > count(*) as has_gaps +from vis_gaps; + rows | has_gaps +--------+---------- + 100000 | t +(1 row) + +explain (costs off) :check vis_gaps; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on vis_gaps + Optimizer: Postgres query optimizer +(5 rows) + +:check vis_gaps; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 8961550000 +(1 row) + +-- +-- Filters +-- +-- A filter can be evaluated inside the scan. For AOCS that is +-- aocs_getnext_withqual(), whose first gp_predicate_pushdown_sample_rows rows +-- go through aocs_getnext_sample(). Both loops must stop at the end of a +-- chunk. +-- +create table filt (k int, a int, c text) + with (parallel_workers = 4) distributed by (k); +insert into filt select 0, i, md5(i::text) from generate_series(1, 100000) i; +-- One column in the filter: 10000 rows match. +explain (costs off) :check filt where a % 10 = 3; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on filt + Filter: ((a % 10) = 3) + Optimizer: Postgres query optimizer +(6 rows) + +:check filt where a % 10 = 3; + rows | sum_a | misaligned | sum_rownum +-------+-----------+------------+------------ + 10000 | 499980000 | 0 | 499980000 +(1 row) + +-- Two columns in the filter. +explain (costs off) :check filt where a % 10 = 3 and c > '8'; + QUERY PLAN +------------------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on filt + Filter: ((c > '8'::text) AND ((a % 10) = 3)) + Optimizer: Postgres query optimizer +(6 rows) + +:check filt where a % 10 = 3 and c > '8'; + rows | sum_a | misaligned | sum_rownum +------+-----------+------------+------------ + 4957 | 247106691 | 0 | 247106691 +(1 row) + +-- No row matches, and every row matches. +explain (costs off) :check filt where a < 0; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on filt + Filter: (a < 0) + Optimizer: Postgres query optimizer +(6 rows) + +:check filt where a < 0; + rows | sum_a | misaligned | sum_rownum +------+-------+------------+------------ + 0 | | | +(1 row) + +explain (costs off) :check filt where a > 0; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on filt + Filter: (a > 0) + Optimizer: Postgres query optimizer +(6 rows) + +:check filt where a > 0; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +-- Sampling over more rows than a chunk, so the sampling loop reaches the end +-- of chunks; and no sampling at all. +set gp_predicate_pushdown_sample_rows = 50000; +explain (costs off) :check filt where a % 10 = 3; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on filt + Filter: ((a % 10) = 3) + Optimizer: Postgres query optimizer +(6 rows) + +:check filt where a % 10 = 3; + rows | sum_a | misaligned | sum_rownum +-------+-----------+------------+------------ + 10000 | 499980000 | 0 | 499980000 +(1 row) + +set gp_predicate_pushdown_sample_rows = 0; +explain (costs off) :check filt where a % 10 = 3; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on filt + Filter: ((a % 10) = 3) + Optimizer: Postgres query optimizer +(6 rows) + +:check filt where a % 10 = 3; + rows | sum_a | misaligned | sum_rownum +-------+-----------+------------+------------ + 10000 | 499980000 | 0 | 499980000 +(1 row) + +reset gp_predicate_pushdown_sample_rows; +-- +-- Joins and aggregates +-- +-- A parallel hash join of two parallel scans. +explain (costs off) +select count(*), sum(x.a) from filt x join filt y on x.a = y.a; + QUERY PLAN +--------------------------------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Hash Join + Hash Cond: (x.a = y.a) + -> Redistribute Motion 12:12 (slice2; segments: 12) + Hash Key: x.a + Hash Module: 3 + -> Parallel Seq Scan on filt x + -> Parallel Hash + -> Redistribute Motion 12:12 (slice3; segments: 12) + Hash Key: y.a + Hash Module: 3 + -> Parallel Seq Scan on filt y + Optimizer: Postgres query optimizer +(15 rows) + +select count(*), sum(x.a) from filt x join filt y on x.a = y.a; + count | sum +--------+------------ + 100000 | 5000050000 +(1 row) + +-- Partial and final aggregation. +explain (costs off) +select a % 7 as g, count(*), sum(a) from filt group by 1 order by 1; + QUERY PLAN +--------------------------------------------------------------------- + Gather Motion 12:1 (slice1; segments: 12) + Merge Key: ((a % 7)) + -> Sort + Sort Key: ((a % 7)) + -> Finalize HashAggregate + Group Key: ((a % 7)) + -> Redistribute Motion 12:12 (slice2; segments: 12) + Hash Key: ((a % 7)) + Hash Module: 3 + -> Partial HashAggregate + Group Key: (a % 7) + -> Parallel Seq Scan on filt + Optimizer: Postgres query optimizer +(13 rows) + +select a % 7 as g, count(*), sum(a) from filt group by 1 order by 1; + g | count | sum +---+-------+----------- + 0 | 14285 | 714264285 + 1 | 14286 | 714278571 + 2 | 14286 | 714292857 + 3 | 14286 | 714307143 + 4 | 14286 | 714321429 + 5 | 14286 | 714335715 + 6 | 14285 | 714250000 +(7 rows) + +-- +-- Partitioned tables +-- +-- Each partition is split into its own chunks. +-- +create table part (k int, a int, c text) partition by range (a) distributed by (k); +create table part_1 partition of part for values from (1) to (30001) + with (parallel_workers = 4); +NOTICE: table has parent, setting distribution columns to match parent table +create table part_2 partition of part for values from (30001) to (60001) + with (parallel_workers = 4); +NOTICE: table has parent, setting distribution columns to match parent table +create table part_3 partition of part for values from (60001) to (90001) + with (parallel_workers = 4); +NOTICE: table has parent, setting distribution columns to match parent table +insert into part select 0, i, md5(i::text) from generate_series(1, 90000) i; +analyze part; +explain (costs off) :check part; + QUERY PLAN +----------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Append + -> Parallel Seq Scan on part_1 + -> Parallel Seq Scan on part_2 + -> Parallel Seq Scan on part_3 + Optimizer: Postgres query optimizer +(8 rows) + +:check part; + rows | sum_a | misaligned | sum_rownum +-------+------------+------------+------------ + 90000 | 4050045000 | 0 | 1350045000 +(1 row) + +-- +-- Rescans +-- +-- A parallel scan hands out each chunk once, so it cannot be scanned again. +-- Neither planner rescans one: it becomes the outer side of a nested loop, is +-- read once into a Materialize, or the plan is not parallel. These queries +-- have shapes where a rescan could happen. +-- +create table outer_rows (a int) + with (parallel_workers = 4) distributed by (a); +insert into outer_rows select i * 1000 from generate_series(1, 20) i; +create table inner_rows (k int, a int, c text) + with (parallel_workers = 4) distributed by (a); +insert into inner_rows select i % 7, i, md5(i::text) from generate_series(1, 200000) i; +analyze outer_rows; +analyze inner_rows; +set enable_hashjoin = off; +set enable_mergejoin = off; +set enable_material = off; +explain (costs off) +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; + QUERY PLAN +---------------------------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Nested Loop + Join Filter: (i.a < o.a) + -> Parallel Seq Scan on inner_rows i + Filter: (k = 3) + -> Materialize + -> Broadcast Motion 3:12 (slice2; segments: 3) + -> Seq Scan on outer_rows o + Optimizer: Postgres query optimizer +(11 rows) + +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; + count | sum +-------+----------- + 30000 | 204979006 +(1 row) + +explain (costs off) +select sum(n) from + (select (select count(*) from inner_rows i where i.a < o.a and i.k = 3) as n + from outer_rows o) s; + QUERY PLAN +----------------------------------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 3:1 (slice1; segments: 3) + -> Partial Aggregate + -> Seq Scan on outer_rows o + SubPlan 1 + -> Aggregate + -> Result + Filter: (i.a < o.a) + -> Materialize + -> Broadcast Motion 3:3 (slice2; segments: 3) + -> Seq Scan on inner_rows i + Filter: (k = 3) + Optimizer: Postgres query optimizer +(13 rows) + +select sum(n) from + (select (select count(*) from inner_rows i where i.a < o.a and i.k = 3) as n + from outer_rows o) s; + sum +------- + 30000 +(1 row) + +explain (costs off) +select count(*) from outer_rows o + where exists (select 1 from inner_rows i where i.a = o.a + 1); + QUERY PLAN +---------------------------------------------------------------------------------------- + Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> HashAggregate + Group Key: (RowIdExpr) + -> Redistribute Motion 12:12 (slice2; segments: 12) + Hash Key: (RowIdExpr) + Hash Module: 3 + -> Merge Join + Merge Cond: (i.a = ((o.a + 1))) + -> Sort + Sort Key: i.a + -> Parallel Seq Scan on inner_rows i + -> Sort + Sort Key: ((o.a + 1)) + -> Result + -> Broadcast Motion 3:12 (slice3; segments: 3) + -> Seq Scan on outer_rows o + Optimizer: Postgres query optimizer +(18 rows) + +select count(*) from outer_rows o + where exists (select 1 from inner_rows i where i.a = o.a + 1); + count +------- + 20 +(1 row) + +explain (costs off) +select count(*) from outer_rows o + where o.a in (select i.a from inner_rows i where i.k = o.a % 7); + QUERY PLAN +---------------------------------------------------------------------------------------- + Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> HashAggregate + Group Key: (RowIdExpr) + -> Redistribute Motion 12:12 (slice2; segments: 12) + Hash Key: (RowIdExpr) + Hash Module: 3 + -> Merge Join + Merge Cond: ((i.a = o.a) AND (i.k = ((o.a % 7)))) + -> Sort + Sort Key: i.a, i.k + -> Parallel Seq Scan on inner_rows i + -> Sort + Sort Key: o.a, ((o.a % 7)) + -> Result + -> Broadcast Motion 3:12 (slice3; segments: 3) + -> Seq Scan on outer_rows o + Optimizer: Postgres query optimizer +(18 rows) + +select count(*) from outer_rows o + where o.a in (select i.a from inner_rows i where i.k = o.a % 7); + count +------- + 20 +(1 row) + +explain (costs off) +select count(*), sum(l.n) from outer_rows o, + lateral (select count(*) as n from inner_rows i where i.a < o.a) l; + QUERY PLAN +------------------------------------------------------------------------------ + Aggregate + -> Nested Loop + -> Gather Motion 3:1 (slice1; segments: 3) + -> Seq Scan on outer_rows o + -> Materialize + -> Aggregate + -> Result + Filter: (i.a < o.a) + -> Materialize + -> Gather Motion 3:1 (slice2; segments: 3) + -> Seq Scan on inner_rows i + Optimizer: Postgres query optimizer +(12 rows) + +select count(*), sum(l.n) from outer_rows o, + lateral (select count(*) as n from inner_rows i where i.a < o.a) l; + count | sum +-------+-------- + 20 | 209980 +(1 row) + +reset enable_material; +reset enable_mergejoin; +reset enable_hashjoin; +-- The same plan executed several times. +prepare inner_rows_count(int) as select count(*), sum(a) from inner_rows where k = $1; +explain (costs off) execute inner_rows_count(3); + QUERY PLAN +--------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on inner_rows + Filter: (k = 3) + Optimizer: Postgres query optimizer +(6 rows) + +execute inner_rows_count(3); + count | sum +-------+------------ + 28572 | 2857242858 +(1 row) + +execute inner_rows_count(4); + count | sum +-------+------------ + 28571 | 2857071429 +(1 row) + +execute inner_rows_count(3); + count | sum +-------+------------ + 28572 | 2857242858 +(1 row) + +deallocate inner_rows_count; +-- A cursor. +begin; +declare c cursor for select count(*), sum(a) from inner_rows where k = 3; +fetch all from c; + count | sum +-------+------------ + 28572 | 2857242858 +(1 row) + +close c; +commit; +-- +-- Planner +-- +-- With the Postgres planner, a table without the parallel_workers option +-- gets workers according to its size, like a heap table, because its +-- segment files are split into chunks. +-- +create table plan_small (a int, c text) distributed by (a); +insert into plan_small select i, md5(i::text) from generate_series(1, 1000) i; +create table plan_mid (a int, c text) distributed by (a); +insert into plan_mid select i, md5(i::text) from generate_series(1, 100000) i; +create table plan_big (a int, c text) distributed by (a); +insert into plan_big select i, md5(i::text) from generate_series(1, 1000000) i; +create table plan_option (a int, c text) + with (parallel_workers = 3) distributed by (a); +insert into plan_option select i, md5(i::text) from generate_series(1, 1000) i; +create table plan_three (a int, c text) distributed by (a); +set gp_appendonly_insert_files = 3; +set gp_appendonly_insert_files_tuples_range = 10000; +insert into plan_three select i, md5(i::text) from generate_series(1, 1000000) i; +reset gp_appendonly_insert_files; +reset gp_appendonly_insert_files_tuples_range; +analyze plan_small; +analyze plan_mid; +analyze plan_big; +analyze plan_option; +analyze plan_three; +select c.relname, a.segfilecount +from pg_class c join pg_appendonly a on a.relid = c.oid +where c.relnamespace = 'ao_parallel_scan_@amname@'::regnamespace + and c.relname like 'plan\_%' order by 1; + relname | segfilecount +-------------+-------------- + plan_big | 1 + plan_mid | 1 + plan_option | 1 + plan_small | 1 + plan_three | 3 +(5 rows) + +set min_parallel_table_scan_size = '8MB'; +-- One segment file, and still a parallel plan. +explain (costs off) select count(*) from plan_big; + QUERY PLAN +------------------------------------------------- + Finalize Aggregate + -> Gather Motion 6:1 (slice1; segments: 6) + -> Partial Aggregate + -> Parallel Seq Scan on plan_big + Optimizer: Postgres query optimizer +(5 rows) + +-- A table smaller than min_parallel_table_scan_size is scanned serially. +explain (costs off) select count(*) from plan_small; + QUERY PLAN +------------------------------------------------ + Finalize Aggregate + -> Gather Motion 3:1 (slice1; segments: 3) + -> Partial Aggregate + -> Seq Scan on plan_small + Optimizer: Postgres query optimizer +(5 rows) + +-- The parallel_workers option overrides the size: 3 workers per segment. +explain (costs off) select count(*) from plan_option; + QUERY PLAN +---------------------------------------------------- + Finalize Aggregate + -> Gather Motion 9:1 (slice1; segments: 9) + -> Partial Aggregate + -> Parallel Seq Scan on plan_option + Optimizer: Postgres query optimizer +(5 rows) + +-- A larger table gets more workers, up to max_parallel_workers_per_gather. +set min_parallel_table_scan_size = '128kB'; +set max_parallel_workers_per_gather = 8; +explain (costs off) select count(*) from plan_mid; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on plan_mid + Optimizer: Postgres query optimizer +(5 rows) + +explain (costs off) select count(*) from plan_big; + QUERY PLAN +-------------------------------------------------- + Finalize Aggregate + -> Gather Motion 18:1 (slice1; segments: 18) + -> Partial Aggregate + -> Parallel Seq Scan on plan_big + Optimizer: Postgres query optimizer +(5 rows) + +set max_parallel_workers_per_gather = 2; +explain (costs off) select count(*) from plan_big; + QUERY PLAN +------------------------------------------------- + Finalize Aggregate + -> Gather Motion 6:1 (slice1; segments: 6) + -> Partial Aggregate + -> Parallel Seq Scan on plan_big + Optimizer: Postgres query optimizer +(5 rows) + +set max_parallel_workers_per_gather = 8; +-- The number of segment files does not matter: three segment files get the +-- same workers as the one of plan_big. +explain (costs off) select count(*) from plan_three; + QUERY PLAN +--------------------------------------------------- + Finalize Aggregate + -> Gather Motion 18:1 (slice1; segments: 18) + -> Partial Aggregate + -> Parallel Seq Scan on plan_three + Optimizer: Postgres query optimizer +(5 rows) + +-- No parallel index scans. +create index plan_big_a on plan_big (a); +set enable_seqscan = off; +set min_parallel_index_scan_size = 0; +explain (costs off) select count(*) from plan_big where a < 100; + QUERY PLAN +--------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 3:1 (slice1; segments: 3) + -> Partial Aggregate + -> Bitmap Heap Scan on plan_big + Recheck Cond: (a < 100) + -> Bitmap Index Scan on plan_big_a + Index Cond: (a < 100) + Optimizer: Postgres query optimizer +(8 rows) + +reset min_parallel_index_scan_size; +reset enable_seqscan; +reset min_parallel_table_scan_size; +set max_parallel_workers_per_gather = 4; +-- +-- ORCA +-- +-- ORCA plans a parallel scan only where it needs no rewinding, and puts a +-- Materialize above it on the inner side of a nested loop. +-- +set optimizer = on; +analyze empty_segfiles; +explain (costs off) :check single_segfile; + QUERY PLAN +-------------------------------------------------- + Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Parallel Seq Scan on single_segfile + Optimizer: GPORCA +(4 rows) + +:check single_segfile; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +explain (costs off) :check empty_segfiles; + QUERY PLAN +------------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on empty_segfiles + Optimizer: GPORCA +(5 rows) + +:check empty_segfiles; + rows | sum_a | misaligned | sum_rownum +-------+------------+------------+------------ + 30000 | 2250015000 | 0 | 450015000 +(1 row) + +explain (costs off) :check_cols compress_zstd; + QUERY PLAN +------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Seq Scan on compress_zstd + Optimizer: GPORCA +(5 rows) + +:check_cols compress_zstd; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 5000050000 +(1 row) + +explain (costs off) :check vis_gaps; + QUERY PLAN +-------------------------------------------------- + Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Parallel Seq Scan on vis_gaps + Optimizer: GPORCA +(4 rows) + +:check vis_gaps; + rows | sum_a | misaligned | sum_rownum +--------+------------+------------+------------ + 100000 | 5000050000 | 0 | 8961550000 +(1 row) + +-- LIMIT: the processes stop before all chunks are handed out. The Postgres +-- planner does not choose a parallel scan for these. +explain (costs off) select count(*) from (select a from filt limit 10) s; + QUERY PLAN +-------------------------------------------------------- + Aggregate + -> Limit + -> Gather Motion 12:1 (slice1; segments: 12) + -> Parallel Seq Scan on filt + Optimizer: GPORCA +(5 rows) + +select count(*) from (select a from filt limit 10) s; + count +------- + 10 +(1 row) + +explain (costs off) select count(*) from (select a from filt limit 20000) s; + QUERY PLAN +-------------------------------------------------------- + Aggregate + -> Limit + -> Gather Motion 12:1 (slice1; segments: 12) + -> Parallel Seq Scan on filt + Optimizer: GPORCA +(5 rows) + +select count(*) from (select a from filt limit 20000) s; + count +------- + 20000 +(1 row) + +set optimizer_enable_parallel_append = on; +explain (costs off) :check part; + QUERY PLAN +----------------------------------------------------- + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Parallel Append + -> Parallel Seq Scan on part_1 + -> Parallel Seq Scan on part_2 + -> Parallel Seq Scan on part_3 + Optimizer: GPORCA +(8 rows) + +:check part; + rows | sum_a | misaligned | sum_rownum +-------+------------+------------+------------ + 90000 | 4050045000 | 0 | 1350045000 +(1 row) + +reset optimizer_enable_parallel_append; +set enable_hashjoin = off; +set enable_mergejoin = off; +explain (costs off) +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; + QUERY PLAN +------------------------------------------------------------------------ + Finalize Aggregate + -> Gather Motion 12:1 (slice1; segments: 12) + -> Partial Aggregate + -> Nested Loop + Join Filter: (i.a < o.a) + -> Broadcast Motion 12:12 (slice2; segments: 12) + -> Parallel Seq Scan on outer_rows o + -> Materialize + -> Parallel Seq Scan on inner_rows i + Filter: (k = 3) + Optimizer: GPORCA +(11 rows) + +select count(*), sum(i.a) from outer_rows o join inner_rows i on i.a < o.a and i.k = 3; + count | sum +-------+----------- + 30000 | 204979006 +(1 row) + +reset enable_mergejoin; +reset enable_hashjoin; +reset optimizer; +reset max_parallel_workers_per_gather; +reset enable_parallel; +reset default_table_access_method; +drop schema ao_parallel_scan_@amname@ cascade; +NOTICE: drop cascades to 30 other objects +DETAIL: drop cascades to function segfile_rows(regclass) +drop cascades to function chunk_rows(regclass) +drop cascades to table single_segfile +drop cascades to table multi_segfile +drop cascades to table empty_segfiles +drop cascades to table boundary_eq +drop cascades to table boundary_minus +drop cascades to table boundary_plus +drop cascades to table tiny +drop cascades to table empty +drop cascades to function t_for(integer) +drop cascades to table wide_cols +drop cascades to table block_8k +drop cascades to table block_2m +drop cascades to table compress_zstd +drop cascades to table compress_zlib +drop cascades to table added_col +drop cascades to table vis_scattered +drop cascades to table vis_chunks +drop cascades to table vis_updated +drop cascades to table vis_gaps +drop cascades to table filt +drop cascades to table part +drop cascades to table outer_rows +drop cascades to table inner_rows +drop cascades to table plan_small +drop cascades to table plan_mid +drop cascades to table plan_big +drop cascades to table plan_option +drop cascades to table plan_three diff --git a/src/test/regress/sql/cbdb_parallel.sql b/src/test/regress/sql/cbdb_parallel.sql index 0df40bb32c0..7a65179950c 100644 --- a/src/test/regress/sql/cbdb_parallel.sql +++ b/src/test/regress/sql/cbdb_parallel.sql @@ -213,6 +213,40 @@ drop table ao1; drop table ao2; drop table aocs1; +-- +-- Parallel scan of AOCS columns with RLE compression. The other cases of +-- parallel AO row and AOCS scans are in input/uao_dml/ao_parallel_scan.source; +-- rle_type is AOCS-only. A parallel process positions every column at the +-- first row of its chunk, which is inside a block; rle_type packs regular +-- values densely, so a block holds many rows. t is mostly 32 bytes, but +-- every 1000th value is larger than a block. +-- +begin; +set local enable_parallel = on; +set local max_parallel_workers_per_gather = 4; +create table aocs_rle (k int, a int, + b bigint encoding (compresstype = rle_type, compresslevel = 4), + n numeric encoding (compresstype = rle_type, compresslevel = 1), + t text encoding (compresstype = rle_type, compresslevel = 2)) + using ao_column with (parallel_workers = 4) distributed by (k); +insert into aocs_rle + select 0, i, i::bigint * 7, i / 4.0, + case when i % 1000 = 0 then repeat(md5(i::text), 2000) else md5(i::text) end + from generate_series(1, 100000) i; +-- misaligned counts rows whose columns were read from different rows +explain(costs off) +select count(*), sum(a), + sum(case when b <> a::bigint * 7 or n <> a / 4.0 or + t <> case when a % 1000 = 0 then repeat(md5(a::text), 2000) else md5(a::text) end + then 1 else 0 end) as misaligned +from aocs_rle; +select count(*), sum(a), + sum(case when b <> a::bigint * 7 or n <> a / 4.0 or + t <> case when a % 1000 = 0 then repeat(md5(a::text), 2000) else md5(a::text) end + then 1 else 0 end) as misaligned +from aocs_rle; +abort; + -- test Parallel Bitmap Heap Scan begin; create table t1(c1 int, c2 int) with(parallel_workers=2) distributed by (c1); @@ -668,7 +702,7 @@ abort; -- -- Test Parallel Hash Left Anti Semi (Not-In) Join(parallel-oblivious). -- -create table t1(c1 int, c2 int) using ao_row distributed by (c1); +create table t1(c1 int, c2 int) using ao_row with(parallel_workers=2) distributed by (c1); create table t2(c1 int, c2 int) using ao_row distributed by (c1); create table t3_null(c1 int, c2 int) using ao_row distributed by (c1); begin; diff --git a/src/test/regress/sql/zlib.sql b/src/test/regress/sql/zlib.sql index 22dbf1d7609..98232ec2035 100644 --- a/src/test/regress/sql/zlib.sql +++ b/src/test/regress/sql/zlib.sql @@ -14,7 +14,9 @@ SET gp_workfile_compression = on; DROP TABLE IF EXISTS test_zlib_hashjoin; -CREATE TABLE test_zlib_hashjoin (i1 int, i2 int, i3 int, i4 int, i5 int, i6 int, i7 int, i8 int) WITH (APPENDONLY=true) DISTRIBUTED BY (i1) ; +-- No parallel scan: a Parallel Hash Join spills to shared files, which are +-- not created by BufFileCreateTempInSet() and do not hit the fault below. +CREATE TABLE test_zlib_hashjoin (i1 int, i2 int, i3 int, i4 int, i5 int, i6 int, i7 int, i8 int) WITH (APPENDONLY=true, parallel_workers=0) DISTRIBUTED BY (i1) ; INSERT INTO test_zlib_hashjoin SELECT i,i,i,i,i,i,i,i FROM (select generate_series(1, nsegments * 333333) as i from (select count(*) as nsegments from gp_segment_configuration where role='p' and content >= 0) foo) bar;