mirrors/postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HashAgg: release write buffers sooner by rewinding tape.

Browse Source 
This was an oversight. The purpose of 7fdd919ae7 was to avoid keeping
tape buffers around unnecessisarily, but HashAgg didn't rewind early
enough.

Reviewed-by: Peter Geoghegan
Discussion: https://postgr.es/m/1fb1151c2cddf8747d14e0532da283c3f97e2685.camel@j-davis.com
Backpatch-through: 13
This commit is contained in:
Jeff Davis 2020-09-15 21:16:31 -07:00
parent 873cb8fca9
commit 42a46f5a76
1 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/backend/executor/nodeAgg.c
View File

@ -2639,8 +2639,6 @@ agg_refill_hash_table(AggState *aggstate)
*/ */
hashagg_recompile_expressions(aggstate, true, true); hashagg_recompile_expressions(aggstate, true, true);
LogicalTapeRewindForRead(tapeinfo->tapeset, batch->input_tapenum,
HASHAGG_READ_BUFFER_SIZE);
for (;;) for (;;)
{ {
TupleTableSlot *spillslot = aggstate->hash_spill_rslot; TupleTableSlot *spillslot = aggstate->hash_spill_rslot;
@ -2923,6 +2921,7 @@ hashagg_tapeinfo_assign(HashTapeInfo *tapeinfo, int *partitions,
static void static void
hashagg_tapeinfo_release(HashTapeInfo *tapeinfo, int tapenum) hashagg_tapeinfo_release(HashTapeInfo *tapeinfo, int tapenum)
{ {
/* rewinding frees the buffer while not in use */
LogicalTapeRewindForWrite(tapeinfo->tapeset, tapenum); LogicalTapeRewindForWrite(tapeinfo->tapeset, tapenum);
if (tapeinfo->freetapes_alloc == tapeinfo->nfreetapes) if (tapeinfo->freetapes_alloc == tapeinfo->nfreetapes)
{ {
@ -3152,6 +3151,7 @@ hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno)
for (i = 0; i < spill->npartitions; i++) for (i = 0; i < spill->npartitions; i++)
{ {
LogicalTapeSet *tapeset = aggstate->hash_tapeinfo->tapeset;
int tapenum = spill->partitions[i]; int tapenum = spill->partitions[i];
HashAggBatch *new_batch; HashAggBatch *new_batch;
double cardinality; double cardinality;
@ -3163,9 +3163,13 @@ hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno)
cardinality = estimateHyperLogLog(&spill->hll_card[i]); cardinality = estimateHyperLogLog(&spill->hll_card[i]);
freeHyperLogLog(&spill->hll_card[i]); freeHyperLogLog(&spill->hll_card[i]);
new_batch = hashagg_batch_new(aggstate->hash_tapeinfo->tapeset, /* rewinding frees the buffer while not in use */
tapenum, setno, spill->ntuples[i], LogicalTapeRewindForRead(tapeset, tapenum,
cardinality, used_bits); HASHAGG_READ_BUFFER_SIZE);
new_batch = hashagg_batch_new(tapeset, tapenum, setno,
spill->ntuples[i], cardinality,
used_bits);
aggstate->hash_batches = lcons(new_batch, aggstate->hash_batches); aggstate->hash_batches = lcons(new_batch, aggstate->hash_batches);
aggstate->hash_batches_used++; aggstate->hash_batches_used++;
} }