public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v10 09/17] Make table_scan_bitmap_next_block() async friendly 8+ messages / 4 participants [nested] [flat]
* [PATCH v10 09/17] Make table_scan_bitmap_next_block() async friendly @ 2024-03-14 16:39 Melanie Plageman <[email protected]> 0 siblings, 0 replies; 8+ messages in thread From: Melanie Plageman @ 2024-03-14 16:39 UTC (permalink / raw) table_scan_bitmap_next_block() previously returned false if we did not wish to call table_scan_bitmap_next_tuple() on the tuples on the page. This could happen when there were no visible tuples on the page or, due to concurrent activity on the table, the block returned by the iterator is past the end of the table recorded when the scan started. This forced the caller to be responsible for determining if additional blocks should be fetched and then for invoking table_scan_bitmap_next_block() for these blocks. It makes more sense for table_scan_bitmap_next_block() to be responsible for finding a block that is not past the end of the table (as of the time that the scan began) and for table_scan_bitmap_next_tuple() to return false if there are no visible tuples on the page. This also allows us to move responsibility for the iterator to table AM specific code. This means handling invalid blocks is entirely up to the table AM. These changes will enable bitmapheapscan to use the future streaming read API [1]. Table AMs will implement a streaming read API callback returning the next block to fetch. In heap AM's case, the callback will use the iterator to identify the next block to fetch. Since choosing the next block will no longer the responsibility of BitmapHeapNext(), the streaming read control flow requires these changes to table_scan_bitmap_next_block(). [1] https://www.postgresql.org/message-id/flat/CA%2BhUKGJkOiOCa%2Bmag4BF%2BzHo7qo%3Do9CFheB8%3Dg6uT5TUm2... --- src/backend/access/heap/heapam_handler.c | 59 +++++-- src/backend/executor/nodeBitmapHeapscan.c | 198 ++++++++++------------ src/include/access/relscan.h | 7 + src/include/access/tableam.h | 68 +++++--- src/include/nodes/execnodes.h | 12 +- 5 files changed, 195 insertions(+), 149 deletions(-) diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index cf4387f443..2ad785e511 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -2114,18 +2114,51 @@ heapam_estimate_rel_size(Relation rel, int32 *attr_widths, static bool heapam_scan_bitmap_next_block(TableScanDesc scan, - TBMIterateResult *tbmres, - bool *lossy) + bool *recheck, bool *lossy, BlockNumber *blockno) { HeapScanDesc hscan = (HeapScanDesc) scan; - BlockNumber block = tbmres->blockno; + BlockNumber block; Buffer buffer; Snapshot snapshot; int ntup; + TBMIterateResult *tbmres; hscan->rs_cindex = 0; hscan->rs_ntuples = 0; + *blockno = InvalidBlockNumber; + *recheck = true; + + do + { + CHECK_FOR_INTERRUPTS(); + + if (scan->shared_tbmiterator) + tbmres = tbm_shared_iterate(scan->shared_tbmiterator); + else + tbmres = tbm_iterate(scan->tbmiterator); + + if (tbmres == NULL) + { + /* no more entries in the bitmap */ + Assert(hscan->rs_empty_tuples_pending == 0); + return false; + } + + /* + * Ignore any claimed entries past what we think is the end of the + * relation. It may have been extended after the start of our scan (we + * only hold an AccessShareLock, and it could be inserts from this + * backend). We don't take this optimization in SERIALIZABLE + * isolation though, as we need to examine all invisible tuples + * reachable by the index. + */ + } while (!IsolationIsSerializable() && tbmres->blockno >= hscan->rs_nblocks); + + /* Got a valid block */ + *blockno = tbmres->blockno; + *recheck = tbmres->recheck; + /* * We can skip fetching the heap page if we don't need any fields from the * heap, the bitmap entries don't need rechecking, and all tuples on the @@ -2144,16 +2177,7 @@ heapam_scan_bitmap_next_block(TableScanDesc scan, return true; } - /* - * Ignore any claimed entries past what we think is the end of the - * relation. It may have been extended after the start of our scan (we - * only hold an AccessShareLock, and it could be inserts from this - * backend). We don't take this optimization in SERIALIZABLE isolation - * though, as we need to examine all invisible tuples reachable by the - * index. - */ - if (!IsolationIsSerializable() && block >= hscan->rs_nblocks) - return false; + block = tbmres->blockno; /* * Acquire pin on the target heap page, trading in any pin we held before. @@ -2245,7 +2269,14 @@ heapam_scan_bitmap_next_block(TableScanDesc scan, *lossy = tbmres->ntuples < 0; - return ntup > 0; + /* + * Return true to indicate that a valid block was found and the bitmap is + * not exhausted. If there are no visible tuples on this page, + * hscan->rs_ntuples will be 0 and heapam_scan_bitmap_next_tuple() will + * return false returning control to this function to advance to the next + * block in the bitmap. + */ + return true; } static bool diff --git a/src/backend/executor/nodeBitmapHeapscan.c b/src/backend/executor/nodeBitmapHeapscan.c index 282dcb9791..7e73583fe5 100644 --- a/src/backend/executor/nodeBitmapHeapscan.c +++ b/src/backend/executor/nodeBitmapHeapscan.c @@ -51,8 +51,7 @@ static TupleTableSlot *BitmapHeapNext(BitmapHeapScanState *node); static inline void BitmapDoneInitializingSharedState(ParallelBitmapHeapState *pstate); -static inline void BitmapAdjustPrefetchIterator(BitmapHeapScanState *node, - BlockNumber blockno); +static inline void BitmapAdjustPrefetchIterator(BitmapHeapScanState *node); static inline void BitmapAdjustPrefetchTarget(BitmapHeapScanState *node); static inline void BitmapPrefetch(BitmapHeapScanState *node, TableScanDesc scan); @@ -70,8 +69,8 @@ BitmapHeapNext(BitmapHeapScanState *node) { ExprContext *econtext; TableScanDesc scan; + bool lossy; TIDBitmap *tbm; - TBMIterateResult *tbmres; TupleTableSlot *slot; ParallelBitmapHeapState *pstate = node->pstate; dsa_area *dsa = node->ss.ps.state->es_query_dsa; @@ -83,7 +82,6 @@ BitmapHeapNext(BitmapHeapScanState *node) slot = node->ss.ss_ScanTupleSlot; scan = node->ss.ss_currentScanDesc; tbm = node->tbm; - tbmres = node->tbmres; /* * If we haven't yet performed the underlying index scan, do it, and begin @@ -111,7 +109,6 @@ BitmapHeapNext(BitmapHeapScanState *node) node->tbm = tbm; tbmiterator = tbm_begin_iterate(tbm); - node->tbmres = tbmres = NULL; #ifdef USE_PREFETCH if (node->prefetch_maximum > 0) @@ -164,7 +161,6 @@ BitmapHeapNext(BitmapHeapScanState *node) /* Allocate a private iterator and attach the shared state to it */ shared_tbmiterator = tbm_attach_shared_iterate(dsa, pstate->tbmiterator); - node->tbmres = tbmres = NULL; #ifdef USE_PREFETCH if (node->prefetch_maximum > 0) @@ -202,55 +198,19 @@ BitmapHeapNext(BitmapHeapScanState *node) extra_flags); } - node->tbmiterator = tbmiterator; - node->shared_tbmiterator = shared_tbmiterator; + scan->tbmiterator = tbmiterator; + scan->shared_tbmiterator = shared_tbmiterator; + node->initialized = true; + + goto new_page; } for (;;) { - bool valid, lossy; - - CHECK_FOR_INTERRUPTS(); - - /* - * Get next page of results if needed - */ - if (tbmres == NULL) - { - if (!pstate) - node->tbmres = tbmres = tbm_iterate(node->tbmiterator); - else - node->tbmres = tbmres = tbm_shared_iterate(node->shared_tbmiterator); - if (tbmres == NULL) - { - /* no more entries in the bitmap */ - break; - } - - BitmapAdjustPrefetchIterator(node, tbmres->blockno); - - valid = table_scan_bitmap_next_block(scan, tbmres, &lossy); - - if (lossy) - node->lossy_pages++; - else - node->exact_pages++; - - if (!valid) - { - /* AM doesn't think this block is valid, skip */ - continue; - } - - /* Adjust the prefetch target */ - BitmapAdjustPrefetchTarget(node); - } - else + while (table_scan_bitmap_next_tuple(scan, slot)) { - /* - * Continuing in previously obtained page. - */ + CHECK_FOR_INTERRUPTS(); #ifdef USE_PREFETCH @@ -272,45 +232,60 @@ BitmapHeapNext(BitmapHeapScanState *node) SpinLockRelease(&pstate->mutex); } #endif /* USE_PREFETCH */ - } - /* - * We issue prefetch requests *after* fetching the current page to try - * to avoid having prefetching interfere with the main I/O. Also, this - * should happen only when we have determined there is still something - * to do on the current page, else we may uselessly prefetch the same - * page we are just about to request for real. - */ - BitmapPrefetch(node, scan); + /* + * We issue prefetch requests *after* fetching the current page to + * try to avoid having prefetching interfere with the main I/O. + * Also, this should happen only when we have determined there is + * still something to do on the current page, else we may + * uselessly prefetch the same page we are just about to request + * for real. + */ + BitmapPrefetch(node, scan); - /* - * Attempt to fetch tuple from AM. - */ - if (!table_scan_bitmap_next_tuple(scan, slot)) - { - /* nothing more to look at on this page */ - node->tbmres = tbmres = NULL; - continue; + /* + * If we are using lossy info, we have to recheck the qual + * conditions at every tuple. + */ + if (node->recheck) + { + econtext->ecxt_scantuple = slot; + if (!ExecQualAndReset(node->bitmapqualorig, econtext)) + { + /* Fails recheck, so drop it and loop back for another */ + InstrCountFiltered2(node, 1); + ExecClearTuple(slot); + continue; + } + } + + /* OK to return this tuple */ + return slot; } +new_page: + + BitmapAdjustPrefetchIterator(node); + + if (!table_scan_bitmap_next_block(scan, &node->recheck, &lossy, &node->blockno)) + break; + + if (lossy) + node->lossy_pages++; + else + node->exact_pages++; + /* - * If we are using lossy info, we have to recheck the qual conditions - * at every tuple. + * If serial, we can error out if the the prefetch block doesn't stay + * ahead of the current block. */ - if (tbmres->recheck) - { - econtext->ecxt_scantuple = slot; - if (!ExecQualAndReset(node->bitmapqualorig, econtext)) - { - /* Fails recheck, so drop it and loop back for another */ - InstrCountFiltered2(node, 1); - ExecClearTuple(slot); - continue; - } - } + if (node->pstate == NULL && + node->prefetch_iterator && + node->pfblockno > node->blockno) + elog(ERROR, "prefetch and main iterators are out of sync"); - /* OK to return this tuple */ - return slot; + /* Adjust the prefetch target */ + BitmapAdjustPrefetchTarget(node); } /* @@ -336,13 +311,17 @@ BitmapDoneInitializingSharedState(ParallelBitmapHeapState *pstate) /* * BitmapAdjustPrefetchIterator - Adjust the prefetch iterator + * + * We keep track of how far the prefetch iterator is ahead of the main + * iterator in prefetch_pages. For each block the main iterator returns, we + * decrement prefetch_pages. */ static inline void -BitmapAdjustPrefetchIterator(BitmapHeapScanState *node, - BlockNumber blockno) +BitmapAdjustPrefetchIterator(BitmapHeapScanState *node) { #ifdef USE_PREFETCH ParallelBitmapHeapState *pstate = node->pstate; + TBMIterateResult *tbmpre; if (pstate == NULL) { @@ -356,14 +335,17 @@ BitmapAdjustPrefetchIterator(BitmapHeapScanState *node, else if (prefetch_iterator) { /* Do not let the prefetch iterator get behind the main one */ - TBMIterateResult *tbmpre = tbm_iterate(prefetch_iterator); - - if (tbmpre == NULL || tbmpre->blockno != blockno) - elog(ERROR, "prefetch and main iterators are out of sync"); + tbmpre = tbm_iterate(prefetch_iterator); + node->pfblockno = tbmpre ? tbmpre->blockno : InvalidBlockNumber; } return; } + /* + * Adjusting the prefetch iterator before invoking + * table_scan_bitmap_next_block() keeps prefetch distance higher across + * the parallel workers. + */ if (node->prefetch_maximum > 0) { TBMSharedIterator *prefetch_iterator = node->shared_prefetch_iterator; @@ -388,7 +370,10 @@ BitmapAdjustPrefetchIterator(BitmapHeapScanState *node, * case. */ if (prefetch_iterator) - tbm_shared_iterate(prefetch_iterator); + { + tbmpre = tbm_shared_iterate(prefetch_iterator); + node->pfblockno = tbmpre ? tbmpre->blockno : InvalidBlockNumber; + } } } #endif /* USE_PREFETCH */ @@ -466,6 +451,7 @@ BitmapPrefetch(BitmapHeapScanState *node, TableScanDesc scan) break; } node->prefetch_pages++; + node->pfblockno = tbmpre->blockno; /* * If we expect not to have to actually read this heap page, @@ -523,6 +509,8 @@ BitmapPrefetch(BitmapHeapScanState *node, TableScanDesc scan) break; } + node->pfblockno = tbmpre->blockno; + /* As above, skip prefetch if we expect not to need page */ skip_fetch = (!(scan->rs_flags & SO_NEED_TUPLE) && !tbmpre->recheck && @@ -584,12 +572,8 @@ ExecReScanBitmapHeapScan(BitmapHeapScanState *node) table_rescan(node->ss.ss_currentScanDesc, NULL); /* release bitmaps and buffers if any */ - if (node->tbmiterator) - tbm_end_iterate(node->tbmiterator); if (node->prefetch_iterator) tbm_end_iterate(node->prefetch_iterator); - if (node->shared_tbmiterator) - tbm_end_shared_iterate(node->shared_tbmiterator); if (node->shared_prefetch_iterator) tbm_end_shared_iterate(node->shared_prefetch_iterator); if (node->tbm) @@ -597,13 +581,13 @@ ExecReScanBitmapHeapScan(BitmapHeapScanState *node) if (node->pvmbuffer != InvalidBuffer) ReleaseBuffer(node->pvmbuffer); node->tbm = NULL; - node->tbmiterator = NULL; - node->tbmres = NULL; node->prefetch_iterator = NULL; node->initialized = false; - node->shared_tbmiterator = NULL; node->shared_prefetch_iterator = NULL; node->pvmbuffer = InvalidBuffer; + node->recheck = true; + node->blockno = InvalidBlockNumber; + node->pfblockno = InvalidBlockNumber; ExecScanReScan(&node->ss); @@ -634,28 +618,24 @@ ExecEndBitmapHeapScan(BitmapHeapScanState *node) */ ExecEndNode(outerPlanState(node)); + + /* + * close heap scan + */ + if (scanDesc) + table_endscan(scanDesc); + /* * release bitmaps and buffers if any */ - if (node->tbmiterator) - tbm_end_iterate(node->tbmiterator); if (node->prefetch_iterator) tbm_end_iterate(node->prefetch_iterator); if (node->tbm) tbm_free(node->tbm); - if (node->shared_tbmiterator) - tbm_end_shared_iterate(node->shared_tbmiterator); if (node->shared_prefetch_iterator) tbm_end_shared_iterate(node->shared_prefetch_iterator); if (node->pvmbuffer != InvalidBuffer) ReleaseBuffer(node->pvmbuffer); - - /* - * close heap scan - */ - if (scanDesc) - table_endscan(scanDesc); - } /* ---------------------------------------------------------------- @@ -688,8 +668,6 @@ ExecInitBitmapHeapScan(BitmapHeapScan *node, EState *estate, int eflags) scanstate->ss.ps.ExecProcNode = ExecBitmapHeapScan; scanstate->tbm = NULL; - scanstate->tbmiterator = NULL; - scanstate->tbmres = NULL; scanstate->pvmbuffer = InvalidBuffer; scanstate->exact_pages = 0; scanstate->lossy_pages = 0; @@ -697,9 +675,11 @@ ExecInitBitmapHeapScan(BitmapHeapScan *node, EState *estate, int eflags) scanstate->prefetch_pages = 0; scanstate->prefetch_target = 0; scanstate->initialized = false; - scanstate->shared_tbmiterator = NULL; scanstate->shared_prefetch_iterator = NULL; scanstate->pstate = NULL; + scanstate->recheck = true; + scanstate->blockno = InvalidBlockNumber; + scanstate->pfblockno = InvalidBlockNumber; /* * Miscellaneous initialization diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h index 521043304a..92b829cebc 100644 --- a/src/include/access/relscan.h +++ b/src/include/access/relscan.h @@ -24,6 +24,9 @@ struct ParallelTableScanDescData; +struct TBMIterator; +struct TBMSharedIterator; + /* * Generic descriptor for table scans. This is the base-class for table scans, * which needs to be embedded in the scans of individual AMs. @@ -40,6 +43,10 @@ typedef struct TableScanDescData ItemPointerData rs_mintid; ItemPointerData rs_maxtid; + /* Only used for Bitmap table scans */ + struct TBMIterator *tbmiterator; + struct TBMSharedIterator *shared_tbmiterator; + /* * Information about type and behaviour of the scan, a bitmask of members * of the ScanOptions enum (see tableam.h). diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h index bcf1497f67..a820cc8c99 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -21,6 +21,7 @@ #include "access/sdir.h" #include "access/xact.h" #include "executor/tuptable.h" +#include "nodes/tidbitmap.h" #include "utils/rel.h" #include "utils/snapshot.h" @@ -788,19 +789,14 @@ typedef struct TableAmRoutine */ /* - * Prepare to fetch / check / return tuples from `tbmres->blockno` as part - * of a bitmap table scan. `scan` was started via table_beginscan_bm(). - * Return false if there are no tuples to be found on the page, true - * otherwise. + * Prepare to fetch / check / return tuples from `blockno` as part of a + * bitmap table scan. `scan` was started via table_beginscan_bm(). Return + * false if the bitmap is exhausted and true otherwise. * * This will typically read and pin the target block, and do the necessary * work to allow scan_bitmap_next_tuple() to return tuples (e.g. it might * make sense to perform tuple visibility checks at this time). * - * If `tbmres->blockno` is -1, this is a lossy scan and all visible tuples - * on the page have to be returned, otherwise the tuples at offsets in - * `tbmres->offsets` need to be returned. - * * lossy indicates whether or not the block's representation in the bitmap * is lossy or exact. * @@ -819,8 +815,8 @@ typedef struct TableAmRoutine * scan_bitmap_next_tuple need to exist, or neither. */ bool (*scan_bitmap_next_block) (TableScanDesc scan, - struct TBMIterateResult *tbmres, - bool *lossy); + bool *recheck, bool *lossy, + BlockNumber *blockno); /* * Fetch the next tuple of a bitmap table scan into `slot` and return true @@ -957,9 +953,13 @@ static inline TableScanDesc table_beginscan_bm(Relation rel, Snapshot snapshot, int nkeys, struct ScanKeyData *key, uint32 extra_flags) { + TableScanDesc result; uint32 flags = SO_TYPE_BITMAPSCAN | SO_ALLOW_PAGEMODE | extra_flags; - return rel->rd_tableam->scan_begin(rel, snapshot, nkeys, key, NULL, flags); + result = rel->rd_tableam->scan_begin(rel, snapshot, nkeys, key, NULL, flags); + result->shared_tbmiterator = NULL; + result->tbmiterator = NULL; + return result; } /* @@ -1019,6 +1019,21 @@ table_beginscan_analyze(Relation rel) static inline void table_endscan(TableScanDesc scan) { + if (scan->rs_flags & SO_TYPE_BITMAPSCAN) + { + if (scan->shared_tbmiterator) + { + tbm_end_shared_iterate(scan->shared_tbmiterator); + scan->shared_tbmiterator = NULL; + } + + if (scan->tbmiterator) + { + tbm_end_iterate(scan->tbmiterator); + scan->tbmiterator = NULL; + } + } + scan->rs_rd->rd_tableam->scan_end(scan); } @@ -1029,6 +1044,21 @@ static inline void table_rescan(TableScanDesc scan, struct ScanKeyData *key) { + if (scan->rs_flags & SO_TYPE_BITMAPSCAN) + { + if (scan->shared_tbmiterator) + { + tbm_end_shared_iterate(scan->shared_tbmiterator); + scan->shared_tbmiterator = NULL; + } + + if (scan->tbmiterator) + { + tbm_end_iterate(scan->tbmiterator); + scan->tbmiterator = NULL; + } + } + scan->rs_rd->rd_tableam->scan_rescan(scan, key, false, false, false, false); } @@ -1981,19 +2011,17 @@ table_relation_estimate_size(Relation rel, int32 *attr_widths, */ /* - * Prepare to fetch / check / return tuples from `tbmres->blockno` as part of - * a bitmap table scan. `scan` needs to have been started via - * table_beginscan_bm(). Returns false if there are no tuples to be found on - * the page, true otherwise. lossy is set to true if bitmap is lossy for the - * selected block and false otherwise. + * Prepare to fetch / check / return tuples as part of a bitmap table scan. + * `scan` needs to have been started via table_beginscan_bm(). Returns false if + * there are no more blocks in the bitmap, true otherwise. lossy is set to true + * if bitmap is lossy for the selected block and false otherwise. * * Note, this is an optionally implemented function, therefore should only be * used after verifying the presence (at plan time or such). */ static inline bool table_scan_bitmap_next_block(TableScanDesc scan, - struct TBMIterateResult *tbmres, - bool *lossy) + bool *recheck, bool *lossy, BlockNumber *blockno) { /* * We don't expect direct calls to table_scan_bitmap_next_block with valid @@ -2003,8 +2031,8 @@ table_scan_bitmap_next_block(TableScanDesc scan, if (unlikely(TransactionIdIsValid(CheckXidAlive) && !bsysscan)) elog(ERROR, "unexpected table_scan_bitmap_next_block call during logical decoding"); - return scan->rs_rd->rd_tableam->scan_bitmap_next_block(scan, - tbmres, lossy); + return scan->rs_rd->rd_tableam->scan_bitmap_next_block(scan, recheck, + lossy, blockno); } /* diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 6871db9b21..8688bc5ab0 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1783,8 +1783,6 @@ typedef struct ParallelBitmapHeapState * * bitmapqualorig execution state for bitmapqualorig expressions * tbm bitmap obtained from child index scan(s) - * tbmiterator iterator for scanning current pages - * tbmres current-page data * pvmbuffer buffer for visibility-map lookups of prefetched pages * exact_pages total number of exact pages retrieved * lossy_pages total number of lossy pages retrieved @@ -1793,9 +1791,11 @@ typedef struct ParallelBitmapHeapState * prefetch_target current target prefetch distance * prefetch_maximum maximum value for prefetch_target * initialized is node is ready to iterate - * shared_tbmiterator shared iterator * shared_prefetch_iterator shared iterator for prefetching * pstate shared state for parallel bitmap scan + * recheck do current page's tuples need recheck + * blockno used to validate pf and current block in sync + * pfblockno used to validate pf stays ahead of current block * ---------------- */ typedef struct BitmapHeapScanState @@ -1803,8 +1803,6 @@ typedef struct BitmapHeapScanState ScanState ss; /* its first field is NodeTag */ ExprState *bitmapqualorig; TIDBitmap *tbm; - TBMIterator *tbmiterator; - TBMIterateResult *tbmres; Buffer pvmbuffer; long exact_pages; long lossy_pages; @@ -1813,9 +1811,11 @@ typedef struct BitmapHeapScanState int prefetch_target; int prefetch_maximum; bool initialized; - TBMSharedIterator *shared_tbmiterator; TBMSharedIterator *shared_prefetch_iterator; ParallelBitmapHeapState *pstate; + bool recheck; + BlockNumber blockno; + BlockNumber pfblockno; } BitmapHeapScanState; /* ---------------- -- 2.40.1 --3o7pc6dfau5a5hry Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v10-0010-Unify-parallel-and-serial-BitmapHeapScan-iterato.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Partial aggregates pushdown @ 2024-08-20 18:41 Bruce Momjian <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: Bruce Momjian @ 2024-08-20 18:41 UTC (permalink / raw) To: Jelte Fennema-Nio <[email protected]>; +Cc: [email protected] <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> On Tue, Aug 20, 2024 at 07:03:56PM +0200, Jelte Fennema-Nio wrote: > On Tue, 20 Aug 2024 at 18:50, Bruce Momjian <[email protected]> wrote: > > Okay, so we can do MAX easily, and AVG if the count can be represented > > as the same data type as the sum? Is that correct? Our only problem is > > that something like AVG(interval) can't use an array because arrays have > > to have the same data type for all array elements, and an interval can't > > represent a count? > > Close, but still not completely correct. AVG(bigint) can also not be > supported by patch 1, because the sum and the count for that both > stored using an int128. So we'd need an array of int128, and there's > currently no int128 SQL type. Okay. Have we considered having the FDW return a record: SELECT (oid, relname) FROM pg_class LIMIT 1; row --------------------- (2619,pg_statistic) SELECT pg_typeof((oid, relname)) FROM pg_class LIMIT 1; pg_typeof ----------- record SELECT pg_typeof(oid) FROM pg_class LIMIT 1; pg_typeof ----------- oid SELECT pg_typeof(relname) FROM pg_class LIMIT 1; pg_typeof ----------- name -- Bruce Momjian <[email protected]> https://momjian.us EDB https://enterprisedb.com Only you can decide what is important to you. ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Partial aggregates pushdown @ 2024-08-21 14:59 Tomas Vondra <[email protected]> parent: Bruce Momjian <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: Tomas Vondra @ 2024-08-21 14:59 UTC (permalink / raw) To: Bruce Momjian <[email protected]>; Jelte Fennema-Nio <[email protected]>; +Cc: [email protected] <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> On 8/20/24 20:41, Bruce Momjian wrote: > On Tue, Aug 20, 2024 at 07:03:56PM +0200, Jelte Fennema-Nio wrote: >> On Tue, 20 Aug 2024 at 18:50, Bruce Momjian <[email protected]> wrote: >>> Okay, so we can do MAX easily, and AVG if the count can be represented >>> as the same data type as the sum? Is that correct? Our only problem is >>> that something like AVG(interval) can't use an array because arrays have >>> to have the same data type for all array elements, and an interval can't >>> represent a count? >> >> Close, but still not completely correct. AVG(bigint) can also not be >> supported by patch 1, because the sum and the count for that both >> stored using an int128. So we'd need an array of int128, and there's >> currently no int128 SQL type. > > Okay. Have we considered having the FDW return a record: > > SELECT (oid, relname) FROM pg_class LIMIT 1; > row > --------------------- > (2619,pg_statistic) > > SELECT pg_typeof((oid, relname)) FROM pg_class LIMIT 1; > pg_typeof > ----------- > record > > SELECT pg_typeof(oid) FROM pg_class LIMIT 1; > pg_typeof > ----------- > oid > > SELECT pg_typeof(relname) FROM pg_class LIMIT 1; > pg_typeof > ----------- > name > How would this help with the AVG(bigint) case? We don't have int128 as SQL type, so what would be part of the record? Also, which part of the code would produce the record? If the internal state is "internal", that would probably need to be something aggregate specific, and that's kinda what this patch series is adding, no? Or am I missing some cases where the record would make it work? regards -- Tomas Vondra ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Partial aggregates pushdown @ 2024-08-22 17:22 Bruce Momjian <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: Bruce Momjian @ 2024-08-22 17:22 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Jelte Fennema-Nio <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> On Wed, Aug 21, 2024 at 04:59:12PM +0200, Tomas Vondra wrote: > On 8/20/24 20:41, Bruce Momjian wrote: > > SELECT (oid, relname) FROM pg_class LIMIT 1; > > row > > --------------------- > > (2619,pg_statistic) > > > > SELECT pg_typeof((oid, relname)) FROM pg_class LIMIT 1; > > pg_typeof > > ----------- > > record > > > > SELECT pg_typeof(oid) FROM pg_class LIMIT 1; > > pg_typeof > > ----------- > > oid > > > > SELECT pg_typeof(relname) FROM pg_class LIMIT 1; > > pg_typeof > > ----------- > > name > > > > How would this help with the AVG(bigint) case? We don't have int128 as > SQL type, so what would be part of the record? Also, which part of the > code would produce the record? If the internal state is "internal", that > would probably need to be something aggregate specific, and that's kinda > what this patch series is adding, no? > > Or am I missing some cases where the record would make it work? Right now, my goal in this thread is to try to concretely explain what is being proposed. Therefore, I think I need to go back and make four categories instead of two: 1. cases like MAX(int), where we return only one value, and the FDW return value is an existing data type, e.g., int 2. cases like AVG(int) where we return multiple FDW values of the same type and can use an array, e.g., bigint array 3. cases like AVG(bigint) where we return multiple FDW values of the same type (or can), but the one of the FDW return values is not an existing data type, e.g. int128 4. cases like AVG(interval) where we return multiple FDW values of different types, e.g. interval and an integral count For #1, all MAX cases have aggregate input parameters the same as the FDW return types (aggtranstype): SELECT proargtypes[0]::regtype, aggtranstype::regtype FROM pg_aggregate a JOIN pg_proc p ON a.aggfnoid = p.oid WHERE proname = 'max' AND proargtypes[0] != aggtranstype; proargtypes | aggtranstype -------------+-------------- For #2-4, we have for AVG: SELECT proargtypes[0]::regtype, aggtranstype::regtype FROM pg_aggregate a JOIN pg_proc p ON a.aggfnoid = p.oid WHERE proname = 'avg'; proargtypes | aggtranstype ------------------+-------------------- 3-> bigint | internal 2-> integer | bigint[] 2-> smallint | bigint[] 3-> numeric | internal 2-> real | double precision[] 2-> double precision | double precision[] 4-> interval | internal You can see which AVG items fall into which categories. It seems we have #1 and #2 handled cleanly in the patch. My question is related to #3 and #4. For #3, if we are going to be building infrastructure to handle passing int128 for AVG, wouldn't it be wiser to create an int128 type and an int128 array type, and then use method #2 to handle those, rather than creating custom code just to read/write int128 values for FDWs aggregate passing alone. For #4, can we use or improve the RECORD data type to handle #4 --- that seems preferable to creating custom FDWs aggregate passing code. I know the open question was whether we should create custom FDWs aggregate passing functions or custom data types for FDWs aggregate passing, but I am asking if we can improve existing facilities, like int128 or record passing, to reduce the need for some of these. -- Bruce Momjian <[email protected]> https://momjian.us EDB https://enterprisedb.com Only you can decide what is important to you. ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Partial aggregates pushdown @ 2024-08-22 18:31 Tomas Vondra <[email protected]> parent: Bruce Momjian <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: Tomas Vondra @ 2024-08-22 18:31 UTC (permalink / raw) To: Bruce Momjian <[email protected]>; +Cc: Jelte Fennema-Nio <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> On 8/22/24 19:22, Bruce Momjian wrote: > On Wed, Aug 21, 2024 at 04:59:12PM +0200, Tomas Vondra wrote: >> On 8/20/24 20:41, Bruce Momjian wrote: >>> SELECT (oid, relname) FROM pg_class LIMIT 1; >>> row >>> --------------------- >>> (2619,pg_statistic) >>> >>> SELECT pg_typeof((oid, relname)) FROM pg_class LIMIT 1; >>> pg_typeof >>> ----------- >>> record >>> >>> SELECT pg_typeof(oid) FROM pg_class LIMIT 1; >>> pg_typeof >>> ----------- >>> oid >>> >>> SELECT pg_typeof(relname) FROM pg_class LIMIT 1; >>> pg_typeof >>> ----------- >>> name >>> >> >> How would this help with the AVG(bigint) case? We don't have int128 as >> SQL type, so what would be part of the record? Also, which part of the >> code would produce the record? If the internal state is "internal", that >> would probably need to be something aggregate specific, and that's kinda >> what this patch series is adding, no? >> >> Or am I missing some cases where the record would make it work? > > Right now, my goal in this thread is to try to concretely explain what > is being proposed. Therefore, I think I need to go back and make four > categories instead of two: > > 1. cases like MAX(int), where we return only one value, and the FDW > return value is an existing data type, e.g., int > > 2. cases like AVG(int) where we return multiple FDW values of the same > type and can use an array, e.g., bigint array > > 3. cases like AVG(bigint) where we return multiple FDW values of the > same type (or can), but the one of the FDW return values is not an > existing data type, e.g. int128 > > 4. cases like AVG(interval) where we return multiple FDW values of > different types, e.g. interval and an integral count > > For #1, all MAX cases have aggregate input parameters the same as the > FDW return types (aggtranstype): > > SELECT proargtypes[0]::regtype, aggtranstype::regtype > FROM pg_aggregate a JOIN pg_proc p ON a.aggfnoid = p.oid > WHERE proname = 'max' AND proargtypes[0] != aggtranstype; > > proargtypes | aggtranstype > -------------+-------------- > > For #2-4, we have for AVG: > > SELECT proargtypes[0]::regtype, aggtranstype::regtype > FROM pg_aggregate a JOIN pg_proc p ON a.aggfnoid = p.oid > WHERE proname = 'avg'; > > proargtypes | aggtranstype > ------------------+-------------------- > 3-> bigint | internal > 2-> integer | bigint[] > 2-> smallint | bigint[] > 3-> numeric | internal > 2-> real | double precision[] > 2-> double precision | double precision[] > 4-> interval | internal > > You can see which AVG items fall into which categories. It seems we > have #1 and #2 handled cleanly in the patch. > Agreed. > My question is related to #3 and #4. For #3, if we are going to be > building infrastructure to handle passing int128 for AVG, wouldn't it be > wiser to create an int128 type and an int128 array type, and then use > method #2 to handle those, rather than creating custom code just to > read/write int128 values for FDWs aggregate passing alone. > Yep, adding int128 as a data type would extend this to aggregates that store state as int128 (or array of int128). > For #4, can we use or improve the RECORD data type to handle #4 --- that > seems preferable to creating custom FDWs aggregate passing code. > > I know the open question was whether we should create custom FDWs > aggregate passing functions or custom data types for FDWs aggregate > passing, but I am asking if we can improve existing facilities, like > int128 or record passing, to reduce the need for some of these. > But which code would produce the record? AFAIK it can't happen in some generic executor code, because that only sees "internal" for each aggregate. The exact structure of the aggstate is private within the code of each aggregate - the record would have to be built there, no? I imagine we'd add this for each aggregate as a new pair of functions to build/parse the record, but that's kinda the serial/deserial way we discussed earlier. Or are you suggesting we'd actually say: CREATE AGGREGATE xyz(...) ( STYPE = record, ... ) or something like that? I have no idea if that would work, maybe it would. In a way it'd not be that different from the "custom data type" except that it doesn't actually need a custom data type (assuming we know how to serialize such "generic" records - I'm not familiar with this code enough to answer that). The reason why I found the "custom data type" approach interesting is that it sets clear guarantees / expectations about the stability of the output between versions. For serial/deserial we make no guarantees, it can change even in a minor release. But here we need something stable even for major releases, to allow pushdown when querying older server. We have that strong expectation for data types, and everyone knows it. But I guess if we are OK with just sending the array aggregates, or or even just plain scalar types, we kinda shift this responsibility to aggregates. Because while the data type in/out functions will stay the same, we require the aggregate to not switch to some other state. regards -- Tomas Vondra ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Partial aggregates pushdown @ 2024-08-22 19:30 Bruce Momjian <[email protected]> parent: Tomas Vondra <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: Bruce Momjian @ 2024-08-22 19:30 UTC (permalink / raw) To: Tomas Vondra <[email protected]>; +Cc: Jelte Fennema-Nio <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> On Thu, Aug 22, 2024 at 08:31:11PM +0200, Tomas Vondra wrote: > > My question is related to #3 and #4. For #3, if we are going to be > > building infrastructure to handle passing int128 for AVG, wouldn't it be > > wiser to create an int128 type and an int128 array type, and then use > > method #2 to handle those, rather than creating custom code just to > > read/write int128 values for FDWs aggregate passing alone. > > > > Yep, adding int128 as a data type would extend this to aggregates that > store state as int128 (or array of int128). Great, I am not too far off then. > > For #4, can we use or improve the RECORD data type to handle #4 --- that > > seems preferable to creating custom FDWs aggregate passing code. > > > > I know the open question was whether we should create custom FDWs > > aggregate passing functions or custom data types for FDWs aggregate > > passing, but I am asking if we can improve existing facilities, like > > int128 or record passing, to reduce the need for some of these. > > > > But which code would produce the record? AFAIK it can't happen in some > generic executor code, because that only sees "internal" for each > aggregate. The exact structure of the aggstate is private within the > code of each aggregate - the record would have to be built there, no? > > I imagine we'd add this for each aggregate as a new pair of functions to > build/parse the record, but that's kinda the serial/deserial way we > discussed earlier. > > Or are you suggesting we'd actually say: > > CREATE AGGREGATE xyz(...) ( > STYPE = record, > ... > ) So my idea from the email I just sent is to create a pg_proc.proargtypes-like column (data type oidvector) for pg_aggregate which stores the oids of the values we want to return, so AVG(interval) would have an array of the oids for interval and int8, e.g.: SELECT oid FROM pg_type WHERE typname = 'interval'; oid ------ 1186 SELECT oid FROM pg_type WHERE typname = 'int8'; oid ----- 20 SELECT '1186 20'::oidvector; oidvector ----------- 1186 20 It seems all four methods could use this, again assuming we create int128/int16 and whatever other types we need. -- Bruce Momjian <[email protected]> https://momjian.us EDB https://enterprisedb.com Only you can decide what is important to you. ^ permalink raw reply [nested|flat] 8+ messages in thread
* RE: Partial aggregates pushdown @ 2025-03-28 02:00 [email protected] <[email protected]> parent: Bruce Momjian <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: [email protected] @ 2025-03-28 02:00 UTC (permalink / raw) To: Bruce Momjian <[email protected]>; Tomas Vondra <[email protected]>; +Cc: Jelte Fennema-Nio <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> Hi Bruce, Jelte, hackers. I apologize for my late response. > From: Jelte Fennema-Nio <[email protected]> > Sent: Thursday, August 8, 2024 8:49 PM > SUMMARY OF THREAD > > The design of patch 0001 is agreed upon by everyone on the thread (so far). This adds the PARTIAL_AGGREGATE label for > aggregates, which will cause the finalfunc not to run. It also starts using PARTIAL_AGGREGATE for pushdown of > aggregates in postgres_fdw. In 0001 PARTIAL_AGGREGATE is only supported for aggregates with a non-internal/pseudo > type as the stype. > > The design for patch 0002 is still under debate. This would expand on the functionality added by adding support for > PARTIAL_AGGREGATE for aggregates with an internal stype. This is done by returning a byte array containing the bytes > that the serialfunc of the aggregate returns. > > A competing proposal for 0002 is to instead change aggregates to not use an internal stype anymore, and create dedicated > types. The main downside here is that infunc and outfunc would need to be added for text serialization, in addition to the > binary serialization. An open question is: Can we change the requirements for CREATE TYPE, so that types can be created > without infunc and outfunc. I rebased the patch 0001 and add the documentation to it. Best regards, Yuki Fujii -- Yuki Fujii Information Technology R&D Center, Mitsubishi Electric Corporation > -----Original Message----- > From: Bruce Momjian <[email protected]> > Sent: Friday, August 23, 2024 4:31 AM > To: Tomas Vondra <[email protected]> > Cc: Jelte Fennema-Nio <[email protected]>; Fujii Yuki/藤井 雄規(MELCO/情報総研 ＤＭ最適Ｇ) > <[email protected]>; PostgreSQL-development <[email protected]>; Robert Haas > <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund > <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; > Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov > <[email protected]> > Subject: Re: Partial aggregates pushdown > > On Thu, Aug 22, 2024 at 08:31:11PM +0200, Tomas Vondra wrote: > > > My question is related to #3 and #4. For #3, if we are going to be > > > building infrastructure to handle passing int128 for AVG, wouldn't > > > it be wiser to create an int128 type and an int128 array type, and > > > then use method #2 to handle those, rather than creating custom code > > > just to read/write int128 values for FDWs aggregate passing alone. > > > > > > > Yep, adding int128 as a data type would extend this to aggregates that > > store state as int128 (or array of int128). > > Great, I am not too far off then. > > > > For #4, can we use or improve the RECORD data type to handle #4 --- > > > that seems preferable to creating custom FDWs aggregate passing code. > > > > > > I know the open question was whether we should create custom FDWs > > > aggregate passing functions or custom data types for FDWs aggregate > > > passing, but I am asking if we can improve existing facilities, like > > > int128 or record passing, to reduce the need for some of these. > > > > > > > But which code would produce the record? AFAIK it can't happen in some > > generic executor code, because that only sees "internal" for each > > aggregate. The exact structure of the aggstate is private within the > > code of each aggregate - the record would have to be built there, no? > > > > I imagine we'd add this for each aggregate as a new pair of functions > > to build/parse the record, but that's kinda the serial/deserial way we > > discussed earlier. > > > > Or are you suggesting we'd actually say: > > > > CREATE AGGREGATE xyz(...) ( > > STYPE = record, > > ... > > ) > > So my idea from the email I just sent is to create a pg_proc.proargtypes-like column (data type oidvector) for pg_aggregate > which stores the oids of the values we want to return, so AVG(interval) would have an array of the oids for interval and int8, > e.g.: > > SELECT oid FROM pg_type WHERE typname = 'interval'; > oid > ------ > 1186 > > SELECT oid FROM pg_type WHERE typname = 'int8'; > oid > ----- > 20 > > SELECT '1186 20'::oidvector; > oidvector > ----------- > 1186 20 > > It seems all four methods could use this, again assuming we create > int128/int16 and whatever other types we need. > > -- > Bruce Momjian <[email protected]> https://momjian.us > EDB https://enterprisedb.com > > Only you can decide what is important to you. Attachments: [application/octet-stream] 0001-Partial-Aggregate-Pushdown-Add-PARTIAL_AGGREGATE-key.patch (113.8K, ../../TYRPR01MB13941CEA16574771B1BFD130595A02@TYRPR01MB13941.jpnprd01.prod.outlook.com/2-0001-Partial-Aggregate-Pushdown-Add-PARTIAL_AGGREGATE-key.patch) download | inline diff: From ac871aa602a8330af87c7c9cef3e918b65e60030 Mon Sep 17 00:00:00 2001 From: Yuki Fujii <[email protected]> Date: Mon, 17 Mar 2025 09:20:58 -0400 Subject: [PATCH] Partial Aggregate Pushdown; Add PARTIAL_AGGREGATE keyword --- contrib/postgres_fdw/deparse.c | 75 +- .../postgres_fdw/expected/postgres_fdw.out | 638 ++++++++++++++++-- contrib/postgres_fdw/option.c | 4 +- contrib/postgres_fdw/postgres_fdw.c | 43 +- contrib/postgres_fdw/postgres_fdw.h | 12 +- contrib/postgres_fdw/sql/postgres_fdw.sql | 112 ++- doc/src/sgml/postgres-fdw.sgml | 103 ++- doc/src/sgml/syntax.sgml | 27 +- src/backend/executor/nodeAgg.c | 6 +- src/backend/nodes/makefuncs.c | 1 + src/backend/optimizer/plan/planner.c | 133 +++- src/backend/optimizer/plan/setrefs.c | 1 + src/backend/optimizer/prep/prepagg.c | 1 + src/backend/parser/gram.y | 18 +- src/backend/parser/parse_agg.c | 21 +- src/backend/parser/parse_clause.c | 1 + src/backend/parser/parse_expr.c | 3 +- src/backend/parser/parse_func.c | 24 +- src/backend/utils/adt/ruleutils.c | 3 +- src/include/nodes/parsenodes.h | 1 + src/include/nodes/pathnodes.h | 5 + src/include/nodes/primnodes.h | 3 + src/include/parser/kwlist.h | 1 + src/include/parser/parse_agg.h | 2 +- src/test/regress/expected/aggregates.out | 8 + .../regress/expected/partition_aggregate.out | 29 + src/test/regress/sql/aggregates.sql | 4 + src/test/regress/sql/partition_aggregate.sql | 5 + 28 files changed, 1149 insertions(+), 135 deletions(-) diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c index 6e7dc3d2df9..e30bf8c4329 100644 --- a/contrib/postgres_fdw/deparse.c +++ b/contrib/postgres_fdw/deparse.c @@ -204,7 +204,7 @@ static bool is_subquery_var(Var *node, RelOptInfo *foreignrel, int *relno, int *colno); static void get_relation_column_alias_ids(Var *node, RelOptInfo *foreignrel, int *relno, int *colno); - +static bool partial_agg_ok(Aggref *agg, PgFdwRelationInfo *fpinfo); /* * Examine each qual clause in input_conds, and classify them into two groups, @@ -909,8 +909,9 @@ foreign_expr_walker(Node *node, if (!IS_UPPER_REL(glob_cxt->foreignrel)) return false; - /* Only non-split aggregates are pushable. */ - if (agg->aggsplit != AGGSPLIT_SIMPLE) + if (agg->aggsplit != AGGSPLIT_SIMPLE && agg->aggsplit != AGGSPLIT_INITIAL_SERIAL) + return false; + if (agg->aggsplit == AGGSPLIT_INITIAL_SERIAL && !partial_agg_ok(agg, fpinfo)) return false; /* As usual, it must be shippable. */ @@ -3657,18 +3658,21 @@ deparseAggref(Aggref *node, deparse_expr_cxt *context) StringInfo buf = context->buf; bool use_variadic; - /* Only basic, non-split aggregation accepted. */ - Assert(node->aggsplit == AGGSPLIT_SIMPLE); + + Assert((node->aggsplit == AGGSPLIT_SIMPLE) || + (node->aggsplit == AGGSPLIT_INITIAL_SERIAL)); /* Check if need to print VARIADIC (cf. ruleutils.c) */ use_variadic = node->aggvariadic; /* Find aggregate name from aggfnoid which is a pg_proc entry */ appendFunctionName(node->aggfnoid, context); + appendStringInfoChar(buf, '('); - /* Add DISTINCT */ - appendStringInfoString(buf, (node->aggdistinct != NIL) ? "DISTINCT " : ""); + /* Add DISTINCT or PARTIAL_AGGREGATE */ + if (node->aggdistinct != NIL) + appendStringInfoString(buf, "DISTINCT "); if (AGGKIND_IS_ORDERED_SET(node->aggkind)) { @@ -3738,6 +3742,9 @@ deparseAggref(Aggref *node, deparse_expr_cxt *context) } appendStringInfoChar(buf, ')'); + + if (node->aggsplit == AGGSPLIT_INITIAL_SERIAL) + appendStringInfoString(buf, " PARTIAL_AGGREGATE "); } /* @@ -3863,9 +3870,10 @@ appendGroupByClause(List *tlist, deparse_expr_cxt *context) Query *query = context->root->parse; ListCell *lc; bool first = true; + PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) context->foreignrel->fdw_private; /* Nothing to be done, if there's no GROUP BY clause in the query. */ - if (!query->groupClause) + if (!fpinfo->group_clause) return; appendStringInfoString(buf, " GROUP BY "); @@ -3883,7 +3891,7 @@ appendGroupByClause(List *tlist, deparse_expr_cxt *context) * to empty, and in any case the redundancy situation on the remote might * be different than what we think here. */ - foreach(lc, query->groupClause) + foreach(lc, fpinfo->group_clause) { SortGroupClause *grp = (SortGroupClause *) lfirst(lc); @@ -4204,3 +4212,52 @@ get_relation_column_alias_ids(Var *node, RelOptInfo *foreignrel, /* Shouldn't get here */ elog(ERROR, "unexpected expression in subquery output"); } + +/* + * ---------- + * Check that partial aggregate "agg" is safe to push down. + * + * It is pushdown-safe when all of the following conditions are true: + * + * * agg is an AGGKIND_NORMAL aggregate which contains no DISTINCT or + * ORDER BY clauses + * * remote server can return partial aggregate results + * ---------- + */ +static bool +partial_agg_ok(Aggref *agg, PgFdwRelationInfo *fpinfo) +{ + HeapTuple aggtup; + Form_pg_aggregate aggform; + bool partial_agg_ok = true; + + Assert(agg->aggsplit == AGGSPLIT_INITIAL_SERIAL); + + /* We don't support complex partial aggregates */ + if (agg->aggdistinct || agg->aggkind != AGGKIND_NORMAL || agg->aggorder != NIL) + return false; + + aggtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(agg->aggfnoid)); + if (!HeapTupleIsValid(aggtup)) + elog(ERROR, "cache lookup failed for aggregate %u", agg->aggfnoid); + aggform = (Form_pg_aggregate) GETSTRUCT(aggtup); + + partial_agg_ok = fpinfo->partial_aggregate_support; + if(aggform->aggtranstype == INTERNALOID || + get_typtype(aggform->aggtranstype) == TYPTYPE_PSEUDO) + partial_agg_ok = false; + if(partial_agg_ok){ + if (fpinfo->remoteversion == 0) + { + PGconn *conn = GetConnection(fpinfo->user, false, NULL); + + fpinfo->remoteversion = PQserverVersion(conn); + } + + if (fpinfo->remoteversion != PG_VERSION_NUM) + partial_agg_ok = false; + } + + ReleaseSysCache(aggtup); + return partial_agg_ok; +} diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index bb4ed3059c4..f99d93c1805 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -10260,36 +10260,57 @@ RESET enable_partitionwise_join; -- =================================================================== -- test partitionwise aggregates -- =================================================================== -CREATE TABLE pagg_tab (a int, b int, c text) PARTITION BY RANGE(a); +ALTER SERVER loopback OPTIONS (ADD fdw_tuple_cost '0.1'); +CREATE TYPE mood AS enum ('sad', 'ok', 'happy'); +ALTER EXTENSION postgres_fdw ADD TYPE mood; +CREATE TABLE pagg_tab (a int, b int, c text, c_serial int4, + c_int4array _int4, c_interval interval, + c_money money, c_1c text, c_1b bytea, + c_bit bit(2), c_1or3int2 int2, + c_1or3int4 int4, c_1or3int8 int8, + c_bool bool, c_enum mood, c_pg_lsn pg_lsn, + c_tid tid, c_int4range int4range, + c_int4multirange int4multirange, + c_time time, c_timetz timetz, + c_timestamp timestamp, c_timestamptz timestamptz, + c_xid8 xid8) + PARTITION BY RANGE(a); CREATE TABLE pagg_tab_p1 (LIKE pagg_tab); CREATE TABLE pagg_tab_p2 (LIKE pagg_tab); CREATE TABLE pagg_tab_p3 (LIKE pagg_tab); -INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 10; -INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10; -INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20; +SET TimeZone = 'UTC'; +INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i, array[(i % 2), 0], ((i % 2) || ' seconds')::interval, (i % 2)::money, ((i - 1) % 10)::text, (((i - 1) % 10)::text)::bytea, case when (i % 2) = 0 then B'01' else B'11' end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then true else false end, (case when (i % 2) = 0 then 'sad' else 'happy' end)::mood, ('0/' || i)::pg_lsn, ('(0,' || i || ')')::tid, int4range(0, i), int4multirange(int4range(0, i), int4range(100, 100 + i)), '00:00:00'::time + (i || ' seconds')::interval, '00:00:00'::timetz + (i || ' seconds')::interval, '2000-01-01'::timestamp + (i || ' seconds')::interval, '2000-01-01'::timestamptz + (i || ' seconds')::interval, ((i % 10)::text)::xid8 FROM generate_series(1, 3000) i WHERE (i % 30) < 10; +INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i, array[(i % 2), 0], ((i % 2) || ' seconds')::interval, (i % 2)::money, ((i - 1) % 10)::text, (((i - 1) % 10)::text)::bytea, case when (i % 2) = 0 then B'01' else B'11' end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then true else false end, (case when (i % 2) = 0 then 'sad' else 'happy' end)::mood, ('0/' || i)::pg_lsn, ('(0,' || i || ')')::tid, int4range(0, i), int4multirange(int4range(0, i), int4range(100, 100 + i)), '00:00:00'::time + (i || ' seconds')::interval, '00:00:00'::timetz + (i || ' seconds')::interval, '2000-01-01'::timestamp + (i || ' seconds')::interval, '2000-01-01'::timestamptz + (i || ' seconds')::interval, ((i % 10)::text)::xid8 FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10; +INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i, array[(i % 2), 0], ((i % 2) || ' seconds')::interval, (i % 2)::money, ((i - 1) % 10)::text, (((i - 1) % 10)::text)::bytea, case when (i % 2) = 0 then B'01' else B'11' end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then true else false end, (case when (i % 2) = 0 then 'sad' else 'happy' end)::mood, ('0/' || i)::pg_lsn, ('(0,' || i || ')')::tid, int4range(0, i), int4multirange(int4range(0, i), int4range(100, 100 + i)), '00:00:00'::time + (i || ' seconds')::interval, '00:00:00'::timetz + (i || ' seconds')::interval, '2000-01-01'::timestamp + (i || ' seconds')::interval, '2000-01-01'::timestamptz + (i || ' seconds')::interval, ((i % 10)::text)::xid8 FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20; -- Create foreign partitions CREATE FOREIGN TABLE fpagg_tab_p1 PARTITION OF pagg_tab FOR VALUES FROM (0) TO (10) SERVER loopback OPTIONS (table_name 'pagg_tab_p1'); CREATE FOREIGN TABLE fpagg_tab_p2 PARTITION OF pagg_tab FOR VALUES FROM (10) TO (20) SERVER loopback OPTIONS (table_name 'pagg_tab_p2'); CREATE FOREIGN TABLE fpagg_tab_p3 PARTITION OF pagg_tab FOR VALUES FROM (20) TO (30) SERVER loopback OPTIONS (table_name 'pagg_tab_p3'); ANALYZE pagg_tab; +ANALYZE pagg_tab_p1; +ANALYZE pagg_tab_p2; +ANALYZE pagg_tab_p3; ANALYZE fpagg_tab_p1; ANALYZE fpagg_tab_p2; ANALYZE fpagg_tab_p3; +SET extra_float_digits = 0; -- When GROUP BY clause matches with PARTITION KEY. -- Plan with partitionwise aggregates is disabled SET enable_partitionwise_aggregate TO false; EXPLAIN (COSTS OFF) SELECT a, sum(b), min(b), count(*) FROM pagg_tab GROUP BY a HAVING avg(b) < 22 ORDER BY 1; - QUERY PLAN ------------------------------------------------------ - GroupAggregate - Group Key: pagg_tab.a - Filter: (avg(pagg_tab.b) < '22'::numeric) - -> Append - -> Foreign Scan on fpagg_tab_p1 pagg_tab_1 - -> Foreign Scan on fpagg_tab_p2 pagg_tab_2 - -> Foreign Scan on fpagg_tab_p3 pagg_tab_3 -(7 rows) + QUERY PLAN +----------------------------------------------------------- + Sort + Sort Key: pagg_tab.a + -> HashAggregate + Group Key: pagg_tab.a + Filter: (avg(pagg_tab.b) < '22'::numeric) + -> Append + -> Foreign Scan on fpagg_tab_p1 pagg_tab_1 + -> Foreign Scan on fpagg_tab_p2 pagg_tab_2 + -> Foreign Scan on fpagg_tab_p3 pagg_tab_3 +(9 rows) -- Plan with partitionwise aggregates is enabled SET enable_partitionwise_aggregate TO true; @@ -10323,32 +10344,34 @@ SELECT a, sum(b), min(b), count(*) FROM pagg_tab GROUP BY a HAVING avg(b) < 22 O -- Should have all the columns in the target list for the given relation EXPLAIN (VERBOSE, COSTS OFF) SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1; - QUERY PLAN --------------------------------------------------------------------------------------------- - Merge Append + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Sort + Output: t1.a, (count(((t1.*)::pagg_tab))) Sort Key: t1.a - -> GroupAggregate - Output: t1.a, count(((t1.*)::pagg_tab)) - Group Key: t1.a - Filter: (avg(t1.b) < '22'::numeric) - -> Foreign Scan on public.fpagg_tab_p1 t1 - Output: t1.a, t1.*, t1.b - Remote SQL: SELECT a, b, c FROM public.pagg_tab_p1 ORDER BY a ASC NULLS LAST - -> GroupAggregate - Output: t1_1.a, count(((t1_1.*)::pagg_tab)) - Group Key: t1_1.a - Filter: (avg(t1_1.b) < '22'::numeric) - -> Foreign Scan on public.fpagg_tab_p2 t1_1 - Output: t1_1.a, t1_1.*, t1_1.b - Remote SQL: SELECT a, b, c FROM public.pagg_tab_p2 ORDER BY a ASC NULLS LAST - -> GroupAggregate - Output: t1_2.a, count(((t1_2.*)::pagg_tab)) - Group Key: t1_2.a - Filter: (avg(t1_2.b) < '22'::numeric) - -> Foreign Scan on public.fpagg_tab_p3 t1_2 - Output: t1_2.a, t1_2.*, t1_2.b - Remote SQL: SELECT a, b, c FROM public.pagg_tab_p3 ORDER BY a ASC NULLS LAST -(23 rows) + -> Append + -> HashAggregate + Output: t1.a, count(((t1.*)::pagg_tab)) + Group Key: t1.a + Filter: (avg(t1.b) < '22'::numeric) + -> Foreign Scan on public.fpagg_tab_p1 t1 + Output: t1.a, t1.*, t1.b + Remote SQL: SELECT a, b, c, c_serial, c_int4array, c_interval, c_money, c_1c, c_1b, c_bit, c_1or3int2, c_1or3int4, c_1or3int8, c_bool, c_enum, c_pg_lsn, c_tid, c_int4range, c_int4multirange, c_time, c_timetz, c_timestamp, c_timestamptz, c_xid8 FROM public.pagg_tab_p1 + -> HashAggregate + Output: t1_1.a, count(((t1_1.*)::pagg_tab)) + Group Key: t1_1.a + Filter: (avg(t1_1.b) < '22'::numeric) + -> Foreign Scan on public.fpagg_tab_p2 t1_1 + Output: t1_1.a, t1_1.*, t1_1.b + Remote SQL: SELECT a, b, c, c_serial, c_int4array, c_interval, c_money, c_1c, c_1b, c_bit, c_1or3int2, c_1or3int4, c_1or3int8, c_bool, c_enum, c_pg_lsn, c_tid, c_int4range, c_int4multirange, c_time, c_timetz, c_timestamp, c_timestamptz, c_xid8 FROM public.pagg_tab_p2 + -> HashAggregate + Output: t1_2.a, count(((t1_2.*)::pagg_tab)) + Group Key: t1_2.a + Filter: (avg(t1_2.b) < '22'::numeric) + -> Foreign Scan on public.fpagg_tab_p3 t1_2 + Output: t1_2.a, t1_2.*, t1_2.b + Remote SQL: SELECT a, b, c, c_serial, c_int4array, c_interval, c_money, c_1c, c_1b, c_bit, c_1or3int2, c_1or3int4, c_1or3int8, c_bool, c_enum, c_pg_lsn, c_tid, c_int4range, c_int4multirange, c_time, c_timetz, c_timestamp, c_timestamptz, c_xid8 FROM public.pagg_tab_p3 +(25 rows) SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1; a | count @@ -10361,27 +10384,534 @@ SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1; 21 | 100 (6 rows) --- When GROUP BY clause does not match with PARTITION KEY. -EXPLAIN (COSTS OFF) +-- Partial aggregates are safe to push down when there is a HAVING clause +EXPLAIN (VERBOSE, COSTS OFF) SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1; - QUERY PLAN ------------------------------------------------------------ + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Sort + Output: pagg_tab.b, (avg(pagg_tab.a)), (max(pagg_tab.a)), (count(*)) + Sort Key: pagg_tab.b + -> Finalize HashAggregate + Output: pagg_tab.b, avg(pagg_tab.a), max(pagg_tab.a), count(*) + Group Key: pagg_tab.b + Filter: (sum(pagg_tab.a) < 700) + -> Append + -> Foreign Scan + Output: pagg_tab.b, (PARTIAL avg(pagg_tab.a)), (PARTIAL max(pagg_tab.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab.a)) + Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab) + Remote SQL: SELECT b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE , sum(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p1 GROUP BY 1 + -> Foreign Scan + Output: pagg_tab_1.b, (PARTIAL avg(pagg_tab_1.a)), (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab_1.a)) + Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1) + Remote SQL: SELECT b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE , sum(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p2 GROUP BY 1 + -> Foreign Scan + Output: pagg_tab_2.b, (PARTIAL avg(pagg_tab_2.a)), (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*)), (PARTIAL sum(pagg_tab_2.a)) + Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2) + Remote SQL: SELECT b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE , sum(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p3 GROUP BY 1 +(20 rows) + +SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1; + b | avg | max | count +----+---------------------+-----+------- + 0 | 10.0000000000000000 | 20 | 60 + 1 | 11.0000000000000000 | 21 | 60 + 10 | 10.0000000000000000 | 20 | 60 + 11 | 11.0000000000000000 | 21 | 60 + 20 | 10.0000000000000000 | 20 | 60 + 21 | 11.0000000000000000 | 21 | 60 + 30 | 10.0000000000000000 | 20 | 60 + 31 | 11.0000000000000000 | 21 | 60 + 40 | 10.0000000000000000 | 20 | 60 + 41 | 11.0000000000000000 | 21 | 60 +(10 rows) + +-- Partial aggregates are safe to push down without having clause +EXPLAIN (VERBOSE, COSTS OFF) +SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1; + QUERY PLAN +---------------------------------------------------------------------------------------------------------------------------------------------------------------- + Sort + Output: pagg_tab.b, (avg(pagg_tab.a)), (max(pagg_tab.a)), (count(*)) + Sort Key: pagg_tab.b + -> Finalize HashAggregate + Output: pagg_tab.b, avg(pagg_tab.a), max(pagg_tab.a), count(*) + Group Key: pagg_tab.b + -> Append + -> Foreign Scan + Output: pagg_tab.b, (PARTIAL avg(pagg_tab.a)), (PARTIAL max(pagg_tab.a)), (PARTIAL count(*)) + Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab) + Remote SQL: SELECT b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE FROM public.pagg_tab_p1 GROUP BY 1 + -> Foreign Scan + Output: pagg_tab_1.b, (PARTIAL avg(pagg_tab_1.a)), (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*)) + Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1) + Remote SQL: SELECT b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE FROM public.pagg_tab_p2 GROUP BY 1 + -> Foreign Scan + Output: pagg_tab_2.b, (PARTIAL avg(pagg_tab_2.a)), (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*)) + Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2) + Remote SQL: SELECT b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE FROM public.pagg_tab_p3 GROUP BY 1 +(19 rows) + +SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1; + b | avg | max | count +----+---------------------+-----+------- + 0 | 10.0000000000000000 | 20 | 60 + 1 | 11.0000000000000000 | 21 | 60 + 2 | 12.0000000000000000 | 22 | 60 + 3 | 13.0000000000000000 | 23 | 60 + 4 | 14.0000000000000000 | 24 | 60 + 5 | 15.0000000000000000 | 25 | 60 + 6 | 16.0000000000000000 | 26 | 60 + 7 | 17.0000000000000000 | 27 | 60 + 8 | 18.0000000000000000 | 28 | 60 + 9 | 19.0000000000000000 | 29 | 60 + 10 | 10.0000000000000000 | 20 | 60 + 11 | 11.0000000000000000 | 21 | 60 + 12 | 12.0000000000000000 | 22 | 60 + 13 | 13.0000000000000000 | 23 | 60 + 14 | 14.0000000000000000 | 24 | 60 + 15 | 15.0000000000000000 | 25 | 60 + 16 | 16.0000000000000000 | 26 | 60 + 17 | 17.0000000000000000 | 27 | 60 + 18 | 18.0000000000000000 | 28 | 60 + 19 | 19.0000000000000000 | 29 | 60 + 20 | 10.0000000000000000 | 20 | 60 + 21 | 11.0000000000000000 | 21 | 60 + 22 | 12.0000000000000000 | 22 | 60 + 23 | 13.0000000000000000 | 23 | 60 + 24 | 14.0000000000000000 | 24 | 60 + 25 | 15.0000000000000000 | 25 | 60 + 26 | 16.0000000000000000 | 26 | 60 + 27 | 17.0000000000000000 | 27 | 60 + 28 | 18.0000000000000000 | 28 | 60 + 29 | 19.0000000000000000 | 29 | 60 + 30 | 10.0000000000000000 | 20 | 60 + 31 | 11.0000000000000000 | 21 | 60 + 32 | 12.0000000000000000 | 22 | 60 + 33 | 13.0000000000000000 | 23 | 60 + 34 | 14.0000000000000000 | 24 | 60 + 35 | 15.0000000000000000 | 25 | 60 + 36 | 16.0000000000000000 | 26 | 60 + 37 | 17.0000000000000000 | 27 | 60 + 38 | 18.0000000000000000 | 28 | 60 + 39 | 19.0000000000000000 | 29 | 60 + 40 | 10.0000000000000000 | 20 | 60 + 41 | 11.0000000000000000 | 21 | 60 + 42 | 12.0000000000000000 | 22 | 60 + 43 | 13.0000000000000000 | 23 | 60 + 44 | 14.0000000000000000 | 24 | 60 + 45 | 15.0000000000000000 | 25 | 60 + 46 | 16.0000000000000000 | 26 | 60 + 47 | 17.0000000000000000 | 27 | 60 + 48 | 18.0000000000000000 | 28 | 60 + 49 | 19.0000000000000000 | 29 | 60 +(50 rows) + +-- Partial aggregates are safe to push down even if we need both variable and variable-based expression +EXPLAIN (VERBOSE, COSTS OFF) +SELECT avg(a), max(a), count(*), (b/2)::numeric FROM pagg_tab GROUP BY b/2 ORDER BY 4; + QUERY PLAN +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Sort + Output: (avg(pagg_tab.a)), (max(pagg_tab.a)), (count(*)), ((((pagg_tab.b / 2)))::numeric), ((pagg_tab.b / 2)) + Sort Key: ((((pagg_tab.b / 2)))::numeric) + -> Finalize HashAggregate + Output: avg(pagg_tab.a), max(pagg_tab.a), count(*), (((pagg_tab.b / 2)))::numeric, ((pagg_tab.b / 2)) + Group Key: ((pagg_tab.b / 2)) + -> Append + -> Foreign Scan + Output: ((pagg_tab.b / 2)), (PARTIAL avg(pagg_tab.a)), (PARTIAL max(pagg_tab.a)), (PARTIAL count(*)), pagg_tab.b + Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab) + Remote SQL: SELECT (b / 2), b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE FROM public.pagg_tab_p1 GROUP BY 1, 2 + -> Foreign Scan + Output: ((pagg_tab_1.b / 2)), (PARTIAL avg(pagg_tab_1.a)), (PARTIAL max(pagg_tab_1.a)), (PARTIAL count(*)), pagg_tab_1.b + Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1) + Remote SQL: SELECT (b / 2), b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE FROM public.pagg_tab_p2 GROUP BY 1, 2 + -> Foreign Scan + Output: ((pagg_tab_2.b / 2)), (PARTIAL avg(pagg_tab_2.a)), (PARTIAL max(pagg_tab_2.a)), (PARTIAL count(*)), pagg_tab_2.b + Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2) + Remote SQL: SELECT (b / 2), b, avg(a) PARTIAL_AGGREGATE , max(a) PARTIAL_AGGREGATE , count(*) PARTIAL_AGGREGATE FROM public.pagg_tab_p3 GROUP BY 1, 2 +(19 rows) + +SELECT avg(a), max(a), count(*), (b/2)::numeric FROM pagg_tab GROUP BY b/2 ORDER BY 4; + avg | max | count | numeric +---------------------+-----+-------+--------- + 10.5000000000000000 | 21 | 120 | 0 + 12.5000000000000000 | 23 | 120 | 1 + 14.5000000000000000 | 25 | 120 | 2 + 16.5000000000000000 | 27 | 120 | 3 + 18.5000000000000000 | 29 | 120 | 4 + 10.5000000000000000 | 21 | 120 | 5 + 12.5000000000000000 | 23 | 120 | 6 + 14.5000000000000000 | 25 | 120 | 7 + 16.5000000000000000 | 27 | 120 | 8 + 18.5000000000000000 | 29 | 120 | 9 + 10.5000000000000000 | 21 | 120 | 10 + 12.5000000000000000 | 23 | 120 | 11 + 14.5000000000000000 | 25 | 120 | 12 + 16.5000000000000000 | 27 | 120 | 13 + 18.5000000000000000 | 29 | 120 | 14 + 10.5000000000000000 | 21 | 120 | 15 + 12.5000000000000000 | 23 | 120 | 16 + 14.5000000000000000 | 25 | 120 | 17 + 16.5000000000000000 | 27 | 120 | 18 + 18.5000000000000000 | 29 | 120 | 19 + 10.5000000000000000 | 21 | 120 | 20 + 12.5000000000000000 | 23 | 120 | 21 + 14.5000000000000000 | 25 | 120 | 22 + 16.5000000000000000 | 27 | 120 | 23 + 18.5000000000000000 | 29 | 120 | 24 +(25 rows) + +-- Partial aggregates pushdown behaves sane with non-shipped grouping elements +EXPLAIN (VERBOSE, COSTS OFF) +SELECT avg(a), (b/2)::numeric, 'test' test FROM pagg_tab GROUP BY 3, b/2 ORDER BY 1, 2 LIMIT 10; + QUERY PLAN +-------------------------------------------------------------------------------------------------------------------------- + Limit + Output: (avg(pagg_tab.a)), ((((pagg_tab.b / 2)))::numeric), 'test'::text, ((pagg_tab.b / 2)) + -> Sort + Output: (avg(pagg_tab.a)), ((((pagg_tab.b / 2)))::numeric), 'test'::text, ((pagg_tab.b / 2)) + Sort Key: (avg(pagg_tab.a)), ((((pagg_tab.b / 2)))::numeric) + -> Finalize HashAggregate + Output: avg(pagg_tab.a), (((pagg_tab.b / 2)))::numeric, 'test'::text, ((pagg_tab.b / 2)) + Group Key: ((pagg_tab.b / 2)) + -> Append + -> Foreign Scan + Output: ((pagg_tab.b / 2)), (PARTIAL avg(pagg_tab.a)), pagg_tab.b + Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab) + Remote SQL: SELECT (b / 2), b, avg(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p1 GROUP BY 1, 2 + -> Foreign Scan + Output: ((pagg_tab_1.b / 2)), (PARTIAL avg(pagg_tab_1.a)), pagg_tab_1.b + Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1) + Remote SQL: SELECT (b / 2), b, avg(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p2 GROUP BY 1, 2 + -> Foreign Scan + Output: ((pagg_tab_2.b / 2)), (PARTIAL avg(pagg_tab_2.a)), pagg_tab_2.b + Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2) + Remote SQL: SELECT (b / 2), b, avg(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p3 GROUP BY 1, 2 +(21 rows) + +SELECT avg(a), (b/2)::numeric, 'test' test FROM pagg_tab GROUP BY 3, b/2 ORDER BY 1, 2 LIMIT 10; + avg | numeric | test +---------------------+---------+------ + 10.5000000000000000 | 0 | test + 10.5000000000000000 | 5 | test + 10.5000000000000000 | 10 | test + 10.5000000000000000 | 15 | test + 10.5000000000000000 | 20 | test + 12.5000000000000000 | 1 | test + 12.5000000000000000 | 6 | test + 12.5000000000000000 | 11 | test + 12.5000000000000000 | 16 | test + 12.5000000000000000 | 21 | test +(10 rows) + +-- Partial aggregates are safe to push down for all built-in aggregates +EXPLAIN (VERBOSE, COSTS OFF) +SELECT + /* The cases that don't require import or export functions */ + bit_and(c_bit), bit_and(c_1or3int2), bit_and(c_1or3int4), bit_and(c_1or3int8), + bit_or(c_bit), bit_or(c_1or3int2), bit_or(c_1or3int4), bit_or(c_1or3int8), + bit_xor(c_bit), bit_xor(c_1or3int2), bit_xor(c_1or3int4), bit_xor(c_1or3int8), + bool_and(c_bool), + bool_or(c_bool), + every(c_bool), + max(c_1c::char(1)), max('2000-01-01'::date + b), max('0.0.0.0'::inet + b), max(b::float4), max(b::float8), max(b::int2), max(b::int4), max(b::int8), max(c_interval), max(c_money), max(b::numeric), max(b::oid), max(c_pg_lsn), max(c_tid), max(c_1c), max(c_time), max(c_timetz), max(c_timestamp), max(c_timestamptz), max(c_1b), max(c_xid8), + min(c_1c::char(1)), min('2000-01-01'::date + b), min('0.0.0.0'::inet + b), min(b::float4), min(b::float8), min(b::int2), min(b::int4), min(b::int8), min(c_interval), min(c_money), min(b::numeric), min(b::oid), min(c_pg_lsn), min(c_tid), min(c_1c), min(c_time), min(c_timetz), min(c_timestamp), min(c_timestamptz), max(c_1b), min(c_xid8), + avg(b::int4) + FROM pagg_tab WHERE c_serial between 1 and 30; + QUERY PLAN +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- + Finalize Aggregate + Output: bit_and(pagg_tab.c_bit), bit_and(pagg_tab.c_1or3int2), bit_and(pagg_tab.c_1or3int4), bit_and(pagg_tab.c_1or3int8), bit_or(pagg_tab.c_bit), bit_or(pagg_tab.c_1or3int2), bit_or(pagg_tab.c_1or3int4), bit_or(pagg_tab.c_1or3int8), bit_xor(pagg_tab.c_bit), bit_xor(pagg_tab.c_1or3int2), bit_xor(pagg_tab.c_1or3int4), bit_xor(pagg_tab.c_1or3int8), bool_and(pagg_tab.c_bool), bool_or(pagg_tab.c_bool), every(pagg_tab.c_bool), max((pagg_tab.c_1c)::character(1)), max(('01-01-2000'::date + pagg_tab.b)), max(('0.0.0.0'::inet + (pagg_tab.b)::bigint)), max((pagg_tab.b)::real), max((pagg_tab.b)::double precision), max((pagg_tab.b)::smallint), max(pagg_tab.b), max((pagg_tab.b)::bigint), max(pagg_tab.c_interval), max(pagg_tab.c_money), max((pagg_tab.b)::numeric), max((pagg_tab.b)::oid), max(pagg_tab.c_pg_lsn), max(pagg_tab.c_tid), max(pagg_tab.c_1c), max(pagg_tab.c_time), max(pagg_tab.c_timetz), max(pagg_tab.c_timestamp), max(pagg_tab.c_timestamptz), max(pagg_tab.c_1b), max(pagg_tab.c_xid8), min((pagg_tab.c_1c)::character(1)), min(('01-01-2000'::date + pagg_tab.b)), min(('0.0.0.0'::inet + (pagg_tab.b)::bigint)), min((pagg_tab.b)::real), min((pagg_tab.b)::double precision), min((pagg_tab.b)::smallint), min(pagg_tab.b), min((pagg_tab.b)::bigint), min(pagg_tab.c_interval), min(pagg_tab.c_money), min((pagg_tab.b)::numeric), min((pagg_tab.b)::oid), min(pagg_tab.c_pg_lsn), min(pagg_tab.c_tid), min(pagg_tab.c_1c), min(pagg_tab.c_time), min(pagg_tab.c_timetz), min(pagg_tab.c_timestamp), min(pagg_tab.c_timestamptz), max(pagg_tab.c_1b), min(pagg_tab.c_xid8), avg(pagg_tab.b) + -> Append + -> Foreign Scan + Output: (PARTIAL bit_and(pagg_tab.c_bit)), (PARTIAL bit_and(pagg_tab.c_1or3int2)), (PARTIAL bit_and(pagg_tab.c_1or3int4)), (PARTIAL bit_and(pagg_tab.c_1or3int8)), (PARTIAL bit_or(pagg_tab.c_bit)), (PARTIAL bit_or(pagg_tab.c_1or3int2)), (PARTIAL bit_or(pagg_tab.c_1or3int4)), (PARTIAL bit_or(pagg_tab.c_1or3int8)), (PARTIAL bit_xor(pagg_tab.c_bit)), (PARTIAL bit_xor(pagg_tab.c_1or3int2)), (PARTIAL bit_xor(pagg_tab.c_1or3int4)), (PARTIAL bit_xor(pagg_tab.c_1or3int8)), (PARTIAL bool_and(pagg_tab.c_bool)), (PARTIAL bool_or(pagg_tab.c_bool)), (PARTIAL every(pagg_tab.c_bool)), (PARTIAL max((pagg_tab.c_1c)::character(1))), (PARTIAL max(('01-01-2000'::date + pagg_tab.b))), (PARTIAL max(('0.0.0.0'::inet + (pagg_tab.b)::bigint))), (PARTIAL max((pagg_tab.b)::real)), (PARTIAL max((pagg_tab.b)::double precision)), (PARTIAL max((pagg_tab.b)::smallint)), (PARTIAL max(pagg_tab.b)), (PARTIAL max((pagg_tab.b)::bigint)), (PARTIAL max(pagg_tab.c_interval)), (PARTIAL max(pagg_tab.c_money)), (PARTIAL max((pagg_tab.b)::numeric)), (PARTIAL max((pagg_tab.b)::oid)), (PARTIAL max(pagg_tab.c_pg_lsn)), (PARTIAL max(pagg_tab.c_tid)), (PARTIAL max(pagg_tab.c_1c)), (PARTIAL max(pagg_tab.c_time)), (PARTIAL max(pagg_tab.c_timetz)), (PARTIAL max(pagg_tab.c_timestamp)), (PARTIAL max(pagg_tab.c_timestamptz)), (PARTIAL max(pagg_tab.c_1b)), (PARTIAL max(pagg_tab.c_xid8)), (PARTIAL min((pagg_tab.c_1c)::character(1))), (PARTIAL min(('01-01-2000'::date + pagg_tab.b))), (PARTIAL min(('0.0.0.0'::inet + (pagg_tab.b)::bigint))), (PARTIAL min((pagg_tab.b)::real)), (PARTIAL min((pagg_tab.b)::double precision)), (PARTIAL min((pagg_tab.b)::smallint)), (PARTIAL min(pagg_tab.b)), (PARTIAL min((pagg_tab.b)::bigint)), (PARTIAL min(pagg_tab.c_interval)), (PARTIAL min(pagg_tab.c_money)), (PARTIAL min((pagg_tab.b)::numeric)), (PARTIAL min((pagg_tab.b)::oid)), (PARTIAL min(pagg_tab.c_pg_lsn)), (PARTIAL min(pagg_tab.c_tid)), (PARTIAL min(pagg_tab.c_1c)), (PARTIAL min(pagg_tab.c_time)), (PARTIAL min(pagg_tab.c_timetz)), (PARTIAL min(pagg_tab.c_timestamp)), (PARTIAL min(pagg_tab.c_timestamptz)), (PARTIAL min(pagg_tab.c_xid8)), (PARTIAL avg(pagg_tab.b)) + Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab) + Remote SQL: SELECT bit_and(c_bit) PARTIAL_AGGREGATE , bit_and(c_1or3int2) PARTIAL_AGGREGATE , bit_and(c_1or3int4) PARTIAL_AGGREGATE , bit_and(c_1or3int8) PARTIAL_AGGREGATE , bit_or(c_bit) PARTIAL_AGGREGATE , bit_or(c_1or3int2) PARTIAL_AGGREGATE , bit_or(c_1or3int4) PARTIAL_AGGREGATE , bit_or(c_1or3int8) PARTIAL_AGGREGATE , bit_xor(c_bit) PARTIAL_AGGREGATE , bit_xor(c_1or3int2) PARTIAL_AGGREGATE , bit_xor(c_1or3int4) PARTIAL_AGGREGATE , bit_xor(c_1or3int8) PARTIAL_AGGREGATE , bool_and(c_bool) PARTIAL_AGGREGATE , bool_or(c_bool) PARTIAL_AGGREGATE , every(c_bool) PARTIAL_AGGREGATE , max(c_1c::character(1)) PARTIAL_AGGREGATE , max(('01-01-2000'::date + b)) PARTIAL_AGGREGATE , max(('0.0.0.0'::inet + b)) PARTIAL_AGGREGATE , max(b::real) PARTIAL_AGGREGATE , max(b::double precision) PARTIAL_AGGREGATE , max(b::smallint) PARTIAL_AGGREGATE , max(b) PARTIAL_AGGREGATE , max(b::bigint) PARTIAL_AGGREGATE , max(c_interval) PARTIAL_AGGREGATE , max(c_money) PARTIAL_AGGREGATE , max(b::numeric) PARTIAL_AGGREGATE , max(b::oid) PARTIAL_AGGREGATE , max(c_pg_lsn) PARTIAL_AGGREGATE , max(c_tid) PARTIAL_AGGREGATE , max(c_1c) PARTIAL_AGGREGATE , max(c_time) PARTIAL_AGGREGATE , max(c_timetz) PARTIAL_AGGREGATE , max(c_timestamp) PARTIAL_AGGREGATE , max(c_timestamptz) PARTIAL_AGGREGATE , max(c_1b) PARTIAL_AGGREGATE , max(c_xid8) PARTIAL_AGGREGATE , min(c_1c::character(1)) PARTIAL_AGGREGATE , min(('01-01-2000'::date + b)) PARTIAL_AGGREGATE , min(('0.0.0.0'::inet + b)) PARTIAL_AGGREGATE , min(b::real) PARTIAL_AGGREGATE , min(b::double precision) PARTIAL_AGGREGATE , min(b::smallint) PARTIAL_AGGREGATE , min(b) PARTIAL_AGGREGATE , min(b::bigint) PARTIAL_AGGREGATE , min(c_interval) PARTIAL_AGGREGATE , min(c_money) PARTIAL_AGGREGATE , min(b::numeric) PARTIAL_AGGREGATE , min(b::oid) PARTIAL_AGGREGATE , min(c_pg_lsn) PARTIAL_AGGREGATE , min(c_tid) PARTIAL_AGGREGATE , min(c_1c) PARTIAL_AGGREGATE , min(c_time) PARTIAL_AGGREGATE , min(c_timetz) PARTIAL_AGGREGATE , min(c_timestamp) PARTIAL_AGGREGATE , min(c_timestamptz) PARTIAL_AGGREGATE , min(c_xid8) PARTIAL_AGGREGATE , avg(b) PARTIAL_AGGREGATE FROM public.pagg_tab_p1 WHERE ((c_serial >= 1)) AND ((c_serial <= 30)) + -> Foreign Scan + Output: (PARTIAL bit_and(pagg_tab_1.c_bit)), (PARTIAL bit_and(pagg_tab_1.c_1or3int2)), (PARTIAL bit_and(pagg_tab_1.c_1or3int4)), (PARTIAL bit_and(pagg_tab_1.c_1or3int8)), (PARTIAL bit_or(pagg_tab_1.c_bit)), (PARTIAL bit_or(pagg_tab_1.c_1or3int2)), (PARTIAL bit_or(pagg_tab_1.c_1or3int4)), (PARTIAL bit_or(pagg_tab_1.c_1or3int8)), (PARTIAL bit_xor(pagg_tab_1.c_bit)), (PARTIAL bit_xor(pagg_tab_1.c_1or3int2)), (PARTIAL bit_xor(pagg_tab_1.c_1or3int4)), (PARTIAL bit_xor(pagg_tab_1.c_1or3int8)), (PARTIAL bool_and(pagg_tab_1.c_bool)), (PARTIAL bool_or(pagg_tab_1.c_bool)), (PARTIAL every(pagg_tab_1.c_bool)), (PARTIAL max((pagg_tab_1.c_1c)::character(1))), (PARTIAL max(('01-01-2000'::date + pagg_tab_1.b))), (PARTIAL max(('0.0.0.0'::inet + (pagg_tab_1.b)::bigint))), (PARTIAL max((pagg_tab_1.b)::real)), (PARTIAL max((pagg_tab_1.b)::double precision)), (PARTIAL max((pagg_tab_1.b)::smallint)), (PARTIAL max(pagg_tab_1.b)), (PARTIAL max((pagg_tab_1.b)::bigint)), (PARTIAL max(pagg_tab_1.c_interval)), (PARTIAL max(pagg_tab_1.c_money)), (PARTIAL max((pagg_tab_1.b)::numeric)), (PARTIAL max((pagg_tab_1.b)::oid)), (PARTIAL max(pagg_tab_1.c_pg_lsn)), (PARTIAL max(pagg_tab_1.c_tid)), (PARTIAL max(pagg_tab_1.c_1c)), (PARTIAL max(pagg_tab_1.c_time)), (PARTIAL max(pagg_tab_1.c_timetz)), (PARTIAL max(pagg_tab_1.c_timestamp)), (PARTIAL max(pagg_tab_1.c_timestamptz)), (PARTIAL max(pagg_tab_1.c_1b)), (PARTIAL max(pagg_tab_1.c_xid8)), (PARTIAL min((pagg_tab_1.c_1c)::character(1))), (PARTIAL min(('01-01-2000'::date + pagg_tab_1.b))), (PARTIAL min(('0.0.0.0'::inet + (pagg_tab_1.b)::bigint))), (PARTIAL min((pagg_tab_1.b)::real)), (PARTIAL min((pagg_tab_1.b)::double precision)), (PARTIAL min((pagg_tab_1.b)::smallint)), (PARTIAL min(pagg_tab_1.b)), (PARTIAL min((pagg_tab_1.b)::bigint)), (PARTIAL min(pagg_tab_1.c_interval)), (PARTIAL min(pagg_tab_1.c_money)), (PARTIAL min((pagg_tab_1.b)::numeric)), (PARTIAL min((pagg_tab_1.b)::oid)), (PARTIAL min(pagg_tab_1.c_pg_lsn)), (PARTIAL min(pagg_tab_1.c_tid)), (PARTIAL min(pagg_tab_1.c_1c)), (PARTIAL min(pagg_tab_1.c_time)), (PARTIAL min(pagg_tab_1.c_timetz)), (PARTIAL min(pagg_tab_1.c_timestamp)), (PARTIAL min(pagg_tab_1.c_timestamptz)), (PARTIAL min(pagg_tab_1.c_xid8)), (PARTIAL avg(pagg_tab_1.b)) + Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1) + Remote SQL: SELECT bit_and(c_bit) PARTIAL_AGGREGATE , bit_and(c_1or3int2) PARTIAL_AGGREGATE , bit_and(c_1or3int4) PARTIAL_AGGREGATE , bit_and(c_1or3int8) PARTIAL_AGGREGATE , bit_or(c_bit) PARTIAL_AGGREGATE , bit_or(c_1or3int2) PARTIAL_AGGREGATE , bit_or(c_1or3int4) PARTIAL_AGGREGATE , bit_or(c_1or3int8) PARTIAL_AGGREGATE , bit_xor(c_bit) PARTIAL_AGGREGATE , bit_xor(c_1or3int2) PARTIAL_AGGREGATE , bit_xor(c_1or3int4) PARTIAL_AGGREGATE , bit_xor(c_1or3int8) PARTIAL_AGGREGATE , bool_and(c_bool) PARTIAL_AGGREGATE , bool_or(c_bool) PARTIAL_AGGREGATE , every(c_bool) PARTIAL_AGGREGATE , max(c_1c::character(1)) PARTIAL_AGGREGATE , max(('01-01-2000'::date + b)) PARTIAL_AGGREGATE , max(('0.0.0.0'::inet + b)) PARTIAL_AGGREGATE , max(b::real) PARTIAL_AGGREGATE , max(b::double precision) PARTIAL_AGGREGATE , max(b::smallint) PARTIAL_AGGREGATE , max(b) PARTIAL_AGGREGATE , max(b::bigint) PARTIAL_AGGREGATE , max(c_interval) PARTIAL_AGGREGATE , max(c_money) PARTIAL_AGGREGATE , max(b::numeric) PARTIAL_AGGREGATE , max(b::oid) PARTIAL_AGGREGATE , max(c_pg_lsn) PARTIAL_AGGREGATE , max(c_tid) PARTIAL_AGGREGATE , max(c_1c) PARTIAL_AGGREGATE , max(c_time) PARTIAL_AGGREGATE , max(c_timetz) PARTIAL_AGGREGATE , max(c_timestamp) PARTIAL_AGGREGATE , max(c_timestamptz) PARTIAL_AGGREGATE , max(c_1b) PARTIAL_AGGREGATE , max(c_xid8) PARTIAL_AGGREGATE , min(c_1c::character(1)) PARTIAL_AGGREGATE , min(('01-01-2000'::date + b)) PARTIAL_AGGREGATE , min(('0.0.0.0'::inet + b)) PARTIAL_AGGREGATE , min(b::real) PARTIAL_AGGREGATE , min(b::double precision) PARTIAL_AGGREGATE , min(b::smallint) PARTIAL_AGGREGATE , min(b) PARTIAL_AGGREGATE , min(b::bigint) PARTIAL_AGGREGATE , min(c_interval) PARTIAL_AGGREGATE , min(c_money) PARTIAL_AGGREGATE , min(b::numeric) PARTIAL_AGGREGATE , min(b::oid) PARTIAL_AGGREGATE , min(c_pg_lsn) PARTIAL_AGGREGATE , min(c_tid) PARTIAL_AGGREGATE , min(c_1c) PARTIAL_AGGREGATE , min(c_time) PARTIAL_AGGREGATE , min(c_timetz) PARTIAL_AGGREGATE , min(c_timestamp) PARTIAL_AGGREGATE , min(c_timestamptz) PARTIAL_AGGREGATE , min(c_xid8) PARTIAL_AGGREGATE , avg(b) PARTIAL_AGGREGATE FROM public.pagg_tab_p2 WHERE ((c_serial >= 1)) AND ((c_serial <= 30)) + -> Foreign Scan + Output: (PARTIAL bit_and(pagg_tab_2.c_bit)), (PARTIAL bit_and(pagg_tab_2.c_1or3int2)), (PARTIAL bit_and(pagg_tab_2.c_1or3int4)), (PARTIAL bit_and(pagg_tab_2.c_1or3int8)), (PARTIAL bit_or(pagg_tab_2.c_bit)), (PARTIAL bit_or(pagg_tab_2.c_1or3int2)), (PARTIAL bit_or(pagg_tab_2.c_1or3int4)), (PARTIAL bit_or(pagg_tab_2.c_1or3int8)), (PARTIAL bit_xor(pagg_tab_2.c_bit)), (PARTIAL bit_xor(pagg_tab_2.c_1or3int2)), (PARTIAL bit_xor(pagg_tab_2.c_1or3int4)), (PARTIAL bit_xor(pagg_tab_2.c_1or3int8)), (PARTIAL bool_and(pagg_tab_2.c_bool)), (PARTIAL bool_or(pagg_tab_2.c_bool)), (PARTIAL every(pagg_tab_2.c_bool)), (PARTIAL max((pagg_tab_2.c_1c)::character(1))), (PARTIAL max(('01-01-2000'::date + pagg_tab_2.b))), (PARTIAL max(('0.0.0.0'::inet + (pagg_tab_2.b)::bigint))), (PARTIAL max((pagg_tab_2.b)::real)), (PARTIAL max((pagg_tab_2.b)::double precision)), (PARTIAL max((pagg_tab_2.b)::smallint)), (PARTIAL max(pagg_tab_2.b)), (PARTIAL max((pagg_tab_2.b)::bigint)), (PARTIAL max(pagg_tab_2.c_interval)), (PARTIAL max(pagg_tab_2.c_money)), (PARTIAL max((pagg_tab_2.b)::numeric)), (PARTIAL max((pagg_tab_2.b)::oid)), (PARTIAL max(pagg_tab_2.c_pg_lsn)), (PARTIAL max(pagg_tab_2.c_tid)), (PARTIAL max(pagg_tab_2.c_1c)), (PARTIAL max(pagg_tab_2.c_time)), (PARTIAL max(pagg_tab_2.c_timetz)), (PARTIAL max(pagg_tab_2.c_timestamp)), (PARTIAL max(pagg_tab_2.c_timestamptz)), (PARTIAL max(pagg_tab_2.c_1b)), (PARTIAL max(pagg_tab_2.c_xid8)), (PARTIAL min((pagg_tab_2.c_1c)::character(1))), (PARTIAL min(('01-01-2000'::date + pagg_tab_2.b))), (PARTIAL min(('0.0.0.0'::inet + (pagg_tab_2.b)::bigint))), (PARTIAL min((pagg_tab_2.b)::real)), (PARTIAL min((pagg_tab_2.b)::double precision)), (PARTIAL min((pagg_tab_2.b)::smallint)), (PARTIAL min(pagg_tab_2.b)), (PARTIAL min((pagg_tab_2.b)::bigint)), (PARTIAL min(pagg_tab_2.c_interval)), (PARTIAL min(pagg_tab_2.c_money)), (PARTIAL min((pagg_tab_2.b)::numeric)), (PARTIAL min((pagg_tab_2.b)::oid)), (PARTIAL min(pagg_tab_2.c_pg_lsn)), (PARTIAL min(pagg_tab_2.c_tid)), (PARTIAL min(pagg_tab_2.c_1c)), (PARTIAL min(pagg_tab_2.c_time)), (PARTIAL min(pagg_tab_2.c_timetz)), (PARTIAL min(pagg_tab_2.c_timestamp)), (PARTIAL min(pagg_tab_2.c_timestamptz)), (PARTIAL min(pagg_tab_2.c_xid8)), (PARTIAL avg(pagg_tab_2.b)) + Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2) + Remote SQL: SELECT bit_and(c_bit) PARTIAL_AGGREGATE , bit_and(c_1or3int2) PARTIAL_AGGREGATE , bit_and(c_1or3int4) PARTIAL_AGGREGATE , bit_and(c_1or3int8) PARTIAL_AGGREGATE , bit_or(c_bit) PARTIAL_AGGREGATE , bit_or(c_1or3int2) PARTIAL_AGGREGATE , bit_or(c_1or3int4) PARTIAL_AGGREGATE , bit_or(c_1or3int8) PARTIAL_AGGREGATE , bit_xor(c_bit) PARTIAL_AGGREGATE , bit_xor(c_1or3int2) PARTIAL_AGGREGATE , bit_xor(c_1or3int4) PARTIAL_AGGREGATE , bit_xor(c_1or3int8) PARTIAL_AGGREGATE , bool_and(c_bool) PARTIAL_AGGREGATE , bool_or(c_bool) PARTIAL_AGGREGATE , every(c_bool) PARTIAL_AGGREGATE , max(c_1c::character(1)) PARTIAL_AGGREGATE , max(('01-01-2000'::date + b)) PARTIAL_AGGREGATE , max(('0.0.0.0'::inet + b)) PARTIAL_AGGREGATE , max(b::real) PARTIAL_AGGREGATE , max(b::double precision) PARTIAL_AGGREGATE , max(b::smallint) PARTIAL_AGGREGATE , max(b) PARTIAL_AGGREGATE , max(b::bigint) PARTIAL_AGGREGATE , max(c_interval) PARTIAL_AGGREGATE , max(c_money) PARTIAL_AGGREGATE , max(b::numeric) PARTIAL_AGGREGATE , max(b::oid) PARTIAL_AGGREGATE , max(c_pg_lsn) PARTIAL_AGGREGATE , max(c_tid) PARTIAL_AGGREGATE , max(c_1c) PARTIAL_AGGREGATE , max(c_time) PARTIAL_AGGREGATE , max(c_timetz) PARTIAL_AGGREGATE , max(c_timestamp) PARTIAL_AGGREGATE , max(c_timestamptz) PARTIAL_AGGREGATE , max(c_1b) PARTIAL_AGGREGATE , max(c_xid8) PARTIAL_AGGREGATE , min(c_1c::character(1)) PARTIAL_AGGREGATE , min(('01-01-2000'::date + b)) PARTIAL_AGGREGATE , min(('0.0.0.0'::inet + b)) PARTIAL_AGGREGATE , min(b::real) PARTIAL_AGGREGATE , min(b::double precision) PARTIAL_AGGREGATE , min(b::smallint) PARTIAL_AGGREGATE , min(b) PARTIAL_AGGREGATE , min(b::bigint) PARTIAL_AGGREGATE , min(c_interval) PARTIAL_AGGREGATE , min(c_money) PARTIAL_AGGREGATE , min(b::numeric) PARTIAL_AGGREGATE , min(b::oid) PARTIAL_AGGREGATE , min(c_pg_lsn) PARTIAL_AGGREGATE , min(c_tid) PARTIAL_AGGREGATE , min(c_1c) PARTIAL_AGGREGATE , min(c_time) PARTIAL_AGGREGATE , min(c_timetz) PARTIAL_AGGREGATE , min(c_timestamp) PARTIAL_AGGREGATE , min(c_timestamptz) PARTIAL_AGGREGATE , min(c_xid8) PARTIAL_AGGREGATE , avg(b) PARTIAL_AGGREGATE FROM public.pagg_tab_p3 WHERE ((c_serial >= 1)) AND ((c_serial <= 30)) +(15 rows) + +SELECT + /* The cases that don't require import or export functions */ + bit_and(c_bit), bit_and(c_1or3int2), bit_and(c_1or3int4), bit_and(c_1or3int8), + bit_or(c_bit), bit_or(c_1or3int2), bit_or(c_1or3int4), bit_or(c_1or3int8), + bit_xor(c_bit), bit_xor(c_1or3int2), bit_xor(c_1or3int4), bit_xor(c_1or3int8), + bool_and(c_bool), + bool_or(c_bool), + every(c_bool), + max(c_1c::char(1)), max('2000-01-01'::date + b), max('0.0.0.0'::inet + b), max(b::float4), max(b::float8), max(b::int2), max(b::int4), max(b::int8), max(c_interval), max(c_money), max(b::numeric), max(b::oid), max(c_pg_lsn), max(c_tid), max(c_1c), max(c_time), max(c_timetz), max(c_timestamp), max(c_timestamptz), max(c_1b), max(c_xid8), + min(c_1c::char(1)), min('2000-01-01'::date + b), min('0.0.0.0'::inet + b), min(b::float4), min(b::float8), min(b::int2), min(b::int4), min(b::int8), min(c_interval), min(c_money), min(b::numeric), min(b::oid), min(c_pg_lsn), min(c_tid), min(c_1c), min(c_time), min(c_timetz), min(c_timestamp), min(c_timestamptz), max(c_1b), min(c_xid8), + avg(b::int4) + FROM pagg_tab WHERE c_serial between 1 and 30; + bit_and | bit_and | bit_and | bit_and | bit_or | bit_or | bit_or | bit_or | bit_xor | bit_xor | bit_xor | bit_xor | bool_and | bool_or | every | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | max | min | min | min | min | min | min | min | min | min | min | min | min | min | min | min | min | min | min | min | max | min | avg +---------+---------+---------+---------+--------+--------+--------+--------+---------+---------+---------+---------+----------+---------+-------+-----+------------+----------+-----+-----+-----+-----+-----+---------+-------+-----+-----+------+--------+-----+----------+-------------+--------------------------+------------------------------+------+-----+-----+------------+---------+-----+-----+-----+-----+-----+-----+-------+-----+-----+-----+-------+-----+----------+-------------+--------------------------+------------------------------+------+-----+--------------------- + 01 | 1 | 1 | 1 | 11 | 3 | 3 | 3 | 10 | 2 | 2 | 2 | f | t | f | 9 | 01-31-2000 | 0.0.0.30 | 30 | 30 | 30 | 30 | 30 | @ 1 sec | $1.00 | 30 | 30 | 0/30 | (0,30) | 9 | 00:00:30 | 00:00:30+00 | Sat Jan 01 00:00:30 2000 | Sat Jan 01 00:00:30 2000 UTC | \x39 | 9 | 0 | 01-02-2000 | 0.0.0.1 | 1 | 1 | 1 | 1 | 1 | @ 0 | $0.00 | 1 | 1 | 0/1 | (0,1) | 0 | 00:00:01 | 00:00:01+00 | Sat Jan 01 00:00:01 2000 | Sat Jan 01 00:00:01 2000 UTC | \x39 | 0 | 15.5000000000000000 +(1 row) + +-- Tests for partial_aggregate_support +ALTER SERVER loopback OPTIONS (ADD partial_aggregate_support 'false'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; + QUERY PLAN +--------------------------------------------------------------------------- Finalize GroupAggregate + Output: pagg_tab.b, max(pagg_tab.a) Group Key: pagg_tab.b - Filter: (sum(pagg_tab.a) < 700) + -> Sort + Output: pagg_tab.b, (PARTIAL max(pagg_tab.a)) + Sort Key: pagg_tab.b + -> Append + -> Partial HashAggregate + Output: pagg_tab.b, PARTIAL max(pagg_tab.a) + Group Key: pagg_tab.b + -> Foreign Scan on public.fpagg_tab_p1 pagg_tab + Output: pagg_tab.b, pagg_tab.a + Remote SQL: SELECT a, b FROM public.pagg_tab_p1 + -> Partial HashAggregate + Output: pagg_tab_1.b, PARTIAL max(pagg_tab_1.a) + Group Key: pagg_tab_1.b + -> Foreign Scan on public.fpagg_tab_p2 pagg_tab_1 + Output: pagg_tab_1.b, pagg_tab_1.a + Remote SQL: SELECT a, b FROM public.pagg_tab_p2 + -> Partial HashAggregate + Output: pagg_tab_2.b, PARTIAL max(pagg_tab_2.a) + Group Key: pagg_tab_2.b + -> Foreign Scan on public.fpagg_tab_p3 pagg_tab_2 + Output: pagg_tab_2.b, pagg_tab_2.a + Remote SQL: SELECT a, b FROM public.pagg_tab_p3 +(25 rows) + +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; + b | max +----+----- + 0 | 20 + 1 | 21 + 2 | 22 + 3 | 23 + 4 | 24 + 5 | 25 + 6 | 26 + 7 | 27 + 8 | 28 + 9 | 29 + 10 | 20 + 11 | 21 + 12 | 22 + 13 | 23 + 14 | 24 + 15 | 25 + 16 | 26 + 17 | 27 + 18 | 28 + 19 | 29 + 20 | 20 + 21 | 21 + 22 | 22 + 23 | 23 + 24 | 24 + 25 | 25 + 26 | 26 + 27 | 27 + 28 | 28 + 29 | 29 + 30 | 20 + 31 | 21 + 32 | 22 + 33 | 23 + 34 | 24 + 35 | 25 + 36 | 26 + 37 | 27 + 38 | 28 + 39 | 29 + 40 | 20 + 41 | 21 + 42 | 22 + 43 | 23 + 44 | 24 + 45 | 25 + 46 | 26 + 47 | 27 + 48 | 28 + 49 | 29 +(50 rows) + +ALTER SERVER loopback OPTIONS (SET partial_aggregate_support 'true'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; + QUERY PLAN +-------------------------------------------------------------------------------------------------------- + Sort + Output: pagg_tab.b, (max(pagg_tab.a)) + Sort Key: pagg_tab.b + -> Finalize HashAggregate + Output: pagg_tab.b, max(pagg_tab.a) + Group Key: pagg_tab.b + -> Append + -> Foreign Scan + Output: pagg_tab.b, (PARTIAL max(pagg_tab.a)) + Relations: Aggregate on (public.fpagg_tab_p1 pagg_tab) + Remote SQL: SELECT b, max(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p1 GROUP BY 1 + -> Foreign Scan + Output: pagg_tab_1.b, (PARTIAL max(pagg_tab_1.a)) + Relations: Aggregate on (public.fpagg_tab_p2 pagg_tab_1) + Remote SQL: SELECT b, max(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p2 GROUP BY 1 + -> Foreign Scan + Output: pagg_tab_2.b, (PARTIAL max(pagg_tab_2.a)) + Relations: Aggregate on (public.fpagg_tab_p3 pagg_tab_2) + Remote SQL: SELECT b, max(a) PARTIAL_AGGREGATE FROM public.pagg_tab_p3 GROUP BY 1 +(19 rows) + +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; + b | max +----+----- + 0 | 20 + 1 | 21 + 2 | 22 + 3 | 23 + 4 | 24 + 5 | 25 + 6 | 26 + 7 | 27 + 8 | 28 + 9 | 29 + 10 | 20 + 11 | 21 + 12 | 22 + 13 | 23 + 14 | 24 + 15 | 25 + 16 | 26 + 17 | 27 + 18 | 28 + 19 | 29 + 20 | 20 + 21 | 21 + 22 | 22 + 23 | 23 + 24 | 24 + 25 | 25 + 26 | 26 + 27 | 27 + 28 | 28 + 29 | 29 + 30 | 20 + 31 | 21 + 32 | 22 + 33 | 23 + 34 | 24 + 35 | 25 + 36 | 26 + 37 | 27 + 38 | 28 + 39 | 29 + 40 | 20 + 41 | 21 + 42 | 22 + 43 | 23 + 44 | 24 + 45 | 25 + 46 | 26 + 47 | 27 + 48 | 28 + 49 | 29 +(50 rows) + +ALTER SERVER loopback OPTIONS (SET fdw_tuple_cost '1.0'); +SET enable_partitionwise_join=on; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT max(t1.b) FROM pagg_tab t1 JOIN pagg_tab t2 USING(a); + QUERY PLAN +-------------------------------------------------------------------------------------------------------------------------------------------------- + Finalize Aggregate + Output: max(t1.b) + -> Append + -> Foreign Scan + Output: (PARTIAL max(t1.b)) + Relations: Aggregate on ((public.fpagg_tab_p1 t1) INNER JOIN (public.fpagg_tab_p1 t2)) + Remote SQL: SELECT max(r4.b) PARTIAL_AGGREGATE FROM (public.pagg_tab_p1 r4 INNER JOIN public.pagg_tab_p1 r7 ON (((r4.a = r7.a)))) + -> Foreign Scan + Output: (PARTIAL max(t1_1.b)) + Relations: Aggregate on ((public.fpagg_tab_p2 t1_1) INNER JOIN (public.fpagg_tab_p2 t2_1)) + Remote SQL: SELECT max(r5.b) PARTIAL_AGGREGATE FROM (public.pagg_tab_p2 r5 INNER JOIN public.pagg_tab_p2 r8 ON (((r5.a = r8.a)))) + -> Foreign Scan + Output: (PARTIAL max(t1_2.b)) + Relations: Aggregate on ((public.fpagg_tab_p3 t1_2) INNER JOIN (public.fpagg_tab_p3 t2_2)) + Remote SQL: SELECT max(r6.b) PARTIAL_AGGREGATE FROM (public.pagg_tab_p3 r6 INNER JOIN public.pagg_tab_p3 r9 ON (((r6.a = r9.a)))) +(15 rows) + +SELECT max(t1.b) FROM pagg_tab t1 JOIN pagg_tab t2 USING(a); + max +----- + 49 +(1 row) + +RESET enable_partitionwise_join; +ALTER SERVER loopback OPTIONS (SET fdw_tuple_cost '0.1'); +ALTER SERVER loopback OPTIONS (DROP partial_aggregate_support); +-- It is unsafe to push down partial aggregates which contain DISTINCT clauses +EXPLAIN (VERBOSE, COSTS OFF) +SELECT max(a), count(distinct b) FROM pagg_tab; + QUERY PLAN +----------------------------------------------------------------------------------------- + Aggregate + Output: max(pagg_tab.a), count(DISTINCT pagg_tab.b) -> Merge Append Sort Key: pagg_tab.b - -> Partial GroupAggregate - Group Key: pagg_tab.b - -> Foreign Scan on fpagg_tab_p1 pagg_tab - -> Partial GroupAggregate - Group Key: pagg_tab_1.b - -> Foreign Scan on fpagg_tab_p2 pagg_tab_1 - -> Partial GroupAggregate - Group Key: pagg_tab_2.b - -> Foreign Scan on fpagg_tab_p3 pagg_tab_2 -(14 rows) + -> Foreign Scan on public.fpagg_tab_p1 pagg_tab_1 + Output: pagg_tab_1.a, pagg_tab_1.b + Remote SQL: SELECT a, b FROM public.pagg_tab_p1 ORDER BY b ASC NULLS LAST + -> Foreign Scan on public.fpagg_tab_p2 pagg_tab_2 + Output: pagg_tab_2.a, pagg_tab_2.b + Remote SQL: SELECT a, b FROM public.pagg_tab_p2 ORDER BY b ASC NULLS LAST + -> Foreign Scan on public.fpagg_tab_p3 pagg_tab_3 + Output: pagg_tab_3.a, pagg_tab_3.b + Remote SQL: SELECT a, b FROM public.pagg_tab_p3 ORDER BY b ASC NULLS LAST +(13 rows) + +SELECT max(a), count(distinct b) FROM pagg_tab; + max | count +-----+------- + 29 | 50 +(1 row) + +-- It is unsafe to push down partial aggregates which contain ORDER BY clauses +EXPLAIN (VERBOSE, COSTS OFF) SELECT array_agg(b order by b) FROM pagg_tab WHERE c_serial between 1 and 30; + QUERY PLAN +----------------------------------------------------------------------------------------------------------------- + Aggregate + Output: array_agg(pagg_tab.b ORDER BY pagg_tab.b) + -> Sort + Output: pagg_tab.b + Sort Key: pagg_tab.b + -> Append + -> Foreign Scan on public.fpagg_tab_p1 pagg_tab_1 + Output: pagg_tab_1.b + Remote SQL: SELECT b FROM public.pagg_tab_p1 WHERE ((c_serial >= 1)) AND ((c_serial <= 30)) + -> Foreign Scan on public.fpagg_tab_p2 pagg_tab_2 + Output: pagg_tab_2.b + Remote SQL: SELECT b FROM public.pagg_tab_p2 WHERE ((c_serial >= 1)) AND ((c_serial <= 30)) + -> Foreign Scan on public.fpagg_tab_p3 pagg_tab_3 + Output: pagg_tab_3.b + Remote SQL: SELECT b FROM public.pagg_tab_p3 WHERE ((c_serial >= 1)) AND ((c_serial <= 30)) +(15 rows) + +SELECT array_agg(b order by b) FROM pagg_tab WHERE c_serial between 1 and 30; + array_agg +------------------------------------------------------------------------------------ + {1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30} +(1 row) +ALTER SERVER loopback OPTIONS (DROP fdw_tuple_cost); +RESET TimeZone; -- =================================================================== -- access rights and superuser -- =================================================================== diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c index d0766f007d2..d3aad176942 100644 --- a/contrib/postgres_fdw/option.c +++ b/contrib/postgres_fdw/option.c @@ -125,7 +125,8 @@ postgres_fdw_validator(PG_FUNCTION_ARGS) strcmp(def->defname, "async_capable") == 0 || strcmp(def->defname, "parallel_commit") == 0 || strcmp(def->defname, "parallel_abort") == 0 || - strcmp(def->defname, "keep_connections") == 0) + strcmp(def->defname, "keep_connections") == 0 || + strcmp(def->defname, "partial_aggregate_support") == 0) { /* these accept only boolean values */ (void) defGetBoolean(def); @@ -267,6 +268,7 @@ InitPgFdwOptions(void) /* batch_size is available on both server and table */ {"batch_size", ForeignServerRelationId, false}, {"batch_size", ForeignTableRelationId, false}, + {"partial_aggregate_support", ForeignServerRelationId, false}, /* async_capable is available on both server and table */ {"async_capable", ForeignServerRelationId, false}, {"async_capable", ForeignTableRelationId, false}, diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index 1131a8bf77e..bf75f347fc0 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -18,6 +18,8 @@ #include "access/sysattr.h" #include "access/table.h" #include "catalog/pg_opfamily.h" +#include "catalog/pg_aggregate.h" +#include "catalog/pg_proc.h" #include "commands/defrem.h" #include "commands/explain.h" #include "commands/explain_format.h" @@ -34,6 +36,7 @@ #include "optimizer/pathnode.h" #include "optimizer/paths.h" #include "optimizer/planmain.h" +#include "optimizer/planner.h" #include "optimizer/prep.h" #include "optimizer/restrictinfo.h" #include "optimizer/tlist.h" @@ -48,6 +51,7 @@ #include "utils/rel.h" #include "utils/sampling.h" #include "utils/selfuncs.h" +#include "utils/syscache.h" PG_MODULE_MAGIC; @@ -519,7 +523,7 @@ static bool foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel, JoinType jointype, RelOptInfo *outerrel, RelOptInfo *innerrel, JoinPathExtraData *extra); static bool foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel, - Node *havingQual); + GroupPathExtraData *extra); static List *get_useful_pathkeys_for_relation(PlannerInfo *root, RelOptInfo *rel); static List *get_useful_ecs_for_relation(PlannerInfo *root, RelOptInfo *rel); @@ -650,6 +654,8 @@ postgresGetForeignRelSize(PlannerInfo *root, fpinfo->fdw_tuple_cost = DEFAULT_FDW_TUPLE_COST; fpinfo->shippable_extensions = NIL; fpinfo->fetch_size = 100; + fpinfo->remoteversion = 0; + fpinfo->partial_aggregate_support = true; fpinfo->async_capable = false; apply_server_options(fpinfo); @@ -661,7 +667,7 @@ postgresGetForeignRelSize(PlannerInfo *root, * should match what ExecCheckPermissions() does. If we fail due to lack * of permissions, the query would have failed at runtime anyway. */ - if (fpinfo->use_remote_estimate) + if (fpinfo->use_remote_estimate || fpinfo->partial_aggregate_support) { Oid userid; @@ -6052,9 +6058,9 @@ foreign_join_ok(PlannerInfo *root, RelOptInfo *joinrel, JoinType jointype, fpinfo->pushdown_safe = true; /* Get user mapping */ - if (fpinfo->use_remote_estimate) + if (fpinfo->use_remote_estimate || fpinfo->partial_aggregate_support) { - if (fpinfo_o->use_remote_estimate) + if (fpinfo_o->use_remote_estimate || fpinfo_o->partial_aggregate_support) fpinfo->user = fpinfo_o->user; else fpinfo->user = fpinfo_i->user; @@ -6229,6 +6235,8 @@ apply_server_options(PgFdwRelationInfo *fpinfo) if (strcmp(def->defname, "use_remote_estimate") == 0) fpinfo->use_remote_estimate = defGetBoolean(def); + else if (strcmp(def->defname, "partial_aggregate_support") == 0) + fpinfo->partial_aggregate_support = defGetBoolean(def); else if (strcmp(def->defname, "fdw_startup_cost") == 0) (void) parse_real(defGetString(def), &fpinfo->fdw_startup_cost, 0, NULL); @@ -6299,6 +6307,8 @@ merge_fdw_options(PgFdwRelationInfo *fpinfo, fpinfo->shippable_extensions = fpinfo_o->shippable_extensions; fpinfo->use_remote_estimate = fpinfo_o->use_remote_estimate; fpinfo->fetch_size = fpinfo_o->fetch_size; + fpinfo->remoteversion = fpinfo_o->remoteversion; + fpinfo->partial_aggregate_support = fpinfo_o->partial_aggregate_support; fpinfo->async_capable = fpinfo_o->async_capable; /* Merge the table level options from either side of the join. */ @@ -6485,7 +6495,7 @@ postgresGetForeignJoinPaths(PlannerInfo *root, */ static bool foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel, - Node *havingQual) + GroupPathExtraData *extra) { Query *query = root->parse; PgFdwRelationInfo *fpinfo = (PgFdwRelationInfo *) grouped_rel->fdw_private; @@ -6494,6 +6504,7 @@ foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel, ListCell *lc; int i; List *tlist = NIL; + bool partial = extra->patype == PARTITIONWISE_AGGREGATE_PARTIAL; /* We currently don't support pushing Grouping Sets. */ if (query->groupingSets) @@ -6527,6 +6538,12 @@ foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel, * a node, as long as it's not at top level; then no match is possible. */ i = 0; + fpinfo->group_clause = query->groupClause; + if (partial) + { + fpinfo->group_clause = extra->groupClausePartial; + grouping_target = extra->partial_target; + } foreach(lc, grouping_target->exprs) { Expr *expr = (Expr *) lfirst(lc); @@ -6538,7 +6555,7 @@ foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel, * check the whole GROUP BY clause not just processed_groupClause, * because we will ship all of it, cf. appendGroupByClause. */ - if (sgref && get_sortgroupref_clause_noerr(sgref, query->groupClause)) + if (sgref && get_sortgroupref_clause_noerr(sgref, fpinfo->group_clause)) { TargetEntry *tle; @@ -6624,9 +6641,9 @@ foreign_grouping_ok(PlannerInfo *root, RelOptInfo *grouped_rel, * Classify the pushable and non-pushable HAVING clauses and save them in * remote_conds and local_conds of the grouped rel's fpinfo. */ - if (havingQual) + if (extra->havingQual && !partial) { - foreach(lc, (List *) havingQual) + foreach(lc, (List *) extra->havingQual) { Expr *expr = (Expr *) lfirst(lc); RestrictInfo *rinfo; @@ -6740,6 +6757,7 @@ postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage, /* Ignore stages we don't support; and skip any duplicate calls. */ if ((stage != UPPERREL_GROUP_AGG && + stage != UPPERREL_PARTIAL_GROUP_AGG && stage != UPPERREL_ORDERED && stage != UPPERREL_FINAL) || output_rel->fdw_private) @@ -6756,6 +6774,10 @@ postgresGetForeignUpperPaths(PlannerInfo *root, UpperRelationKind stage, add_foreign_grouping_paths(root, input_rel, output_rel, (GroupPathExtraData *) extra); break; + case UPPERREL_PARTIAL_GROUP_AGG: + add_foreign_grouping_paths(root, input_rel, output_rel, + (GroupPathExtraData *) extra); + break; case UPPERREL_ORDERED: add_foreign_ordered_paths(root, input_rel, output_rel); break; @@ -6797,7 +6819,8 @@ add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel, return; Assert(extra->patype == PARTITIONWISE_AGGREGATE_NONE || - extra->patype == PARTITIONWISE_AGGREGATE_FULL); + extra->patype == PARTITIONWISE_AGGREGATE_FULL || + extra->patype == PARTITIONWISE_AGGREGATE_PARTIAL); /* save the input_rel as outerrel in fpinfo */ fpinfo->outerrel = input_rel; @@ -6817,7 +6840,7 @@ add_foreign_grouping_paths(PlannerInfo *root, RelOptInfo *input_rel, * Use HAVING qual from extra. In case of child partition, it will have * translated Vars. */ - if (!foreign_grouping_ok(root, grouped_rel, extra->havingQual)) + if (!foreign_grouping_ok(root, grouped_rel, extra)) return; /* diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index 81358f3bde7..a9874b07c35 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -85,7 +85,16 @@ typedef struct PgFdwRelationInfo /* Cached catalog information. */ ForeignTable *table; ForeignServer *server; - UserMapping *user; /* only set in use_remote_estimate mode */ + UserMapping *user; /* only set in use_remote_estimate/partial_aggregate_support mode */ + + /* for partial aggregate pushdown */ + bool partial_aggregate_support; + + /* + * If remoteversion is zero, it means the remote server version has not + * been acquired. + */ + int remoteversion; int fetch_size; /* fetch size for this remote table */ @@ -111,6 +120,7 @@ typedef struct PgFdwRelationInfo /* Grouping information */ List *grouped_tlist; + List *group_clause; /* Subquery information */ bool make_outerrel_subquery; /* do we deparse outerrel as a diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index d45e9f8ab52..5901081c138 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -3201,16 +3201,33 @@ RESET enable_partitionwise_join; -- =================================================================== -- test partitionwise aggregates -- =================================================================== - -CREATE TABLE pagg_tab (a int, b int, c text) PARTITION BY RANGE(a); +ALTER SERVER loopback OPTIONS (ADD fdw_tuple_cost '0.1'); + +CREATE TYPE mood AS enum ('sad', 'ok', 'happy'); +ALTER EXTENSION postgres_fdw ADD TYPE mood; + +CREATE TABLE pagg_tab (a int, b int, c text, c_serial int4, + c_int4array _int4, c_interval interval, + c_money money, c_1c text, c_1b bytea, + c_bit bit(2), c_1or3int2 int2, + c_1or3int4 int4, c_1or3int8 int8, + c_bool bool, c_enum mood, c_pg_lsn pg_lsn, + c_tid tid, c_int4range int4range, + c_int4multirange int4multirange, + c_time time, c_timetz timetz, + c_timestamp timestamp, c_timestamptz timestamptz, + c_xid8 xid8) + PARTITION BY RANGE(a); CREATE TABLE pagg_tab_p1 (LIKE pagg_tab); CREATE TABLE pagg_tab_p2 (LIKE pagg_tab); CREATE TABLE pagg_tab_p3 (LIKE pagg_tab); -INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 10; -INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10; -INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000') FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20; +SET TimeZone = 'UTC'; + +INSERT INTO pagg_tab_p1 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i, array[(i % 2), 0], ((i % 2) || ' seconds')::interval, (i % 2)::money, ((i - 1) % 10)::text, (((i - 1) % 10)::text)::bytea, case when (i % 2) = 0 then B'01' else B'11' end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then true else false end, (case when (i % 2) = 0 then 'sad' else 'happy' end)::mood, ('0/' || i)::pg_lsn, ('(0,' || i || ')')::tid, int4range(0, i), int4multirange(int4range(0, i), int4range(100, 100 + i)), '00:00:00'::time + (i || ' seconds')::interval, '00:00:00'::timetz + (i || ' seconds')::interval, '2000-01-01'::timestamp + (i || ' seconds')::interval, '2000-01-01'::timestamptz + (i || ' seconds')::interval, ((i % 10)::text)::xid8 FROM generate_series(1, 3000) i WHERE (i % 30) < 10; +INSERT INTO pagg_tab_p2 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i, array[(i % 2), 0], ((i % 2) || ' seconds')::interval, (i % 2)::money, ((i - 1) % 10)::text, (((i - 1) % 10)::text)::bytea, case when (i % 2) = 0 then B'01' else B'11' end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then true else false end, (case when (i % 2) = 0 then 'sad' else 'happy' end)::mood, ('0/' || i)::pg_lsn, ('(0,' || i || ')')::tid, int4range(0, i), int4multirange(int4range(0, i), int4range(100, 100 + i)), '00:00:00'::time + (i || ' seconds')::interval, '00:00:00'::timetz + (i || ' seconds')::interval, '2000-01-01'::timestamp + (i || ' seconds')::interval, '2000-01-01'::timestamptz + (i || ' seconds')::interval, ((i % 10)::text)::xid8 FROM generate_series(1, 3000) i WHERE (i % 30) < 20 and (i % 30) >= 10; +INSERT INTO pagg_tab_p3 SELECT i % 30, i % 50, to_char(i/30, 'FM0000'), i, array[(i % 2), 0], ((i % 2) || ' seconds')::interval, (i % 2)::money, ((i - 1) % 10)::text, (((i - 1) % 10)::text)::bytea, case when (i % 2) = 0 then B'01' else B'11' end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then 1 else 3 end, case when (i % 2) = 0 then true else false end, (case when (i % 2) = 0 then 'sad' else 'happy' end)::mood, ('0/' || i)::pg_lsn, ('(0,' || i || ')')::tid, int4range(0, i), int4multirange(int4range(0, i), int4range(100, 100 + i)), '00:00:00'::time + (i || ' seconds')::interval, '00:00:00'::timetz + (i || ' seconds')::interval, '2000-01-01'::timestamp + (i || ' seconds')::interval, '2000-01-01'::timestamptz + (i || ' seconds')::interval, ((i % 10)::text)::xid8 FROM generate_series(1, 3000) i WHERE (i % 30) < 30 and (i % 30) >= 20; -- Create foreign partitions CREATE FOREIGN TABLE fpagg_tab_p1 PARTITION OF pagg_tab FOR VALUES FROM (0) TO (10) SERVER loopback OPTIONS (table_name 'pagg_tab_p1'); @@ -3218,9 +3235,13 @@ CREATE FOREIGN TABLE fpagg_tab_p2 PARTITION OF pagg_tab FOR VALUES FROM (10) TO CREATE FOREIGN TABLE fpagg_tab_p3 PARTITION OF pagg_tab FOR VALUES FROM (20) TO (30) SERVER loopback OPTIONS (table_name 'pagg_tab_p3'); ANALYZE pagg_tab; +ANALYZE pagg_tab_p1; +ANALYZE pagg_tab_p2; +ANALYZE pagg_tab_p3; ANALYZE fpagg_tab_p1; ANALYZE fpagg_tab_p2; ANALYZE fpagg_tab_p3; +SET extra_float_digits = 0; -- When GROUP BY clause matches with PARTITION KEY. -- Plan with partitionwise aggregates is disabled @@ -3240,9 +3261,86 @@ EXPLAIN (VERBOSE, COSTS OFF) SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1; SELECT a, count(t1) FROM pagg_tab t1 GROUP BY a HAVING avg(b) < 22 ORDER BY 1; --- When GROUP BY clause does not match with PARTITION KEY. -EXPLAIN (COSTS OFF) +-- Partial aggregates are safe to push down when there is a HAVING clause +EXPLAIN (VERBOSE, COSTS OFF) SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1; +SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b HAVING sum(a) < 700 ORDER BY 1; + +-- Partial aggregates are safe to push down without having clause +EXPLAIN (VERBOSE, COSTS OFF) +SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1; +SELECT b, avg(a), max(a), count(*) FROM pagg_tab GROUP BY b ORDER BY 1; + +-- Partial aggregates are safe to push down even if we need both variable and variable-based expression +EXPLAIN (VERBOSE, COSTS OFF) +SELECT avg(a), max(a), count(*), (b/2)::numeric FROM pagg_tab GROUP BY b/2 ORDER BY 4; +SELECT avg(a), max(a), count(*), (b/2)::numeric FROM pagg_tab GROUP BY b/2 ORDER BY 4; + +-- Partial aggregates pushdown behaves sane with non-shipped grouping elements +EXPLAIN (VERBOSE, COSTS OFF) +SELECT avg(a), (b/2)::numeric, 'test' test FROM pagg_tab GROUP BY 3, b/2 ORDER BY 1, 2 LIMIT 10; +SELECT avg(a), (b/2)::numeric, 'test' test FROM pagg_tab GROUP BY 3, b/2 ORDER BY 1, 2 LIMIT 10; + +-- Partial aggregates are safe to push down for all built-in aggregates +EXPLAIN (VERBOSE, COSTS OFF) +SELECT + /* The cases that don't require import or export functions */ + bit_and(c_bit), bit_and(c_1or3int2), bit_and(c_1or3int4), bit_and(c_1or3int8), + bit_or(c_bit), bit_or(c_1or3int2), bit_or(c_1or3int4), bit_or(c_1or3int8), + bit_xor(c_bit), bit_xor(c_1or3int2), bit_xor(c_1or3int4), bit_xor(c_1or3int8), + bool_and(c_bool), + bool_or(c_bool), + every(c_bool), + max(c_1c::char(1)), max('2000-01-01'::date + b), max('0.0.0.0'::inet + b), max(b::float4), max(b::float8), max(b::int2), max(b::int4), max(b::int8), max(c_interval), max(c_money), max(b::numeric), max(b::oid), max(c_pg_lsn), max(c_tid), max(c_1c), max(c_time), max(c_timetz), max(c_timestamp), max(c_timestamptz), max(c_1b), max(c_xid8), + min(c_1c::char(1)), min('2000-01-01'::date + b), min('0.0.0.0'::inet + b), min(b::float4), min(b::float8), min(b::int2), min(b::int4), min(b::int8), min(c_interval), min(c_money), min(b::numeric), min(b::oid), min(c_pg_lsn), min(c_tid), min(c_1c), min(c_time), min(c_timetz), min(c_timestamp), min(c_timestamptz), max(c_1b), min(c_xid8), + avg(b::int4) + FROM pagg_tab WHERE c_serial between 1 and 30; + +SELECT + /* The cases that don't require import or export functions */ + bit_and(c_bit), bit_and(c_1or3int2), bit_and(c_1or3int4), bit_and(c_1or3int8), + bit_or(c_bit), bit_or(c_1or3int2), bit_or(c_1or3int4), bit_or(c_1or3int8), + bit_xor(c_bit), bit_xor(c_1or3int2), bit_xor(c_1or3int4), bit_xor(c_1or3int8), + bool_and(c_bool), + bool_or(c_bool), + every(c_bool), + max(c_1c::char(1)), max('2000-01-01'::date + b), max('0.0.0.0'::inet + b), max(b::float4), max(b::float8), max(b::int2), max(b::int4), max(b::int8), max(c_interval), max(c_money), max(b::numeric), max(b::oid), max(c_pg_lsn), max(c_tid), max(c_1c), max(c_time), max(c_timetz), max(c_timestamp), max(c_timestamptz), max(c_1b), max(c_xid8), + min(c_1c::char(1)), min('2000-01-01'::date + b), min('0.0.0.0'::inet + b), min(b::float4), min(b::float8), min(b::int2), min(b::int4), min(b::int8), min(c_interval), min(c_money), min(b::numeric), min(b::oid), min(c_pg_lsn), min(c_tid), min(c_1c), min(c_time), min(c_timetz), min(c_timestamp), min(c_timestamptz), max(c_1b), min(c_xid8), + avg(b::int4) + FROM pagg_tab WHERE c_serial between 1 and 30; + +-- Tests for partial_aggregate_support +ALTER SERVER loopback OPTIONS (ADD partial_aggregate_support 'false'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; + +ALTER SERVER loopback OPTIONS (SET partial_aggregate_support 'true'); +EXPLAIN (VERBOSE, COSTS OFF) +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; +SELECT b, max(a) FROM pagg_tab GROUP BY b ORDER BY 1; + +ALTER SERVER loopback OPTIONS (SET fdw_tuple_cost '1.0'); +SET enable_partitionwise_join=on; +EXPLAIN (VERBOSE, COSTS OFF) +SELECT max(t1.b) FROM pagg_tab t1 JOIN pagg_tab t2 USING(a); +SELECT max(t1.b) FROM pagg_tab t1 JOIN pagg_tab t2 USING(a); +RESET enable_partitionwise_join; +ALTER SERVER loopback OPTIONS (SET fdw_tuple_cost '0.1'); + +ALTER SERVER loopback OPTIONS (DROP partial_aggregate_support); + +-- It is unsafe to push down partial aggregates which contain DISTINCT clauses +EXPLAIN (VERBOSE, COSTS OFF) +SELECT max(a), count(distinct b) FROM pagg_tab; +SELECT max(a), count(distinct b) FROM pagg_tab; + +-- It is unsafe to push down partial aggregates which contain ORDER BY clauses +EXPLAIN (VERBOSE, COSTS OFF) SELECT array_agg(b order by b) FROM pagg_tab WHERE c_serial between 1 and 30; +SELECT array_agg(b order by b) FROM pagg_tab WHERE c_serial between 1 and 30; + +ALTER SERVER loopback OPTIONS (DROP fdw_tuple_cost); +RESET TimeZone; -- =================================================================== -- access rights and superuser diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml index a7f2f5ca182..c3a6e3b1975 100644 --- a/doc/src/sgml/postgres-fdw.sgml +++ b/doc/src/sgml/postgres-fdw.sgml @@ -453,6 +453,44 @@ OPTIONS (ADD password_required 'false'); </sect3> + <sect3 id="postgres-fdw-options-partial-aggregate-pushdown"> + <title>Partial Aggregate Pushdown Option</title> + + <para> + When performing partial aggregate pushdown, + <filename>postgres_fdw</filename> gets the state value from the local + server; see <xref linkend="partial-aggregate-pushdown"/> + for partial aggregate pushdown. + By default, <filename>postgres_fdw</filename> assumes that the format + of state values on the local server is the same as the format of state + values on the remote server for every aggregate function. This may be + overridden using the following option: + </para> + + <variablelist> + + <varlistentry> + <term><literal>partial_aggregate_support</literal> (<type>boolean</type>)</term> + <listitem> + <para> + If this option is false, <filename>postgres_fdw</filename> + assumes that the format of state values on the local server is + the same as the format of state values on the remote server for + every aggregate function. If this option is true, + during query planning, <filename>postgres_fdw</filename> will + connect to the remote server and retrieve the remote server version. + If the remote version is the same, the format of state values on the + local server will be assumed to the same as the format of state + values on the remote server for every aggregate function. + If not same, <filename>postgres_fdw</filename> will not perform + partial aggregate pushdown. The default is <literal>false</literal>. + </para> + </listitem> + </varlistentry> + + </variablelist> + </sect3> + <sect3 id="postgres-fdw-options-asynchronous-execution"> <title>Asynchronous Execution Options</title> @@ -1095,14 +1133,15 @@ postgres=# SELECT postgres_fdw_disconnect_all(); <para> <filename>postgres_fdw</filename> attempts to optimize remote queries to reduce the amount of data transferred from foreign servers. This is done by - sending query <literal>WHERE</literal> clauses to the remote server for - execution, and by not retrieving table columns that are not needed for - the current query. To reduce the risk of misexecution of queries, - <literal>WHERE</literal> clauses are not sent to the remote server unless they use - only data types, operators, and functions that are built-in or belong to an - extension that's listed in the foreign server's <literal>extensions</literal> - option. Operators and functions in such clauses must - be <literal>IMMUTABLE</literal> as well. + sending query <literal>WHERE</literal> clauses and aggregate expressions + to the remote server for execution, and by not retrieving table + columns that are not needed for the current query. To reduce the + risk of misexecution of queries, <literal>WHERE</literal> clauses + and aggregate expressions are not sent to the remote server unless + they only use data types, operators, and functions that are built-in + or belong to an extension that is listed in the foreign server's + <literal>extensions</literal> option. Operators and functions in such + clauses must be <literal>IMMUTABLE</literal> as well. For an <command>UPDATE</command> or <command>DELETE</command> query, <filename>postgres_fdw</filename> attempts to optimize the query execution by sending the whole query to the remote server if there are no query @@ -1132,6 +1171,54 @@ postgres=# SELECT postgres_fdw_disconnect_all(); </para> </sect2> + <sect2 id="partial-aggregate-pushdown"> + <title>Partial Aggregate Pushdown</title> + <para> + Partial aggregate pushdown is an optimization for queries that contain + aggregate expressions for a partitioned table across one or more remote + servers. If multiple conditions are met, aggregate expressions with + <literal>PARTIAL_AGGREGATE</literal> keyword are sent to the + remote servers. The conditions under which this sending is active are + as follows. + <itemizedlist spacing="compact"> + <listitem> + <para> + The format of state values on the local server is the same as the + format of state values on the remote server. See <xref + linkend="postgres-fdw-options-partial-aggregate-pushdown"/> for the detail + </para> + </listitem> + <listitem> + <para> + The aggregate expressions in the query do not contain + <literal>DISTINCT</literal> or <literal>ORDER BY</literal> clauses + </para> + </listitem> + <listitem> + <para> + The data type of the state value is neither <type>internal</type> + nor pseudo-type. + </para> + </listitem> + </itemizedlist> + When performing partial aggregate pushdown, + <filename>postgres_fdw</filename> will not send <literal>HAVING</literal> + clause. For example, let us assume that there is one remote server + and the local server receive the following + query, where <literal>f_t</literal> is a foreign table and + <literal>c_integer</literal> is a column whose type is <literal>integer</literal>. + And let us assume that <literal>f_t</literal> corresponds + the table <literal>t</literal> on the local server. +<programlisting> +SELECT avg(c_integer) FROM f_t HAVING sum(c_integer) > 0 +</programlisting> + Then <filename>postgres_fdw</filename> send the following query to the remote servers. +<programlisting> +SELECT avg(c_integer) PARTIAL_AGGREGATE FROM t +</programlisting> + </para> + </sect2> + <sect2 id="postgres-fdw-remote-query-execution-environment"> <title>Remote Query Execution Environment</title> diff --git a/doc/src/sgml/syntax.sgml b/doc/src/sgml/syntax.sgml index 916189a7d68..967ddd3be39 100644 --- a/doc/src/sgml/syntax.sgml +++ b/doc/src/sgml/syntax.sgml @@ -1598,10 +1598,10 @@ sqrt(2) syntax of an aggregate expression is one of the following: <synopsis> -<replaceable>aggregate_name</replaceable> (<replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] -<replaceable>aggregate_name</replaceable> (ALL <replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] -<replaceable>aggregate_name</replaceable> (DISTINCT <replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] -<replaceable>aggregate_name</replaceable> ( * ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] +<replaceable>aggregate_name</replaceable> (<replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] [ PARTIAL_AGGREGATE ] +<replaceable>aggregate_name</replaceable> (ALL <replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] [ PARTIAL_AGGREGATE ] +<replaceable>aggregate_name</replaceable> (DISTINCT <replaceable>expression</replaceable> [ , ... ] [ <replaceable>order_by_clause</replaceable> ] ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] [ PARTIAL_AGGREGATE ] +<replaceable>aggregate_name</replaceable> ( * ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] [ PARTIAL_AGGREGATE ] <replaceable>aggregate_name</replaceable> ( [ <replaceable>expression</replaceable> [ , ... ] ] ) WITHIN GROUP ( <replaceable>order_by_clause</replaceable> ) [ FILTER ( WHERE <replaceable>filter_clause</replaceable> ) ] </synopsis> @@ -1779,6 +1779,25 @@ FROM generate_series(1,10) AS s(i); </programlisting> </para> + <para> + If <literal>PARTIAL_AGGREGATE</literal> is specified, then the aggregate + expression yields the value before finalization of the aggregation. + For example, the data type of the state value of <literal>avg ( integer )</literal> is <literal>bigint[]</literal>. + The state value of <literal>avg ( integer )</literal> has two elements, which are the total number + of input values and the sum of the input values. Therefore, an example of a + call of <literal>avg ( integer )</literal> with <literal>PARTIAL_AGGREGATE</literal> is: +<programlisting> +WITH vals (v) AS ( VALUES (1),(5) ) +SELECT + avg(v) PARTIAL_AGGREGATE +FROM vals; + avg +------- + {2,3} +(1 row) +</programlisting> + </para> + <para> The predefined aggregate functions are described in <xref linkend="functions-aggregate"/>. Other aggregate functions can be added diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c index b4a7698a0b3..4eed281d78b 100644 --- a/src/backend/executor/nodeAgg.c +++ b/src/backend/executor/nodeAgg.c @@ -1074,9 +1074,11 @@ finalize_aggregate(AggState *aggstate, } /* - * Apply the agg's finalfn if one is provided, else return transValue. + * Apply the agg's finalfn if one is provided and PARTIAL_AGGREGATE + * keyword is not specified, else return transValue. */ - if (OidIsValid(peragg->finalfn_oid)) + if (OidIsValid(peragg->finalfn_oid) && + (peragg->aggref->agg_partial == false)) { int numFinalArgs = peragg->numFinalArgs; diff --git a/src/backend/nodes/makefuncs.c b/src/backend/nodes/makefuncs.c index e2d9e9be41a..c17270a9457 100644 --- a/src/backend/nodes/makefuncs.c +++ b/src/backend/nodes/makefuncs.c @@ -685,6 +685,7 @@ makeFuncCall(List *name, List *args, CoercionForm funcformat, int location) n->agg_within_group = false; n->agg_star = false; n->agg_distinct = false; + n->agg_partial = false; n->func_variadic = false; n->funcformat = funcformat; n->location = location; diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c index a4d523dcb0f..9c9bcab648a 100644 --- a/src/backend/optimizer/plan/planner.c +++ b/src/backend/optimizer/plan/planner.c @@ -209,7 +209,7 @@ static PathTarget *make_group_input_target(PlannerInfo *root, PathTarget *final_target); static PathTarget *make_partial_grouping_target(PlannerInfo *root, PathTarget *grouping_target, - Node *havingQual); + GroupPathExtraData *extra); static List *postprocess_setop_tlist(List *new_tlist, List *orig_tlist); static void optimize_window_clauses(PlannerInfo *root, WindowFuncLists *wflists); @@ -5516,6 +5516,99 @@ make_group_input_target(PlannerInfo *root, PathTarget *final_target) return set_pathtarget_cost_width(root, input_target); } +/* + * setGroupClausePartial + * Generate a groupClause and a pathtarget for partial aggregate + * pushdown by FDW and set them to GroupPathExtraData. + */ +static void +setGroupClausePartial(PathTarget *partial_target, List *non_group_exprs, + List *groupClause, GroupPathExtraData *extra) +{ + int exprno, + refno; + ListCell *lc; + Index maxRef = 0; + List *exprs_processed = NIL; + int exprs_num = 0; + + foreach(lc, groupClause) + { + SortGroupClause *sgc = (SortGroupClause *) lfirst(lc); + + if (sgc->tleSortGroupRef > maxRef) + maxRef = sgc->tleSortGroupRef; + } + maxRef++; + + extra->groupClausePartial = list_copy_deep(groupClause); + extra->partial_target = copy_pathtarget(partial_target); + + if (partial_target->exprs) + exprs_num = partial_target->exprs->length; + + foreach(lc, non_group_exprs) + { + Expr *expr = (Expr *) lfirst(lc); + + refno = -1; + if (list_member(exprs_processed, expr) || + (!IsA(expr, Var) && !IsA(expr, PlaceHolderVar))) + continue; + exprs_processed = lappend(exprs_processed, expr); + for (exprno = 0; exprno < exprs_num; exprno++) + { + Expr *target_expr = (Expr *) list_nth(partial_target->exprs, exprno); + + if (equal(target_expr, expr)) + { + refno = exprno; + break; + } + } + if (refno < 0) + { + SortGroupClause *grpcl = makeNode(SortGroupClause); + + grpcl->tleSortGroupRef = maxRef++; + extra->groupClausePartial = lappend(extra->groupClausePartial, grpcl); + add_column_to_pathtarget(extra->partial_target, expr, grpcl->tleSortGroupRef); + } + } +} + +/* + * adjustAggrefForPartial + * Adjust Aggrefs to put them in partial mode + */ +static void +adjustAggrefForPartial(List *exprs) +{ + ListCell *lc; + + foreach(lc, exprs) + { + Aggref *aggref = (Aggref *) lfirst(lc); + + if (IsA(aggref, Aggref)) + { + Aggref *newaggref; + + /* + * We shouldn't need to copy the substructure of the Aggref node, + * but flat-copy the node itself to avoid damaging other trees. + */ + newaggref = makeNode(Aggref); + memcpy(newaggref, aggref, sizeof(Aggref)); + + /* For now, assume serialization is required */ + mark_partial_aggref(newaggref, AGGSPLIT_INITIAL_SERIAL); + + lfirst(lc) = newaggref; + } + } +} + /* * make_partial_grouping_target * Generate appropriate PathTarget for output of partial aggregate @@ -5531,11 +5624,15 @@ make_group_input_target(PlannerInfo *root, PathTarget *final_target) * * grouping_target is the tlist to be emitted by the topmost aggregation step. * havingQual represents the HAVING clause. + * + * Modified PathTarget cannot be used by FDW as-is to deparse this statement. + * So, before modifying PathTarget, setGroupClausePartial generates + * another Pathtarget and another list of SortGroupClauses. */ static PathTarget * make_partial_grouping_target(PlannerInfo *root, PathTarget *grouping_target, - Node *havingQual) + GroupPathExtraData *extra) { PathTarget *partial_target; List *non_group_cols; @@ -5577,8 +5674,8 @@ make_partial_grouping_target(PlannerInfo *root, /* * If there's a HAVING clause, we'll need the Vars/Aggrefs it uses, too. */ - if (havingQual) - non_group_cols = lappend(non_group_cols, havingQual); + if (extra->havingQual) + non_group_cols = lappend(non_group_cols, extra->havingQual); /* * Pull out all the Vars, PlaceHolderVars, and Aggrefs mentioned in @@ -5591,35 +5688,17 @@ make_partial_grouping_target(PlannerInfo *root, PVC_INCLUDE_AGGREGATES | PVC_RECURSE_WINDOWFUNCS | PVC_INCLUDE_PLACEHOLDERS); - + setGroupClausePartial(partial_target, non_group_exprs, root->processed_groupClause, extra); add_new_columns_to_pathtarget(partial_target, non_group_exprs); + add_new_columns_to_pathtarget(extra->partial_target, non_group_exprs); /* * Adjust Aggrefs to put them in partial mode. At this point all Aggrefs * are at the top level of the target list, so we can just scan the list * rather than recursing through the expression trees. */ - foreach(lc, partial_target->exprs) - { - Aggref *aggref = (Aggref *) lfirst(lc); - - if (IsA(aggref, Aggref)) - { - Aggref *newaggref; - - /* - * We shouldn't need to copy the substructure of the Aggref node, - * but flat-copy the node itself to avoid damaging other trees. - */ - newaggref = makeNode(Aggref); - memcpy(newaggref, aggref, sizeof(Aggref)); - - /* For now, assume serialization is required */ - mark_partial_aggref(newaggref, AGGSPLIT_INITIAL_SERIAL); - - lfirst(lc) = newaggref; - } - } + adjustAggrefForPartial(partial_target->exprs); + adjustAggrefForPartial(extra->partial_target->exprs); /* clean up cruft */ list_free(non_group_exprs); @@ -7280,7 +7359,7 @@ create_partial_grouping_paths(PlannerInfo *root, */ partially_grouped_rel->reltarget = make_partial_grouping_target(root, grouped_rel->reltarget, - extra->havingQual); + extra); if (!extra->partial_costs_set) { diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c index 999a5a8ab5a..2d6858e125c 100644 --- a/src/backend/optimizer/plan/setrefs.c +++ b/src/backend/optimizer/plan/setrefs.c @@ -2651,6 +2651,7 @@ convert_combining_aggrefs(Node *node, void *context) parent_agg = copyObject(child_agg); child_agg->args = orig_agg->args; child_agg->aggfilter = orig_agg->aggfilter; + child_agg->agg_partial = orig_agg->agg_partial; /* * Now, set up child_agg to represent the first phase of partial diff --git a/src/backend/optimizer/prep/prepagg.c b/src/backend/optimizer/prep/prepagg.c index c0a2f04a8c3..c3ffd36d6e4 100644 --- a/src/backend/optimizer/prep/prepagg.c +++ b/src/backend/optimizer/prep/prepagg.c @@ -410,6 +410,7 @@ find_compatible_agg(PlannerInfo *root, Aggref *newagg, /* all of the following must be the same or it's no match */ if (newagg->inputcollid != existingRef->inputcollid || newagg->aggtranstype != existingRef->aggtranstype || + newagg->agg_partial != existingRef->agg_partial || newagg->aggstar != existingRef->aggstar || newagg->aggvariadic != existingRef->aggvariadic || newagg->aggkind != existingRef->aggkind || diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 271ae26cbaf..11a49da779f 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -629,6 +629,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type <list> within_group_clause %type <node> filter_clause +%type <boolean> partial_aggregate_clause %type <list> window_clause window_definition_list opt_partition_clause %type <windef> window_definition over_clause window_specification opt_frame_clause frame_extent frame_bound @@ -756,7 +757,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); ORDER ORDINALITY OTHERS OUT_P OUTER_P OVER OVERLAPS OVERLAY OVERRIDING OWNED OWNER - PARALLEL PARAMETER PARSER PARTIAL PARTITION PASSING PASSWORD PATH + PARALLEL PARAMETER PARSER PARTIAL PARTIAL_AGGREGATE PARTITION PASSING PASSWORD PATH PERIOD PLACING PLAN PLANS POLICY POSITION PRECEDING PRECISION PRESERVE PREPARE PREPARED PRIMARY PRIOR PRIVILEGES PROCEDURAL PROCEDURE PROCEDURES PROGRAM PUBLICATION @@ -15758,10 +15759,11 @@ func_application: func_name '(' ')' * (Note that many of the special SQL functions wouldn't actually make any * sense as functional index entries, but we ignore that consideration here.) */ -func_expr: func_application within_group_clause filter_clause over_clause +func_expr: func_application within_group_clause filter_clause partial_aggregate_clause over_clause { FuncCall *n = (FuncCall *) $1; + n->agg_partial = $4; /* * The order clause for WITHIN GROUP and the one for * plain-aggregate ORDER BY share a field, so we have to @@ -15782,6 +15784,11 @@ func_expr: func_application within_group_clause filter_clause over_clause (errcode(ERRCODE_SYNTAX_ERROR), errmsg("cannot use DISTINCT with WITHIN GROUP"), parser_errposition(@2))); + if (n->agg_partial) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("cannot use PARTIAL_AGGREGATE with WITHIN GROUP"), + parser_errposition(@2))); if (n->func_variadic) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), @@ -15791,7 +15798,7 @@ func_expr: func_application within_group_clause filter_clause over_clause n->agg_within_group = true; } n->agg_filter = $3; - n->over = $4; + n->over = $5; $$ = (Node *) n; } | json_aggregate_func filter_clause over_clause @@ -16383,6 +16390,10 @@ filter_clause: | /*EMPTY*/ { $$ = NULL; } ; +partial_aggregate_clause: + PARTIAL_AGGREGATE { $$ = true; } + | /*EMPTY*/ { $$ = false; } + ; /* * Window Definitions @@ -17905,6 +17916,7 @@ unreserved_keyword: | PARAMETER | PARSER | PARTIAL + | PARTIAL_AGGREGATE | PARTITION | PASSING | PASSWORD diff --git a/src/backend/parser/parse_agg.c b/src/backend/parser/parse_agg.c index 0ac8966e30f..cd23eeae248 100644 --- a/src/backend/parser/parse_agg.c +++ b/src/backend/parser/parse_agg.c @@ -16,6 +16,7 @@ #include "access/htup_details.h" #include "catalog/pg_aggregate.h" +#include "catalog/pg_proc.h" #include "catalog/pg_constraint.h" #include "catalog/pg_type.h" #include "common/int.h" @@ -106,8 +107,8 @@ static Node *make_agg_arg(Oid argtype, Oid argcollation); * pstate level. */ void -transformAggregateCall(ParseState *pstate, Aggref *agg, - List *args, List *aggorder, bool agg_distinct) +transformAggregateCall(ParseState *pstate, Aggref *agg, List *args, + List *aggorder, bool agg_distinct, bool agg_partial) { List *argtypes = NIL; List *tlist = NIL; @@ -152,8 +153,8 @@ transformAggregateCall(ParseState *pstate, Aggref *agg, torder, tlist, sortby); } - /* Never any DISTINCT in an ordered-set agg */ - Assert(!agg_distinct); + /* Never any DISTINCT and PARTIAL_AGG in an ordered-set agg */ + Assert(!agg_distinct || !agg_partial); } else { @@ -227,6 +228,18 @@ transformAggregateCall(ParseState *pstate, Aggref *agg, agg->args = tlist; agg->aggorder = torder; agg->aggdistinct = tdistinct; + agg->agg_partial = agg_partial; + if(agg->agg_partial){ + HeapTuple aggtup; + Form_pg_aggregate aggform; + + aggtup = SearchSysCache1(AGGFNOID, ObjectIdGetDatum(agg->aggfnoid)); + if (!HeapTupleIsValid(aggtup)) + elog(ERROR, "cache lookup failed for aggregate %u", agg->aggfnoid); + aggform = (Form_pg_aggregate) GETSTRUCT(aggtup); + ReleaseSysCache(aggtup); + agg->aggtype = aggform->aggtranstype; + } /* * Now build the aggargtypes list with the type OIDs of the direct and diff --git a/src/backend/parser/parse_clause.c b/src/backend/parser/parse_clause.c index 2e64fcae7b2..4351846c5a8 100644 --- a/src/backend/parser/parse_clause.c +++ b/src/backend/parser/parse_clause.c @@ -539,6 +539,7 @@ transformRangeFunction(ParseState *pstate, RangeFunction *r) fc->over == NULL && !fc->agg_star && !fc->agg_distinct && + !fc->agg_partial && !fc->func_variadic && coldeflist == NIL) { diff --git a/src/backend/parser/parse_expr.c b/src/backend/parser/parse_expr.c index 9caf1e481a2..cac8ba8c080 100644 --- a/src/backend/parser/parse_expr.c +++ b/src/backend/parser/parse_expr.c @@ -3893,7 +3893,8 @@ transformJsonAggConstructor(ParseState *pstate, JsonAggConstructor *agg_ctor, aggref->aggtransno = -1; aggref->location = agg_ctor->location; - transformAggregateCall(pstate, aggref, args, agg_ctor->agg_order, false); + transformAggregateCall(pstate, aggref, args, agg_ctor->agg_order, false, + false); node = (Node *) aggref; } diff --git a/src/backend/parser/parse_func.c b/src/backend/parser/parse_func.c index 583bbbf232f..463b1dfdb33 100644 --- a/src/backend/parser/parse_func.c +++ b/src/backend/parser/parse_func.c @@ -97,6 +97,7 @@ ParseFuncOrColumn(ParseState *pstate, List *funcname, List *fargs, bool agg_within_group = (fn ? fn->agg_within_group : false); bool agg_star = (fn ? fn->agg_star : false); bool agg_distinct = (fn ? fn->agg_distinct : false); + bool agg_partial = (fn ? fn->agg_partial : false); bool func_variadic = (fn ? fn->func_variadic : false); CoercionForm funcformat = (fn ? fn->funcformat : COERCE_EXPLICIT_CALL); bool could_be_projection; @@ -222,7 +223,8 @@ ParseFuncOrColumn(ParseState *pstate, List *funcname, List *fargs, */ could_be_projection = (nargs == 1 && !proc_call && agg_order == NIL && agg_filter == NULL && - !agg_star && !agg_distinct && over == NULL && + !agg_star && !agg_distinct && + !agg_partial && over == NULL && !func_variadic && argnames == NIL && list_length(funcname) == 1 && (actual_arg_types[0] == RECORDOID || @@ -322,6 +324,12 @@ ParseFuncOrColumn(ParseState *pstate, List *funcname, List *fargs, errmsg("DISTINCT specified, but %s is not an aggregate function", NameListToString(funcname)), parser_errposition(pstate, location))); + if (agg_partial) + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("PARTIAL_AGGREGATE specified, but %s is not an aggregate function", + NameListToString(funcname)), + parser_errposition(pstate, location))); if (agg_within_group) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), @@ -392,6 +400,8 @@ ParseFuncOrColumn(ParseState *pstate, List *funcname, List *fargs, parser_errposition(pstate, location))); /* gram.y rejects DISTINCT + WITHIN GROUP */ Assert(!agg_distinct); + /* gram.y rejects PARTIAL_AGGREGATE */ + Assert(!agg_partial); /* gram.y rejects VARIADIC + WITHIN GROUP */ Assert(!func_variadic); @@ -814,7 +824,8 @@ ParseFuncOrColumn(ParseState *pstate, List *funcname, List *fargs, parser_errposition(pstate, location))); /* parse_agg.c does additional aggregate-specific processing */ - transformAggregateCall(pstate, aggref, fargs, agg_order, agg_distinct); + transformAggregateCall(pstate, aggref, fargs, agg_order, agg_distinct, + agg_partial); retval = (Node *) aggref; } @@ -846,6 +857,15 @@ ParseFuncOrColumn(ParseState *pstate, List *funcname, List *fargs, errmsg("DISTINCT is not implemented for window functions"), parser_errposition(pstate, location))); + /* + * partial aggregates not allowed in windows yet + */ + if (agg_partial) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("PARTIAL_AGGREGATE is not implemented for window functions"), + parser_errposition(pstate, location))); + /* * Reject attempt to call a parameterless aggregate without (*) * syntax. This is mere pedantry but some folks insisted ... diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c index 9e90acedb91..0d960eab8d2 100644 --- a/src/backend/utils/adt/ruleutils.c +++ b/src/backend/utils/adt/ruleutils.c @@ -10915,7 +10915,8 @@ get_agg_expr_helper(Aggref *aggref, deparse_context *context, /* Print the aggregate name, schema-qualified if needed */ appendStringInfo(buf, "%s(%s", funcname, - (aggref->aggdistinct != NIL) ? "DISTINCT " : ""); + (aggref->aggdistinct != NIL) ? "DISTINCT " : + aggref->agg_partial ? "PARTIAL_AGGREGATE " : ""); if (AGGKIND_IS_ORDERED_SET(aggref->aggkind)) { diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 23c9e3c5abf..5af495d2b90 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -442,6 +442,7 @@ typedef struct FuncCall bool agg_within_group; /* ORDER BY appeared in WITHIN GROUP */ bool agg_star; /* argument was really '*' */ bool agg_distinct; /* arguments were labeled DISTINCT */ + bool agg_partial; /* arguments were labeled PARTIAL_AGGREGATE */ bool func_variadic; /* last argument was labeled VARIADIC */ CoercionForm funcformat; /* how to display this node */ ParseLoc location; /* token location, or -1 if unknown */ diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h index fbf05322c75..03d4cdec740 100644 --- a/src/include/nodes/pathnodes.h +++ b/src/include/nodes/pathnodes.h @@ -3319,6 +3319,9 @@ typedef enum * havingQual gives list of quals to be applied after aggregation. * targetList gives list of columns to be projected. * patype is the type of partitionwise aggregation that is being performed. + * groupClausePartial is List of SortGroupClauses for partial aggregate + * pushdown by FDW + * partial_target is PathTarget for partial aggregate pushdown by FDW */ typedef struct { @@ -3333,6 +3336,8 @@ typedef struct Node *havingQual; List *targetList; PartitionwiseAggregateType patype; + List *groupClausePartial; + PathTarget *partial_target; } GroupPathExtraData; /* diff --git a/src/include/nodes/primnodes.h b/src/include/nodes/primnodes.h index d0576da3e25..79cfd69d214 100644 --- a/src/include/nodes/primnodes.h +++ b/src/include/nodes/primnodes.h @@ -490,6 +490,9 @@ typedef struct Aggref /* DISTINCT (list of SortGroupClause) */ List *aggdistinct; + /* true if there is PARTIAL_AGGREGATE keyword */ + bool agg_partial; + /* FILTER expression, if any */ Expr *aggfilter; diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h index 40cf090ce61..6d7044078ca 100644 --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -335,6 +335,7 @@ PG_KEYWORD("parallel", PARALLEL, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("parameter", PARAMETER, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("parser", PARSER, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("partial", PARTIAL, UNRESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("partial_aggregate", PARTIAL_AGGREGATE, UNRESERVED_KEYWORD, AS_LABEL) PG_KEYWORD("partition", PARTITION, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("passing", PASSING, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("password", PASSWORD, UNRESERVED_KEYWORD, BARE_LABEL) diff --git a/src/include/parser/parse_agg.h b/src/include/parser/parse_agg.h index 277d45baf83..e4ca70787c1 100644 --- a/src/include/parser/parse_agg.h +++ b/src/include/parser/parse_agg.h @@ -17,7 +17,7 @@ extern void transformAggregateCall(ParseState *pstate, Aggref *agg, List *args, List *aggorder, - bool agg_distinct); + bool agg_distinct, bool agg_partial); extern Node *transformGroupingFunc(ParseState *pstate, GroupingFunc *p); diff --git a/src/test/regress/expected/aggregates.out b/src/test/regress/expected/aggregates.out index 6b6371c3e74..97c9fb7bcc2 100644 --- a/src/test/regress/expected/aggregates.out +++ b/src/test/regress/expected/aggregates.out @@ -3536,6 +3536,14 @@ set work_mem to default; ----+----+---- (0 rows) +-- PARTIAL_AGGREGATE tests +-- Check return type of partial aggregate +select pg_typeof(avg(a::int4) partial_aggregate), pg_typeof(avg(a::int4)) from aggtest; + pg_typeof | pg_typeof +-----------+----------- + bigint[] | numeric +(1 row) + drop table agg_group_1; drop table agg_group_2; drop table agg_group_3; diff --git a/src/test/regress/expected/partition_aggregate.out b/src/test/regress/expected/partition_aggregate.out index 5f2c0cf5786..cfde38c6ed1 100644 --- a/src/test/regress/expected/partition_aggregate.out +++ b/src/test/regress/expected/partition_aggregate.out @@ -399,6 +399,35 @@ SELECT a, sum(b order by a) FROM pagg_tab GROUP BY a ORDER BY 1, 2; -> Seq Scan on pagg_tab_p3 pagg_tab_3 (10 rows) +-- PARTIAL_AGGREGATE tests +-- Check partial aggregate over partitioned table +explain (verbose, costs off) +select avg(a) partial_aggregate, avg(a) from pagg_tab; + QUERY PLAN +---------------------------------------------------------------------------------------------- + Finalize Aggregate + Output: avg(PARTIAL_AGGREGATE pagg_tab.a), avg(pagg_tab.a) + -> Append + -> Partial Aggregate + Output: PARTIAL avg(PARTIAL_AGGREGATE pagg_tab.a), PARTIAL avg(pagg_tab.a) + -> Seq Scan on public.pagg_tab_p1 pagg_tab + Output: pagg_tab.a + -> Partial Aggregate + Output: PARTIAL avg(PARTIAL_AGGREGATE pagg_tab_1.a), PARTIAL avg(pagg_tab_1.a) + -> Seq Scan on public.pagg_tab_p2 pagg_tab_1 + Output: pagg_tab_1.a + -> Partial Aggregate + Output: PARTIAL avg(PARTIAL_AGGREGATE pagg_tab_2.a), PARTIAL avg(pagg_tab_2.a) + -> Seq Scan on public.pagg_tab_p3 pagg_tab_2 + Output: pagg_tab_2.a +(15 rows) + +select avg(a) partial_aggregate, avg(a) from pagg_tab; + avg | avg +--------------+-------------------- + {3000,28500} | 9.5000000000000000 +(1 row) + -- JOIN query CREATE TABLE pagg_tab1(x int, y int) PARTITION BY RANGE(x); CREATE TABLE pagg_tab1_p1 PARTITION OF pagg_tab1 FOR VALUES FROM (0) TO (10); diff --git a/src/test/regress/sql/aggregates.sql b/src/test/regress/sql/aggregates.sql index 2c47a462b7e..2b2d4640798 100644 --- a/src/test/regress/sql/aggregates.sql +++ b/src/test/regress/sql/aggregates.sql @@ -1617,6 +1617,10 @@ set work_mem to default; union all (select * from agg_group_4 except select * from agg_hash_4); +-- PARTIAL_AGGREGATE tests +-- Check return type of partial aggregate +select pg_typeof(avg(a::int4) partial_aggregate), pg_typeof(avg(a::int4)) from aggtest; + drop table agg_group_1; drop table agg_group_2; drop table agg_group_3; diff --git a/src/test/regress/sql/partition_aggregate.sql b/src/test/regress/sql/partition_aggregate.sql index ab070fee244..60356a89f1c 100644 --- a/src/test/regress/sql/partition_aggregate.sql +++ b/src/test/regress/sql/partition_aggregate.sql @@ -92,6 +92,11 @@ SELECT c, sum(b order by a) FROM pagg_tab GROUP BY c ORDER BY 1, 2; EXPLAIN (COSTS OFF) SELECT a, sum(b order by a) FROM pagg_tab GROUP BY a ORDER BY 1, 2; +-- PARTIAL_AGGREGATE tests +-- Check partial aggregate over partitioned table +explain (verbose, costs off) +select avg(a) partial_aggregate, avg(a) from pagg_tab; +select avg(a) partial_aggregate, avg(a) from pagg_tab; -- JOIN query -- 2.39.3 ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Partial aggregates pushdown @ 2025-03-29 22:22 Bruce Momjian <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 0 replies; 8+ messages in thread From: Bruce Momjian @ 2025-03-29 22:22 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: Tomas Vondra <[email protected]>; Jelte Fennema-Nio <[email protected]>; pgsql-hackers; Robert Haas <[email protected]>; Tom Lane <[email protected]>; Stephen Frost <[email protected]>; Andres Freund <[email protected]>; Tomas Vondra <[email protected]>; Julien Rouhaud <[email protected]>; Daniel Gustafsson <[email protected]>; vignesh C <[email protected]>; Alexander Pyhalov <[email protected]> On Fri, Mar 28, 2025 at 02:00:44AM +0000, [email protected] wrote: > Hi Bruce, Jelte, hackers. > > I apologize for my late response. > > > From: Jelte Fennema-Nio <[email protected]> > > Sent: Thursday, August 8, 2024 8:49 PM > > SUMMARY OF THREAD > > > > The design of patch 0001 is agreed upon by everyone on the thread (so far). This adds the PARTIAL_AGGREGATE label for > > aggregates, which will cause the finalfunc not to run. It also starts using PARTIAL_AGGREGATE for pushdown of > > aggregates in postgres_fdw. In 0001 PARTIAL_AGGREGATE is only supported for aggregates with a non-internal/pseudo > > type as the stype. > > > > The design for patch 0002 is still under debate. This would expand on the functionality added by adding support for > > PARTIAL_AGGREGATE for aggregates with an internal stype. This is done by returning a byte array containing the bytes > > that the serialfunc of the aggregate returns. > > > > A competing proposal for 0002 is to instead change aggregates to not use an internal stype anymore, and create dedicated > > types. The main downside here is that infunc and outfunc would need to be added for text serialization, in addition to the > > binary serialization. An open question is: Can we change the requirements for CREATE TYPE, so that types can be created > > without infunc and outfunc. > > I rebased the patch 0001 and add the documentation to it. Okay, this is too late for PG 18 but I am hopeful we can make progress on this for PG 19. -- Bruce Momjian <[email protected]> https://momjian.us EDB https://enterprisedb.com Do not let urgent matters crowd out time for investment in the future. ^ permalink raw reply [nested|flat] 8+ messages in thread
end of thread, other threads:[~2025-03-29 22:22 UTC | newest] Thread overview: 8+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2024-03-14 16:39 [PATCH v10 09/17] Make table_scan_bitmap_next_block() async friendly Melanie Plageman <[email protected]> 2024-08-20 18:41 Re: Partial aggregates pushdown Bruce Momjian <[email protected]> 2024-08-21 14:59 ` Re: Partial aggregates pushdown Tomas Vondra <[email protected]> 2024-08-22 17:22 ` Re: Partial aggregates pushdown Bruce Momjian <[email protected]> 2024-08-22 18:31 ` Re: Partial aggregates pushdown Tomas Vondra <[email protected]> 2024-08-22 19:30 ` Re: Partial aggregates pushdown Bruce Momjian <[email protected]> 2025-03-28 02:00 ` RE: Partial aggregates pushdown [email protected] <[email protected]> 2025-03-29 22:22 ` Re: Partial aggregates pushdown Bruce Momjian <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox