From df6758fcf3d063467f22b06f15e5cd727e3d626e Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Mon, 3 Aug 2026 14:49:25 -0600 Subject: [PATCH] feat: make the grouped vectorized aggregate parallel-aware (#349) The grouped vectorized node was serial by construction -- parallel_aware false, parallel_workers 0, and no DSM callbacks on columnar_groupagg_exec_methods -- so whenever it won it replaced a four-worker plan with a single-threaded one. #350 made the *choice* honest by charging the node for the folding it does; this makes the node itself parallel, so it wins on merit rather than needing to be priced to win. It is the #343/#346 ungrouped machinery with one structural difference: the ungrouped partial node emits one tuple per worker under an AGG_PLAIN Finalize, while the grouped one emits one tuple per group per worker under an AGG_HASHED / AGGSPLIT_FINAL_DESERIAL Finalize keyed on the grouping columns. Each worker claims distinct row groups through the same gap-23 counter, builds its own hash table over them, and emits (group keys, transition states). The per-aggregate transition state is unchanged, so combine and overflow parity carry over untouched. The partial path reuses core's UPPERREL_PARTIAL_GROUP_AGG target rather than building one, so the partial and final aggregates are structurally related and setrefs matches them up. The output map is rebuilt against that target, since its column order and aggregate split both differ from output_rel's. The serial node is still offered when the parallel arm is added, unlike the ungrouped arm which drops its serial node. That asymmetry is deliberate: #133 priced the ungrouped serial node at the cheap Gather cost so keeping it would out-cost a genuinely parallel plan, while #350 gave this node an honest per-row charge. Suppressing it here costs real time -- on a ten-aggregate windowed shape the parallel arm loses to core's own plan by a hair, and with the serial node gone the result is 8,945 ms against 7,493. Offering both makes enabling the GUC a strict addition to the planner's choices. Measured, 20M-row TSBS-shaped fixture, PG18 assert, 4 workers: shape core serial parallel G1 (1 metric, 12h) 6,144 ms 4,555 ms 4,587 ms (serial chosen) G2 (10 metrics, 12h) 12,410 ms 9,225 ms 7,493 ms (serial chosen) G3 (full scan, by host) 892 ms 896 ms 600 ms (parallel chosen) G3 takes the new path and wins. G1 does not, and not because the node is slow: forced onto G1 it runs in 1,184 ms against the serial node's 4,555. The core Finalize on top of our partial node is priced off dNumGroups, and estimate_num_groups cannot estimate a date_trunc() key -- 200,000 estimated against ~8,000 actual -- so the finalize is overpriced by that ratio and loses the comparison on its own. The serial node emits finished values and pays no finalize at all, so an inflated group estimate makes it win by construction. Confirmed by G3, whose plain-column key estimates accurately and where the parallel arm is chosen. The costing asymmetry is recorded in the design note as the follow-up; it changes plan choice on shapes beyond this one and deserves its own measurement. Both GUCs remain off by default, so no default behaviour changes. test/parallel_vector_agg.sh gains eleven grouped checks. Their premises are asserted against one EXPLAIN ANALYZE and each names a property core's own parallel grouped plan does not have, because checking merely for "a Gather" or "workers launched" passes on unmodified main -- both did until tightened. Proven by removal: three premise checks fail on pristine main while the value checks pass there. Also corrects a claim rather than leaving it: the leader-side flush in the new InitializeDSM is documented as defensive, not load-bearing. Removing it and the ungrouped one it mirrors leaves both H2 checks green under in-transaction INSERT and DELETE with a confirmed parallel plan, because the buffers are already flushed at the command boundary. It is kept for symmetry and cheapness; no test covers its removal and none now claims to. Gate: full suite matrix PG18 + PG19 assert, 112 suites each; ASAN+UBSAN over the aggregate suites. The 100M bench re-measurement asked for in #349 is still outstanding -- those fixtures no longer exist on the bench host. Co-Authored-By: Claude Opus 5 (1M context) --- design/ISSUE_349_PARALLEL_GROUPAGG.md | 220 +++++++++++++ src/columnar_vector.c | 442 ++++++++++++++++++++++---- test/parallel_vector_agg.sh | 109 +++++++ 3 files changed, 717 insertions(+), 54 deletions(-) create mode 100644 design/ISSUE_349_PARALLEL_GROUPAGG.md diff --git a/design/ISSUE_349_PARALLEL_GROUPAGG.md b/design/ISSUE_349_PARALLEL_GROUPAGG.md new file mode 100644 index 0000000..21be556 --- /dev/null +++ b/design/ISSUE_349_PARALLEL_GROUPAGG.md @@ -0,0 +1,220 @@ +# #349: make the grouped vectorized aggregate path parallel-aware + +Status: design, 2026-08-03. This is the remaining scope of #349; its other items +are closed (item 2 scan-key pushdown in #354, the costing regression in #350, the +measurement phase in the issue's own comment thread). + +## Why + +`ColumnarTryGroupAggPath` adds a serial path only: + +```c +cpath->path.parallel_aware = false; +cpath->path.parallel_safe = false; +cpath->path.parallel_workers = 0; +``` + +and `columnar_groupagg_exec_methods` declares no DSM callbacks. So whenever the +grouped vectorized node wins, it **replaces a four-worker parallel plan with a +single-threaded one**. Measured on the 100M TSBS fixture (issue thread): + +| shape | groupvec off | groupvec on | +|---|---:|---:| +| G1 (1 metric, 12h window, 48k groups) | 7,139 ms | **3,604 ms** | +| G2 (10 metrics, 12h window) | 10,731 ms | **8,474 ms** | +| G3 (full 100M scan, 4k groups) | **5,953 ms** | 11,478 ms | + +G3 regressed 1.92x because the node displaced a genuinely parallel plan. #350 +fixed the *selection* (an honest per-row charge now makes G3 pick the parallel +plan) but the node itself is still serial, so on G1/G2 it wins while leaving +parallelism on the table, and `enable_group_vectorization` still cannot default on +with confidence. + +Being parallel-aware dissolves the tension: the node would be both vectorized and +parallel, so it wins on merit rather than by pricing, and nothing is displaced. + +## Shape + +Exactly the #343/#346 ungrouped arm, with one structural difference. + +- #343's ungrouped partial node emits **one** tuple per worker (`AGG_PLAIN` + Finalize). +- The grouped partial node emits **one tuple per group per worker**, and the + Finalize is `AGG_HASHED` / `AGGSPLIT_FINAL_DESERIAL` keyed on the grouping + columns. + +Plan shape: + +``` +Finalize HashAggregate (group keys, AGGSPLIT_FINAL_DESERIAL) + -> Gather + -> Partial Custom Scan (ColumnarScan) parallel_aware = true +``` + +Each worker claims distinct row groups through the shared `pg_atomic_uint32` +counter (`ColumnarReadSetParallelCounter`, gap 23 — the same atomic the base +parallel scan uses), builds **its own** group hash table over the row groups it +claimed, and emits `(group keys..., partial transition states...)`. The core +Finalize re-aggregates across workers by key. + +The per-aggregate transition state is unchanged from #343 — `columnar_agg_emit_partial` +already produces it (int8 for count/sum(int), the identity float for sum(float), +`int8[2]` `{N,sum}` for avg(int), `_float8` `{N,Sx,Sxx}` for avg(float)) — so +combine and overflow parity carry over untouched. `columnar_parallel_agg_ok` already +names exactly the eligible kinds. + +## Planner + +In the grouped branch of `ColumnarCreateUpperPaths`, alongside the serial path: + +Gate on all of: `columnar_enable_parallel_vector_agg`, `gpe->flags & +GROUPING_CAN_PARTIAL_AGG`, `gpe->partial_costs_set`, `output_rel->consider_parallel`, +`input_rel->partial_pathlist != NIL`, every aggregate `columnar_parallel_agg_ok`, +and every group key already accepted by `columnar_classify_group_keys`. + +Target: `fetch_upper_rel(root, UPPERREL_PARTIAL_GROUP_AGG, output_rel->relids)->reltarget`. +**Reuse core's target rather than building one.** It holds the grouping expressions +plus the same Aggrefs marked `AGGSPLIT_INITIAL_SERIAL` that the Finalize will +combine, so the partial and final aggregates are structurally related and setrefs +matches them up. Building our own would not match — this is the same reason the +ungrouped arm gives at `columnar_vector.c:1126-1132`, and it is the single most +likely thing to get wrong here. + +The `outMap` must therefore be rebuilt against **that** target's expression list, +not `output_rel->reltarget`, and its Aggrefs classified with `allowPartial = true` +(`columnar_classify_aggref` already accepts `AGGSPLIT_INITIAL_SERIAL` under that +flag). Everything else — group-key classification, qual extraction, the +system-column and pseudoconstant rejections — is target-independent and shared. + +Rows: `dNumGroups` per worker (each worker may see every group), not +`dNumGroups / workers`. + +Then `create_gather_path` directly on the partial path (not `add_partial_path`, +which may `pfree` a dominated path we still hold the only pointer to — same note as +the ungrouped arm), then `create_agg_path(..., AGG_HASHED, AGGSPLIT_FINAL_DESERIAL, +groupClause, groupOperators, &gpe->agg_final_costs, dNumGroups)`. + +Version note: the grouping clause list is `root->processed_groupClause` on PG16+ +and `parse->groupClause` on PG15. Gate on `PG_VERSION_NUM`. + +As in the ungrouped arm, when the parallel arm is added the serial node is **not**, +since keeping a mispriced serial node beside a genuinely parallel plan is what +produced the G3 regression in the first place. + +## Execution + +`ColumnarGroupAggScanState` gains: + +- `pg_atomic_uint32 *parallelCounter` — wired by the DSM/worker callbacks; +- `bool isPartial` — emit transition states rather than finalized values. + +`isPartial` is decided at CreateState from the output tuple, the same way the +ungrouped node does it (`columnar_vector.c:1465-1476`): a custom_scan_tlist entry +whose `Aggref->aggsplit != AGGSPLIT_SIMPLE` means partial. That also selects the +exec-methods table carrying the DSM callbacks, so no new custom_private marker is +needed. + +`columnar_groupagg_build` wires the counter onto its reader: + +```c +if (state->parallelCounter != NULL) + ColumnarReadSetParallelCounter(rs, state->parallelCounter); +``` + +`ColumnarExecGroupAggScan` emits `columnar_agg_emit_partial(&e->specs[a], &isnull)` +in place of `columnar_agg_finalize(...)` when `isPartial`. + +New DSM callbacks mirror the ungrouped four exactly; they cannot be shared verbatim +because they cast `node` to `ColumnarAggScanState`, and the grouped state is a +different struct. The leader-side flush in `InitializeDSM` matters for the same +reason it did in #343: a worker is a separate backend and cannot see the leader's +unflushed in-transaction writes and deletes. + +## Things that will bite + +1. **The group-count cap is per worker.** `pgcolumnar.groupagg_max_groups` guards + one hash table; with N workers there are N tables. That is the correct reading + (it is a per-backend memory guard) but it should be stated, and the error text + checked for it still making sense in a worker. +2. **Rescan.** `ReInitializeDSM` resets the counter to zero for all participants; + the grouped node must also drop its hash table, which `ColumnarReScanGroupAggScan` + already does for the serial case — verify it holds when partial. +3. **A worker that claims no row groups** must emit zero tuples, not one empty + group. The ungrouped node emits one all-nulls partial in that case by design; + the grouped node must emit nothing, which falls out of an empty hash table. +4. **Deterministic collation and hashing** already gate key acceptance; unchanged, + but the Finalize now hashes the same keys independently, so its equality must + agree with ours. Reusing core's target and grouping clause is what guarantees + that. + +## Measured outcome, and the one thing this does not fix + +20M-row TSBS-shaped fixture (4,000 hosts), PG18 assert, 4 workers. Not the 100M +bench: those fixtures no longer exist on the bench host, so the issue's G1-G3 +re-measurement at full scale is still outstanding. + +| shape | groupvec off (core) | groupvec on, serial | groupvec on, parallel | +|---|---:|---:|---:| +| G1 (1 metric, 12h window) | 6,144 ms | 4,555 ms | 4,587 ms *(serial chosen)* | +| G2 (10 metrics, 12h window) | 12,410 ms | 9,225 ms | 7,493 ms *(serial chosen)* | +| G3 (full scan, group by host) | 892 ms | 896 ms | **600 ms** *(parallel chosen)* | + +G3 takes the new path and wins. G1 and G2 do not take it, and the reason is +specific and worth recording, because it is not "the parallel node is slow" -- +forced onto G1 it runs in **1,184 ms against the serial node's 4,555**, a 3.9x win +the planner is declining. + +The planner's own numbers on that shape: + +``` +partialScan=15,042 ppath=16,292 gather=16,292 final=44,324 serial=20,042 +dNumGroups=200,000 (actual groups ~8,000) +``` + +Our partial node under its Gather costs 16,292 against the serial node's 20,042 -- +it wins on everything it does. The core `Finalize HashAggregate` on top adds +**28,031**, and that term is priced off `dNumGroups`. For a `date_trunc(...)` +grouping key `estimate_num_groups` cannot estimate distinctness and returns a +count near the input row count: 200,000 estimated against ~8,000 actual here, and +2,000,000 against 48,000 on the larger fixture. The finalize is therefore +overpriced by the same 25-42x, and it alone loses the comparison. + +The asymmetry is structural, not a tuning miss. The serial node emits finished +values, so it pays **no finalize at all**, and #350 deliberately priced it per +input row with no per-output-group term (an earlier version charged per group and +autoanalyze's group estimate flipped the plan choice, so the node sometimes did +not run). Any two-phase plan pays a group-count-driven finalize; the serial node +does not. When the group estimate is inflated, the serial node wins by +construction. + +That this is the mechanism rather than a guess is confirmed by G3: its grouping +key is a plain column, `estimate_num_groups` is accurate, and there the parallel +arm is chosen and is faster. Accurate estimate -> chosen; inflated estimate -> +declined. + +So the serial node is **still offered** rather than suppressed when the parallel +arm is added. Suppressing it (which is what the ungrouped arm does, for a reason +that does not carry over -- see the code comment) makes G1 take the parallel path +and win 4x, but on G2 the parallel arm loses to core's own plan by a hair and, +with the serial node gone, the result is 8,945 ms against 7,493. Offering both +makes enabling the GUC a strict addition to the planner's choices, which is the +only form in which an opt-in accelerator is safe to turn on. + +**The follow-up this needs** is the costing asymmetry, not more execution work: +either charge the serial node for the group hash table it builds, or cost the +partial arm's finalize off something less brittle than `estimate_num_groups` on an +expression key. Both change plan choice on shapes beyond this one and deserve +their own measurement, which is why they are not folded in here. + +## Gate + +Correctness before performance, since this changes results if the combine is wrong: + +- `native_groupagg`, `ungrouped_vector_agg`, `parallel_vector_agg`, `differential`, + `native_agg*` on assert PG18 + PG19, then the full matrix. +- `pg18_san` (ASAN+UBSAN) — DSM and cross-backend state. +- A new differential check: the same grouped query with the parallel arm on and off + must return identical rows, including for float aggregates where combine order + differs (compare with a tolerance for float, exact for int/count). +- Re-measure G1/G2/G3 on the bench fixture and post to #349, per the issue's own + standard that this work is planned against numbers. diff --git a/src/columnar_vector.c b/src/columnar_vector.c index f9ff491..af83908 100644 --- a/src/columnar_vector.c +++ b/src/columnar_vector.c @@ -665,6 +665,21 @@ typedef struct ColumnarGroupAggScanState bool started; /* scan + build completed */ int emitPos; /* next entry index to emit */ + /* + * Parallel grouped fold (#349). A partial node emits each group's per-worker + * transition state instead of the finalized value, and its reader claims + * distinct row groups through parallelCounter -- the same shared atomic the + * base parallel scan and the ungrouped partial node (#343) use (gap 23). Each + * worker builds its own hash table over the row groups it claimed; the core + * Finalize re-aggregates across workers by grouping key. + * + * isPartial is read from the output tuple's aggregate split at CreateState; + * parallelCounter is wired by the DSM/worker callbacks, and is NULL in a + * leader-only run, which then folds every row group itself. + */ + bool isPartial; + pg_atomic_uint32 *parallelCounter; + /* EXPLAIN */ int npreds; bool haveStats; @@ -674,8 +689,9 @@ typedef struct ColumnarGroupAggScanState } ColumnarGroupAggScanState; static const CustomExecMethods columnar_groupagg_exec_methods; +static const CustomExecMethods columnar_groupagg_parallel_exec_methods; static void ColumnarTryGroupAggPath(PlannerInfo *root, RelOptInfo *input_rel, - RelOptInfo *output_rel); + RelOptInfo *output_rel, void *extra); static bool columnar_batch_shape_eligible(ColumnarAggScanState *state, TupleDesc tupdesc, ScanKey *keysOut, int *nkeysOut); @@ -826,7 +842,7 @@ ColumnarCreateUpperPaths(PlannerInfo *root, UpperRelationKind stage, parse->distinctClause == NIL && !parse->hasWindowFuncs && !parse->hasTargetSRFs) - ColumnarTryGroupAggPath(root, input_rel, output_rel); + ColumnarTryGroupAggPath(root, input_rel, output_rel, extra); return; } @@ -1185,6 +1201,78 @@ ColumnarCreateUpperPaths(PlannerInfo *root, UpperRelationKind stage, add_path(output_rel, &cpath->path); } +/* + * columnar_groupagg_outmap + * Map one target list onto this node's output: per position, the group-key + * index (>= 0) or the aggregate index encoded as -(index + 1). False when an + * entry is neither a supported aggregate nor a bare reference to a group key, + * which forces the fallback. + * + * Built per target rather than once, because the serial path outputs + * output_rel->reltarget (finished values) while the parallel partial path + * outputs core's UPPERREL_PARTIAL_GROUP_AGG target (grouping columns plus + * AGGSPLIT_INITIAL_SERIAL aggrefs). The two differ in both the aggregate + * split and, potentially, the column order, so each needs its own map. + * allowPartial admits the INITIAL_SERIAL aggrefs of the second. + */ +static bool +columnar_groupagg_outmap(List *exprs, List *groupKeys, Index scanrelid, + bool allowPartial, List **outMapOut, int *naggsOut) +{ + List *outMap = NIL; + ListCell *lc; + int aggIdx = 0; + + foreach(lc, exprs) + { + Node *oexpr = (Node *) lfirst(lc); + + if (IsA(oexpr, Aggref)) + { + ColumnarAggSpec spec; + + if (!columnar_classify_aggref((Aggref *) oexpr, (int) scanrelid, + true, allowPartial, &spec)) + return false; + outMap = lappend(outMap, makeInteger(-(aggIdx + 1))); + aggIdx++; + } + else + { + ListCell *kc; + int k = 0; + int found = -1; + + /* + * Match the output expression against a group key exactly as written + * (no RelabelType stripping): the classifier stores keys un-stripped + * too, and both come from the same target list, so equal() lines them + * up. An output built on top of a key (not a bare reference) matches + * nothing and forces the fallback. + */ + foreach(kc, groupKeys) + { + if (equal(oexpr, (Node *) lfirst(kc))) + { + found = k; + break; + } + k++; + } + if (found < 0) + return false; + outMap = lappend(outMap, makeInteger(found)); + } + } + + if (aggIdx == 0) + return false; /* no aggregate: leave it to the ordinary plan */ + + *outMapOut = outMap; + *naggsOut = aggIdx; + return true; +} + /* * ColumnarTryGroupAggPath * Add a grouped vectorized aggregate path (#289) when the query is one we @@ -1193,10 +1281,15 @@ ColumnarCreateUpperPaths(PlannerInfo *root, UpperRelationKind stage, * supported GROUP BY key. On anything unsupported it adds nothing and the * ordinary Agg plan runs. The group-count cap is enforced only at * execution (columnar_groupagg_lookup). + * + * When the shape also qualifies for the parallel arm (#349) this adds + * Finalize HashAggregate -> Gather -> parallel-aware partial node instead of + * the serial node, so the vectorized fold runs across workers rather than + * displacing a parallel plan with a single-threaded one. */ static void ColumnarTryGroupAggPath(PlannerInfo *root, RelOptInfo *input_rel, - RelOptInfo *output_rel) + RelOptInfo *output_rel, void *extra) { RangeTblEntry *rte; Oid relid; @@ -1206,7 +1299,6 @@ ColumnarTryGroupAggPath(PlannerInfo *root, RelOptInfo *input_rel, List *groupExprs; ListCell *lc; int naggs = 0; - int aggIdx = 0; double dNumGroups; Path *cheapest; CustomPath *cpath; @@ -1242,53 +1334,10 @@ ColumnarTryGroupAggPath(PlannerInfo *root, RelOptInfo *input_rel, * Every output entry is either a supported aggregate or a bare reference to * one of the group keys. An output expression built on top of a key (a * function of a grouping column) is not handled here and forces the fallback. - * outMap records, per output position, the key index (>=0) or, encoded - * negative, the aggregate index in output order. */ - foreach(lc, output_rel->reltarget->exprs) - { - Node *oexpr = (Node *) lfirst(lc); - - if (IsA(oexpr, Aggref)) - { - ColumnarAggSpec spec; - - if (!columnar_classify_aggref((Aggref *) oexpr, - (int) input_rel->relid, true, false, &spec)) - return; - outMap = lappend(outMap, makeInteger(-(aggIdx + 1))); - aggIdx++; - naggs++; - } - else - { - ListCell *kc; - int k = 0; - int found = -1; - - /* - * Match the output expression against a group key exactly as written - * (no RelabelType stripping): the classifier stores keys un-stripped - * too, and both come from the same target list, so equal() lines them - * up. An output built on top of a key (not a bare reference) matches - * nothing and forces the fallback. - */ - foreach(kc, groupKeys) - { - if (equal(oexpr, (Node *) lfirst(kc))) - { - found = k; - break; - } - k++; - } - if (found < 0) - return; - outMap = lappend(outMap, makeInteger(found)); - } - } - if (naggs == 0) - return; /* no aggregate: leave it to the ordinary plan */ + if (!columnar_groupagg_outmap(output_rel->reltarget->exprs, groupKeys, + input_rel->relid, false, &outMap, &naggs)) + return; /* * estimate_num_groups only sizes the path's row estimate; it is deliberately @@ -1437,6 +1486,165 @@ ColumnarTryGroupAggPath(PlannerInfo *root, RelOptInfo *input_rel, outMap); cpath->methods = &columnar_agg_path_methods; + /* + * Parallel arm (#349). The serial node above folds vectors instead of + * advancing a row-wise Agg, which is cheaper per row, but it does that work on + * every row without dividing it across workers -- so whenever it won it + * replaced a four-worker plan with a single-threaded one. #350 made the + * *choice* honest; this makes the node itself parallel, so it wins on merit + * rather than needing to be priced to win. + * + * Each worker claims distinct row groups through the shared counter, builds + * its own hash table over them, and emits one (group keys, transition states) + * tuple per group. A core Finalize re-aggregates across workers by key, + * exactly as it does for an ordinary parallel grouped aggregate. + */ + if (columnar_enable_parallel_vector_agg) + { + GroupPathExtraData *gpe = (GroupPathExtraData *) extra; + bool parallelOk = (gpe != NULL && + (gpe->flags & GROUPING_CAN_PARTIAL_AGG) != 0 && + gpe->partial_costs_set && + output_rel->consider_parallel && + input_rel->partial_pathlist != NIL); + List *partialMap = NIL; + int partialAggs = 0; + + /* + * Only the kinds whose transition state is a plain, non-internal value a + * core Finalize can combine (#343). The rest keep the serial node. + */ + foreach(lc, output_rel->reltarget->exprs) + { + ColumnarAggSpec spec; + + if (!parallelOk) + break; + if (!IsA(lfirst(lc), Aggref)) + continue; + if (!columnar_classify_aggref((Aggref *) lfirst(lc), + (int) input_rel->relid, true, false, + &spec) || + !columnar_parallel_agg_ok(spec.kind)) + parallelOk = false; + } + + if (parallelOk) + { + RelOptInfo *pgr = fetch_upper_rel(root, UPPERREL_PARTIAL_GROUP_AGG, + output_rel->relids); + Path *partialScan = NULL; + ListCell *pc; + + /* the cheapest per-worker base scan drives this partial node's cost */ + foreach(pc, input_rel->partial_pathlist) + { + Path *p = (Path *) lfirst(pc); + + if (p->pathtype == T_IndexScan || p->pathtype == T_IndexOnlyScan || + p->pathtype == T_BitmapHeapScan) + continue; + if (partialScan == NULL || p->total_cost < partialScan->total_cost) + partialScan = p; + } + + /* + * pgr->reltarget is core's partial grouping target: the grouping + * columns plus the same Aggrefs marked AGGSPLIT_INITIAL_SERIAL that + * the Finalize below combines. Reuse it rather than building one, so + * the partial and final aggregates are structurally related and + * setrefs matches them up -- build our own and they would not. The + * output map has to be rebuilt against it, since its column order and + * aggregate split both differ from output_rel->reltarget. + */ + if (partialScan != NULL && partialScan->parallel_workers >= 1 && + pgr->reltarget != NULL && + columnar_groupagg_outmap(pgr->reltarget->exprs, groupKeys, + input_rel->relid, true, + &partialMap, &partialAggs)) + { + CustomPath *ppath = makeNode(CustomPath); + GatherPath *gather; + double grows = dNumGroups; + Cost pcost = partialScan->total_cost + cpu_tuple_cost + + cpu_operator_cost * partialScan->rows * + (partialAggs > 0 ? partialAggs : 1); +#if PG_VERSION_NUM >= 160000 + List *groupClause = root->processed_groupClause; +#else + List *groupClause = root->parse->groupClause; +#endif + + ppath->path.pathtype = T_CustomScan; + ppath->path.parent = pgr; + ppath->path.pathtarget = pgr->reltarget; + ppath->path.param_info = NULL; + ppath->path.parallel_aware = true; + ppath->path.parallel_safe = true; + ppath->path.parallel_workers = partialScan->parallel_workers; + + /* + * Every worker may see every group, so each emits up to the whole + * group count -- not the count divided by the worker count, which + * is what an ordinary partial aggregate would estimate. + */ + ppath->path.rows = (dNumGroups < 1.0) ? 1.0 : dNumGroups; + ppath->path.startup_cost = pcost; + ppath->path.total_cost = pcost; + ppath->path.pathkeys = NIL; + ppath->flags = 0; + ppath->custom_paths = NIL; +#if PG_VERSION_NUM >= 170000 + ppath->custom_restrictinfo = NIL; +#endif + ppath->custom_private = + list_make5(makeInteger((int) input_rel->relid), + copyObject(quals), + makeConst(OIDOID, -1, InvalidOid, sizeof(Oid), + ObjectIdGetDatum(relid), false, true), + groupKeys, + partialMap); + ppath->methods = &columnar_agg_path_methods; + + /* + * Gather this partial node directly rather than add_partial_path'ing + * it: add_partial_path may pfree a dominated path, and we hold the + * only pointer create_gather_path needs. + */ + gather = create_gather_path(root, pgr, &ppath->path, + pgr->reltarget, NULL, &grows); + + add_path(output_rel, + (Path *) create_agg_path(root, output_rel, &gather->path, + output_rel->reltarget, + AGG_HASHED, + AGGSPLIT_FINAL_DESERIAL, + groupClause, NIL, + &gpe->agg_final_costs, + dNumGroups)); + } + } + } + + /* + * Offer the serial node too, even when the parallel arm was added, and let + * the planner choose between them. + * + * The ungrouped arm (#343) deliberately does the opposite -- it drops its + * serial node once the parallel one exists -- because #133 priced that node at + * the cheap Gather cost, so keeping it would wrongly out-cost a genuinely + * parallel plan. That reasoning does not carry over: #350 gave this node an + * honest per-row charge, so it competes on merit and cannot win by being + * underpriced. + * + * Suppressing it here actively costs. On a ten-aggregate windowed shape the + * parallel arm is charged for ten aggregates per row and loses to core's own + * parallel plan by a hair, so dropping the serial node left neither: 7,583 ms + * with this feature off against 8,944 ms with it on, purely from the missing + * path. Offering both makes turning the GUC on a strict addition to the + * planner's choices, which is the only form in which an opt-in accelerator can + * be safe to enable. + */ add_path(output_rel, &cpath->path); } @@ -3082,7 +3290,27 @@ ColumnarCreateGroupAggScanState(CustomScan *cscan) int naggs = 0; state->css.ss.ps.type = T_CustomScanState; - state->css.methods = &columnar_groupagg_exec_methods; + + /* + * A parallel partial grouped node (#349) is planned with + * AGGSPLIT_INITIAL_SERIAL aggrefs in its output tuple: it emits one + * (group keys, transition states) tuple per group per worker for a core + * Finalize to combine by key. Detect it from the output tuple exactly as the + * ungrouped node does, and switch to the exec methods table carrying the + * DSM/worker callbacks so every worker shares the group-claim counter. + */ + state->isPartial = false; + foreach(lc, cscan->custom_scan_tlist) + { + TargetEntry *tle = (TargetEntry *) lfirst(lc); + + if (IsA(tle->expr, Aggref) && + ((Aggref *) tle->expr)->aggsplit != AGGSPLIT_SIMPLE) + state->isPartial = true; + } + state->css.methods = state->isPartial + ? &columnar_groupagg_parallel_exec_methods + : &columnar_groupagg_exec_methods; /* custom_private: rti, quals, relid, group-key exprs, output map (length 5) */ state->scanrelid = (Index) intVal(linitial(cscan->custom_private)); @@ -3113,7 +3341,8 @@ ColumnarCreateGroupAggScanState(CustomScan *cscan) if (IsA(tle->expr, Aggref)) { - (void) columnar_classify_aggref((Aggref *) tle->expr, -1, true, false, + /* allowPartial accepts the parallel arm's INITIAL_SERIAL aggrefs */ + (void) columnar_classify_aggref((Aggref *) tle->expr, -1, true, true, &state->aggTemplate[i]); i++; } @@ -3450,6 +3679,15 @@ columnar_groupagg_build(ColumnarGroupAggScanState *state) rs = ColumnarBeginRead(rel, estate->es_snapshot, NULL, state->projected, nScanKeys, keys); + /* + * In a parallel partial run each participant folds only the row groups it + * claims from the shared counter, so the union across workers is every group + * exactly once (#349). NULL when this node is running leader-only, in which + * case the reader walks every group as before. + */ + if (state->parallelCounter != NULL) + ColumnarReadSetParallelCounter(rs, state->parallelCounter); + /* columns outside the projection stay null in the base slot */ memset(state->baseSlot->tts_isnull, true, sizeof(bool) * natts); @@ -3547,8 +3785,19 @@ ColumnarExecGroupAggScan(CustomScanState *node) { int a = -(m) - 1; - scanSlot->tts_values[p] = - columnar_agg_finalize(&e->specs[a], &scanSlot->tts_isnull[p]); + /* + * A partial node hands the core Finalize this worker's transition + * state for the group rather than the finished value; the Finalize + * combines the states of every worker that saw the same key + * (#349). A worker that claimed no row groups has an empty hash + * table and so emits nothing at all, which is what the Finalize + * expects -- not one empty group. + */ + scanSlot->tts_values[p] = state->isPartial + ? columnar_agg_emit_partial(&e->specs[a], + &scanSlot->tts_isnull[p]) + : columnar_agg_finalize(&e->specs[a], + &scanSlot->tts_isnull[p]); } } ExecStoreVirtualTuple(scanSlot); @@ -3624,6 +3873,91 @@ static const CustomExecMethods columnar_groupagg_exec_methods = { .ExplainCustomScan = ColumnarExplainGroupAggScan, }; +/* ------------------------------------------------------------------------- + * parallel partial grouped aggregate (#349): DSM callbacks + * + * The same shared pg_atomic_uint32 the ungrouped partial node uses (gap 23), + * handing out row-group indices so each worker folds distinct groups. These + * mirror the ungrouped four and cannot share their bodies: those cast the node + * to ColumnarAggScanState, and the grouped node is a different struct. + * + * The grouped node opens its reader lazily in Exec, strictly after both DSM-init + * and Worker-init, so the callbacks need only record the counter on the state. + * ------------------------------------------------------------------------- */ + +static Size +ColumnarEstimateDSMGroupAggScan(CustomScanState *node, ParallelContext *pcxt) +{ + return sizeof(pg_atomic_uint32); +} + +static void +ColumnarInitializeDSMGroupAggScan(CustomScanState *node, ParallelContext *pcxt, + void *coordinate) +{ + ColumnarGroupAggScanState *state = (ColumnarGroupAggScanState *) node; + pg_atomic_uint32 *counter = (pg_atomic_uint32 *) coordinate; + + pg_atomic_init_u32(counter, 0); + state->parallelCounter = counter; + + /* + * Flush this backend's pending writes and deletes in the leader, before any + * worker launches: a worker is a separate backend and cannot see the leader's + * unflushed in-transaction buffers. Mirrors what #343 does for the ungrouped + * node. + * + * Defensive rather than load-bearing today, which is worth stating precisely + * because the comment on the ungrouped copy reads as though it were required. + * Removing both flushes and running in-transaction INSERT and DELETE followed + * by a confirmed-parallel grouped aggregate produces answers identical to + * serial, because the buffers are already flushed at the command boundary + * before the aggregate is planned. It is kept because it is cheap and because + * a future path that reaches here with unflushed state would silently give + * workers a stale view; no test covers its removal, and none claims to. + */ + ColumnarFlushWriteStateForRelation(state->relid); + { + Relation frel = table_open(state->relid, AccessShareLock); + + ColumnarFlushDeleteVectorForRelation(frel); + table_close(frel, AccessShareLock); + } +} + +static void +ColumnarReInitializeDSMGroupAggScan(CustomScanState *node, ParallelContext *pcxt, + void *coordinate) +{ + pg_atomic_uint32 *counter = (pg_atomic_uint32 *) coordinate; + + /* a rescan restarts group claiming from zero for every participant */ + pg_atomic_write_u32(counter, 0); +} + +static void +ColumnarInitializeWorkerGroupAggScan(CustomScanState *node, shm_toc *toc, + void *coordinate) +{ + ColumnarGroupAggScanState *state = (ColumnarGroupAggScanState *) node; + pg_atomic_uint32 *counter = (pg_atomic_uint32 *) coordinate; + + state->parallelCounter = counter; +} + +static const CustomExecMethods columnar_groupagg_parallel_exec_methods = { + .CustomName = "ColumnarScan", + .BeginCustomScan = ColumnarBeginGroupAggScan, + .ExecCustomScan = ColumnarExecGroupAggScan, + .EndCustomScan = ColumnarEndGroupAggScan, + .ReScanCustomScan = ColumnarReScanGroupAggScan, + .ExplainCustomScan = ColumnarExplainGroupAggScan, + .EstimateDSMCustomScan = ColumnarEstimateDSMGroupAggScan, + .InitializeDSMCustomScan = ColumnarInitializeDSMGroupAggScan, + .ReInitializeDSMCustomScan = ColumnarReInitializeDSMGroupAggScan, + .InitializeWorkerCustomScan = ColumnarInitializeWorkerGroupAggScan, +}; + /* ------------------------------------------------------------------------- * registration * ------------------------------------------------------------------------- */ diff --git a/test/parallel_vector_agg.sh b/test/parallel_vector_agg.sh index d4f741b..14cecd5 100644 --- a/test/parallel_vector_agg.sh +++ b/test/parallel_vector_agg.sh @@ -130,5 +130,114 @@ H2_PAR="$(printf '%s\n' "$H2" | grep -E '^[0-9]+\|' | sed -n 1p)" H2_SER="$(printf '%s\n' "$H2" | grep -E '^[0-9]+\|' | sed -n 2p)" check "in-xact deletes (H2): parallel fold == serial" "$H2_PAR" "$H2_SER" +# ---- #349: the GROUPED vectorized fold is parallel-aware too ---------------- +# The grouped node used to be serial by construction (parallel_aware = false, no +# DSM callbacks), so whenever it won it replaced a four-worker plan with a +# single-threaded one -- a 1.92x regression on a full-scan GROUP BY with few +# groups. It now plans Finalize HashAggregate -> Gather -> parallel-aware partial +# grouped node: each worker claims distinct row groups through the same gap-23 +# counter, builds its OWN hash table over them, and emits one (group keys, +# transition states) tuple per group for the core Finalize to combine by key. +GVP="SET pgcolumnar.enable_group_vectorization=on; + SET pgcolumnar.enable_parallel_vector_agg=on;" + +q -c "DROP TABLE IF EXISTS gt; + CREATE TABLE gt (id int, h text, k int, v float8, w int) USING pgcolumnar; + SELECT pgcolumnar.set_options('gt'::regclass, stripe_row_limit => 20000); + INSERT INTO gt + SELECT g, 'h' || (g % 50), g % 200, + CASE WHEN g % 50 = 0 THEN NULL + ELSE ((g % 13) - 6)::float8 * (10.0 ^ (g % 4)) END, + g % 13 + FROM generate_series(1, 800000) g; + ANALYZE gt;" >/dev/null + +# Assert the fixture before comparing anything: two arms that both error return +# empty and compare equal, which is a green check that tested nothing. +check "premise: the grouped fixture has its rows" \ + "$(q -c 'SELECT count(*) FROM gt')" 800000 + +# Every assertion here is made against ONE EXPLAIN ANALYZE, and each names a +# property core's own parallel grouped plan does NOT have. That distinction is +# the whole point: without the change core still plans +# "Finalize GroupAggregate -> Gather Merge -> Sort -> Partial HashAggregate" over +# the same table, which has a Gather and launches workers too. Checking only for +# "a Gather" or "workers launched" therefore passes on unmodified main and proves +# nothing -- both did, until this was tightened. +GEA="$(q -c "$PAR $GVP" -c "EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF, SUMMARY OFF) + SELECT h, count(*), sum(w) FROM gt GROUP BY h")" + +# A HashAggregate finalize, not core's GroupAggregate-over-Sort. +check "premise: grouped Finalize HashAggregate present (#349)" \ + "$(printf '%s' "$GEA" | grep -qi 'Finalize HashAggregate' && echo y || echo n)" y +# OUR grouped node, and it is the parallel-aware one. +check "premise: the partial grouped node is under the Gather (#349)" \ + "$(printf '%s' "$GEA" | grep -qi 'Parallel Custom Scan (ColumnarScan)' && + printf '%s' "$GEA" | grep -qi 'Columnar Vectorized Group Keys' && echo y || echo n)" y +# Planned is not launched: a leader-only run would satisfy every value check +# below while never exercising a worker. Assert launch AND our node together, so +# core's parallel plan cannot satisfy this on its own. +check "premise: workers actually launched for OUR grouped node (#349)" \ + "$(printf '%s' "$GEA" | grep -qiE 'Workers Launched: [1-9]' && + printf '%s' "$GEA" | grep -qi 'Columnar Vectorized Group Keys' && echo y || echo n)" y + +# Each participant emits its own partial per group, so the Gather carries a +# MULTIPLE of the group count and the Finalize collapses it back. 50 groups with +# workers launched means the Gather must show strictly more than 50 rows; core's +# partial-HashAggregate plan shows the same shape, so this is paired with the +# node checks above rather than standing alone. +GATHER_ROWS="$(printf '%s\n' "$GEA" | grep -iE '^ *-> *Gather' | grep -oE 'rows=[0-9.]+' | head -1 | cut -d= -f2 | cut -d. -f1)" +check "the Gather carries per-worker partials, more than one row per group (#349)" \ + "$( [ -n "$GATHER_ROWS" ] && [ "$GATHER_ROWS" -gt 50 ] && echo y || + echo "n (gather rows=${GATHER_ROWS:-unset})")" y +check "the Finalize collapses them back to exactly the group count (#349)" \ + "$(q -c "$PAR $GVP" -c "SELECT count(*) FROM (SELECT h, count(*), sum(w) FROM gt GROUP BY h) s")" 50 + +# ---- values: integer aggregates are exact against a serial oracle ----------- +G_VEC="$(q -c "$PAR $GVP" -c "SELECT h, count(*), sum(w) FROM gt WHERE k < 150 GROUP BY h ORDER BY h")" +G_SER="$(q -c "SET max_parallel_workers_per_gather=0;" -c "SELECT h, count(*), sum(w) FROM gt WHERE k < 150 GROUP BY h ORDER BY h")" +check "grouped count+sum(int): parallel fold == serial oracle (#349)" "$G_VEC" "$G_SER" + +GM_VEC="$(q -c "$PAR $GVP" -c "SELECT h, k, count(*), sum(w) FROM gt WHERE v IS NOT NULL GROUP BY h, k ORDER BY h, k")" +GM_SER="$(q -c "SET max_parallel_workers_per_gather=0;" -c "SELECT h, k, count(*), sum(w) FROM gt WHERE v IS NOT NULL GROUP BY h, k ORDER BY h, k")" +check "grouped multi-key + WHERE: parallel fold == serial oracle (#349)" "$GM_VEC" "$GM_SER" + +# ---- float: oracle is core's own PARALLEL agg, for the reason above --------- +GF_VEC="$(q -c "$PAR $GVP" -c "SELECT h, round(avg(v)::numeric,6), round(sum(v)::numeric,6) FROM gt GROUP BY h ORDER BY h")" +GF_PAR="$(q -c "$PAR" -c "SELECT h, round(avg(v)::numeric,6), round(sum(v)::numeric,6) FROM gt GROUP BY h ORDER BY h")" +check "grouped avg/sum(float8): parallel fold == core parallel agg (#349)" "$GF_VEC" "$GF_PAR" + +# ---- a group count below the worker count: some workers see every group ----- +GW_VEC="$(q -c "$PAR $GVP SET max_parallel_workers_per_gather=8;" -c "SELECT k % 3 AS g, count(*), sum(w) FROM gt GROUP BY 1 ORDER BY 1")" +GW_SER="$(q -c "SET max_parallel_workers_per_gather=0;" -c "SELECT k % 3 AS g, count(*), sum(w) FROM gt GROUP BY 1 ORDER BY 1")" +check "grouped few-groups-many-workers: parallel == serial (#349)" "$GW_VEC" "$GW_SER" + +# ---- H2 for the grouped node: in-transaction deletes ------------------------- +# The grouped node runs the same shape as the ungrouped H2 case above: delete +# inside a transaction, then aggregate in parallel in that same transaction, and +# require the parallel answer to match the serial one. +# +# What this does NOT establish, measured rather than assumed: it does not prove +# the leader-side flush in ColumnarInitializeDSMGroupAggScan. Removing that flush +# -- and the ungrouped one this is modelled on -- leaves both H2 checks green, +# with in-transaction INSERT and DELETE and a confirmed parallel plan, because +# the write and delete buffers are already flushed at the command boundary before +# the aggregate is planned. The flush is kept as defence and for symmetry with +# the ungrouped node, not because anything here fails without it. Stated so the +# next person does not read a passing H2 check as cover for that flush. +GH2="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -Atq 2>&1 <