Hi Ethan, Xiaoxuan,
Thanks for the reviews.
Attached is v2, rebased on current master (4a934833fb0):
1. Fixed the JIT crash reported by Xiaoxuan.
FIELDNO_AGGSTATE_ALL_PERGROUPS is bumped to 55. I kept
hash_spill_colmap next to the other spill fields rather
than moving it after all_pergroups.
2. Used Ethan lazy initialization.
The compact spill slot and column map are now built on the first
spill, so executions that never spill don't pay any cost.
Thanks for providing patch.
> Make them void?
hashagg_spill_tuple() already returns Size on master, and both
callers discard it there too, so I left it unchanged to keep
patch with minimal changes.
From 64ced2926fa9aea1c3d00adbf158c0f196093bfb Mon Sep 17 00:00:00 2001
From: Mario Karuza <[email protected]>
Date: Wed, 16 Sep 2026 11:35:33 +0200
Subject: [PATCH v2] 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.
The compact spill layout is built on the first spill rather than in
ExecInitAgg, so executions that never spill do not pay for it.
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 | 138 ++++++++++++++++++++++++++-------
src/include/nodes/execnodes.h | 8 +-
2 files changed, 113 insertions(+), 33 deletions(-)
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 16c0692ed5e..f8011406d00 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -436,8 +436,13 @@ 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 void hashagg_init_spill_slot(AggState *aggstate);
+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);
@@ -2755,7 +2760,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;
@@ -2770,8 +2775,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,
@@ -2798,8 +2822,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;
}
@@ -3030,40 +3054,75 @@ hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet
*tapeset, int used_bits,
initHyperLogLog(&spill->hll_card[i], HASHAGG_HLL_BIT_WIDTH);
}
+/*
+ * Initialize the compact spill slot on first use.
+ * Use the query context so it survives tuple resets and rescans.
+ */
+static void
+hashagg_init_spill_slot(AggState *aggstate)
+{
+ EState *estate = aggstate->ss.ps.state;
+ TupleDesc scanDesc =
aggstate->hash_spill_rslot->tts_tupleDescriptor;
+ MemoryContext oldcontext;
+ int natts =
bms_num_members(aggstate->colnos_needed);
+ TupleDesc spilldesc;
+ int spillattno = 0;
+ int colno = -1;
+
+ oldcontext = MemoryContextSwitchTo(estate->es_query_cxt);
+
+ 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);
+ /* Keep omitted input columns NULL. */
+ memset(aggstate->hash_spill_rslot->tts_isnull, true,
+ scanDesc->natts * sizeof(bool));
+
+ MemoryContextSwitchTo(oldcontext);
+}
+
/*
* hashagg_spill_tuple
*
* 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;
+ int natts;
+
+ if (aggstate->hash_spill_wslot == NULL)
+ hashagg_init_spill_slot(aggstate);
+
+ colmap = aggstate->hash_spill_colmap;
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);
}
@@ -3072,6 +3131,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
@@ -3094,9 +3177,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;
}
@@ -3698,11 +3778,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;
@@ -3725,6 +3800,9 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
&aggstate->hash_planned_partitions);
find_hash_columns(aggstate);
+ aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate,
scanDesc,
+
&TTSOpsMinimalTuple);
+
/* 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 8087a986204..c7afdb820be 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2449,8 +2449,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
@@ -2471,7 +2473,7 @@ typedef struct AggState
* per-group pointers */
/* support for evaluation of agg input expressions: */
-#define FIELDNO_AGGSTATE_ALL_PERGROUPS 54
+#define FIELDNO_AGGSTATE_ALL_PERGROUPS 55
AggStatePerGroup *all_pergroups; /* array of first ->pergroups,
than
* ->hash_pergroup */
SharedAggInfo *shared_info; /* one entry per worker */
--
2.56.0
