From c73aa6357d3c70e8645fc7ea6b5981e7e84f99c2 Mon Sep 17 00:00:00 2001 From: Mario Karuza Date: Wed, 16 Sep 2026 11:35:33 +0200 Subject: [PATCH v1] Write only the needed columns to hash aggregate spill files Hash aggregates currently write spilled tuples using the full outer plan tuple descriptor, with unneeded columns set to NULL. For wide inputs, this wastes I/O and CPU on columns that the aggregate does not use. Write spilled tuples using a constructed tuple descriptor containing only the needed columns, and construct a map to reconstruct the original input tuple layout when reading them back. When all columns are needed, the spill format and path remain unchanged. When a tuple read from a spill file needs to be spilled again, write it directly without deforming and forming it again. --- src/backend/executor/nodeAgg.c | 121 +++++++++++++++++++++++++-------- src/include/nodes/execnodes.h | 6 +- 2 files changed, 95 insertions(+), 32 deletions(-) diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c index 29037cf3122..452c55bb5a2 100644 --- a/src/backend/executor/nodeAgg.c +++ b/src/backend/executor/nodeAgg.c @@ -436,8 +436,12 @@ static MinimalTuple hashagg_batch_read(HashAggBatch *batch, uint32 *hashp); static void hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset, int used_bits, double input_groups, double hashentrysize); -static Size hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill, - TupleTableSlot *inputslot, uint32 hash); +static pg_noinline Size hashagg_spill_tuple(AggState *aggstate, + HashAggSpill *spill, + TupleTableSlot *inputslot, + uint32 hash); +static Size hashagg_spill_minimal_tuple(HashAggSpill *spill, MinimalTuple tuple, + uint32 hash); static void hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno); static Datum GetAggInitVal(Datum textInitVal, Oid transtype); @@ -2742,7 +2746,7 @@ agg_refill_hash_table(AggState *aggstate) INJECTION_POINT("hash-aggregate-process-batch", NULL); for (;;) { - TupleTableSlot *spillslot = aggstate->hash_spill_rslot; + TupleTableSlot *inputslot = aggstate->hash_spill_rslot; TupleTableSlot *hashslot = perhash->hashslot; TupleHashTable hashtable = perhash->hashtable; TupleHashEntry entry; @@ -2757,8 +2761,27 @@ agg_refill_hash_table(AggState *aggstate) if (tuple == NULL) break; - ExecStoreMinimalTuple(tuple, spillslot, true); - aggstate->tmpcontext->ecxt_outertuple = spillslot; + if (aggstate->all_cols_needed) + ExecStoreMinimalTuple(tuple, inputslot, true); + else + { + TupleTableSlot *spillslot = aggstate->hash_spill_wslot; + int *colmap = aggstate->hash_spill_colmap; + int natts = spillslot->tts_tupleDescriptor->natts; + + ExecStoreMinimalTuple(tuple, spillslot, true); + slot_getallattrs(spillslot); + ExecClearTuple(inputslot); + for (int i = 0; i < natts; i++) + { + int input_colno = colmap[i]; + + inputslot->tts_values[input_colno] = spillslot->tts_values[i]; + inputslot->tts_isnull[input_colno] = spillslot->tts_isnull[i]; + } + ExecStoreVirtualTuple(inputslot); + } + aggstate->tmpcontext->ecxt_outertuple = inputslot; prepare_hash_slot(perhash, aggstate->tmpcontext->ecxt_outertuple, @@ -2785,8 +2808,8 @@ agg_refill_hash_table(AggState *aggstate) hashagg_spill_init(&spill, tapeset, batch->used_bits, batch->input_card, aggstate->hashentrysize); } - /* no memory for a new group, spill */ - hashagg_spill_tuple(aggstate, &spill, spillslot, hash); + /* no memory for a new group, spill the tuple that was read */ + hashagg_spill_minimal_tuple(&spill, tuple, hash); aggstate->hash_pergroup[batch->setno] = NULL; } @@ -3023,34 +3046,31 @@ hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset, int used_bits, * No room for new groups in the hash table. Save for later in the appropriate * partition. */ -static Size +static pg_noinline Size hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill, TupleTableSlot *inputslot, uint32 hash) { TupleTableSlot *spillslot; - int partition; MinimalTuple tuple; - LogicalTape *tape; - int total_written = 0; + Size total_written; bool shouldFree; - Assert(spill->partitions != NULL); - - /* spill only attributes that we actually need */ + /* spill only attributes that we actually need, in column order */ if (!aggstate->all_cols_needed) { + int *colmap = aggstate->hash_spill_colmap; + int natts; + spillslot = aggstate->hash_spill_wslot; + natts = spillslot->tts_tupleDescriptor->natts; slot_getsomeattrs(inputslot, aggstate->max_colno_needed); ExecClearTuple(spillslot); - for (int i = 0; i < spillslot->tts_tupleDescriptor->natts; i++) + for (int i = 0; i < natts; i++) { - if (bms_is_member(i + 1, aggstate->colnos_needed)) - { - spillslot->tts_values[i] = inputslot->tts_values[i]; - spillslot->tts_isnull[i] = inputslot->tts_isnull[i]; - } - else - spillslot->tts_isnull[i] = true; + int input_colno = colmap[i]; + + spillslot->tts_values[i] = inputslot->tts_values[input_colno]; + spillslot->tts_isnull[i] = inputslot->tts_isnull[input_colno]; } ExecStoreVirtualTuple(spillslot); } @@ -3059,6 +3079,30 @@ hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill, tuple = ExecFetchSlotMinimalTuple(spillslot, &shouldFree); + total_written = hashagg_spill_minimal_tuple(spill, tuple, hash); + + if (shouldFree) + pfree(tuple); + + return total_written; +} + +/* + * hashagg_spill_minimal_tuple + * + * Write a MinimalTuple that is already in spill format to the appropriate + * partition. + */ +static Size +hashagg_spill_minimal_tuple(HashAggSpill *spill, MinimalTuple tuple, + uint32 hash) +{ + int partition; + LogicalTape *tape; + Size total_written = 0; + + Assert(spill->partitions != NULL); + if (spill->shift < 32) partition = (hash & spill->mask) >> spill->shift; else @@ -3081,9 +3125,6 @@ hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill, LogicalTapeWrite(tape, tuple, tuple->t_len); total_written += tuple->t_len; - if (shouldFree) - pfree(tuple); - return total_written; } @@ -3685,11 +3726,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) Plan *outerplan = outerPlan(node); double totalGroups = 0; - aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate, scanDesc, - &TTSOpsMinimalTuple); - aggstate->hash_spill_wslot = ExecInitExtraTupleSlot(estate, scanDesc, - &TTSOpsVirtual); - /* this is an array of pointers, not structures */ aggstate->hash_pergroup = pergroups; @@ -3712,6 +3748,31 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) &aggstate->hash_planned_partitions); find_hash_columns(aggstate); + aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate, scanDesc, + &TTSOpsMinimalTuple); + + if (!aggstate->all_cols_needed) + { + int natts = bms_num_members(aggstate->colnos_needed); + TupleDesc spilldesc; + int spillattno = 0; + int colno = -1; + + spilldesc = CreateTemplateTupleDesc(natts); + aggstate->hash_spill_colmap = palloc_array(int, natts); + while ((colno = bms_next_member(aggstate->colnos_needed, colno)) > 0) + { + aggstate->hash_spill_colmap[spillattno] = colno - 1; + TupleDescCopyEntry(spilldesc, ++spillattno, scanDesc, colno); + } + TupleDescFinalize(spilldesc); + aggstate->hash_spill_wslot = ExecInitExtraTupleSlot(estate, spilldesc, + &TTSOpsMinimalTuple); + /* Unspilled columns are never used from and stay NULL. */ + memset(aggstate->hash_spill_rslot->tts_isnull, true, + scanDesc->natts * sizeof(bool)); + } + /* Skip massive memory allocation if we are just doing EXPLAIN */ if (!(eflags & EXEC_FLAG_EXPLAIN_ONLY)) build_hash_tables(aggstate); diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index f0cb21444b2..59dbff28412 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2476,8 +2476,10 @@ typedef struct AggState struct LogicalTapeSet *hash_tapeset; /* tape set for hash spill tapes */ struct HashAggSpill *hash_spills; /* HashAggSpill for each grouping set, * exists only during first pass */ - TupleTableSlot *hash_spill_rslot; /* for reading spill files */ - TupleTableSlot *hash_spill_wslot; /* for writing spill files */ + TupleTableSlot *hash_spill_rslot; /* input slot for spilled tuples */ + TupleTableSlot *hash_spill_wslot; /* spill slot, NULL if all_cols_needed */ + int *hash_spill_colmap; /* maps spill tuple columns to input + * columns, NULL if all_cols_needed */ List *hash_batches; /* hash batches remaining to be processed */ bool hash_ever_spilled; /* ever spilled during this execution? */ bool hash_spill_mode; /* we hit a limit during the current batch -- 2.55.0