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 <