Date: Wed, 8 Jan 2020 15:12:02 +0800
From: Adam Lee <ali@pivotal.io>
To: Jeff Davis <pgsql@j-davis.com>
Cc: Tomas Vondra <tomas.vondra@2ndquadrant.com>,
        Taylor Vesely <tvesely@pivotal.io>,
        Melanie Plageman <mplageman@pivotal.io>, pgsql-hackers@postgresql.org
Subject: Re: Memory-Bounded Hash Aggregation
Message-ID: <20200108071202.GA1511@mars.local>
References: <617fea8258ba3df7bc8b21cd7f3afc219f2383d9.camel@j-davis.com>
 <20190802152123.ichmacgx3y3x7dc3@development>
 <CAFaX_4Ls5UGQ2UDYkWhn2xFq=V6BKpUbKSQnWmy27VWhQ2=enA@mail.gmail.com>
 <e52707d4665e32867fb6dd181825ef15f853b8bf.camel@j-davis.com>
 <20191128174644.snpuqldt5rndjru7@development>
 <2c8dc5c9178c391493dfbf6c2f595dc298dd477e.camel@j-davis.com>
 <20191205035013.GB1345@mars.local>
 <1b5c79a14f85e20899b100381c1e8d41e3148c5b.camel@j-davis.com>
 <20191210213422.GD1345@mars.local>
 <2370ca86d4c3b46befc031d8421de02a58df20f1.camel@j-davis.com>
MIME-Version: 1.0
Content-Type: multipart/mixed; boundary="zYM0uCDKw75PZbzx"
Content-Disposition: inline
In-Reply-To: <2370ca86d4c3b46befc031d8421de02a58df20f1.camel@j-davis.com>
Precedence: bulk


--zYM0uCDKw75PZbzx
Content-Type: text/plain; charset=us-ascii
Content-Disposition: inline

Hi, Jeff

I tried to use the logical tape APIs for hash agg spilling, based on
your 1220 version.

Turns out it doesn't make much of performance difference with the
default 8K block size (might be my patch's problem), but the disk space
(not I/O) would be saved a lot because I force the respilling to use the
same LogicalTapeSet.

Logtape APIs with default block size 8K:
```
postgres=# EXPLAIN ANALYZE SELECT avg(g) FROM generate_series(0,5000000) g GROUP BY g;
                                                                 QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------
 HashAggregate  (cost=75000.02..75002.52 rows=200 width=36) (actual time=7701.706..24473.002 rows=5000001 loops=1)
   Group Key: g
   Memory Usage: 4096kB  Batches: 516  Disk: 116921kB
   ->  Function Scan on generate_series g  (cost=0.00..50000.01 rows=5000001 width=4) (actual time=1611.829..3253.150 rows=5000001 loops=1)
 Planning Time: 0.194 ms
 Execution Time: 25129.239 ms
(6 rows)
```

Bare BufFile APIs:
```
postgres=# EXPLAIN ANALYZE SELECT avg(g) FROM generate_series(0,5000000) g GROUP BY g;
                                                                 QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------
 HashAggregate  (cost=75000.02..75002.52 rows=200 width=36) (actual time=7339.835..24472.466 rows=5000001 loops=1)
   Group Key: g
   Memory Usage: 4096kB  Batches: 516  Disk: 232773kB
   ->  Function Scan on generate_series g  (cost=0.00..50000.01 rows=5000001 width=4) (actual time=1580.057..3128.749 rows=5000001 loops=1)
 Planning Time: 0.769 ms
 Execution Time: 26696.502 ms
(6 rows)
```

Even though, I'm not sure which API is better, because we should avoid
the respilling as much as we could in the planner, and hash join uses
the bare BufFile.

Attached my hacky and probably not robust diff for your reference.

-- 
Adam Lee

--zYM0uCDKw75PZbzx
Content-Type: text/plain; charset=us-ascii
Content-Disposition: attachment; filename="hashagg_logtape.diff"

diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index f1989b10ea..8c743d7561 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -247,6 +247,7 @@
 #include "utils/datum.h"
 #include "utils/dynahash.h"
 #include "utils/expandeddatum.h"
+#include "utils/logtape.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/syscache.h"
@@ -288,8 +289,9 @@ typedef struct HashAggSpill
 	int       n_partitions;		/* number of output partitions */
 	int       partition_bits;	/* number of bits for partition mask
 								   log2(n_partitions) parent partition bits */
-	BufFile **partitions;		/* output partition files */
+	int      *partitions;		/* output logtape numbers */
 	int64    *ntuples;			/* number of tuples in each partition */
+	LogicalTapeSet *lts;
 } HashAggSpill;
 
 /*
@@ -298,11 +300,12 @@ typedef struct HashAggSpill
  */
 typedef struct HashAggBatch
 {
-	BufFile *input_file;		/* input partition */
+	int      input_tape;		/* input partition */
 	int      input_bits;		/* number of bits for input partition mask */
 	int64    input_tuples;		/* number of tuples in this batch */
 	int		 setno;				/* grouping set */
 	HashAggSpill spill;			/* spill output */
+	LogicalTapeSet *lts;
 } HashAggBatch;
 
 static void select_current_set(AggState *aggstate, int setno, bool is_hash);
@@ -359,9 +362,8 @@ static void hash_spill_init(HashAggSpill *spill, int input_bits,
 							uint64 input_tuples, double hashentrysize);
 static Size hash_spill_tuple(HashAggSpill *spill, int input_bits,
 							 TupleTableSlot *slot, uint32 hash);
-static MinimalTuple hash_read_spilled(BufFile *file, uint32 *hashp);
-static HashAggBatch *hash_batch_new(BufFile *input_file, int setno,
-									int64 input_tuples, int input_bits);
+static MinimalTuple hash_read_spilled(LogicalTapeSet *lts, int tapenum, uint32 *hashp);
+static HashAggBatch *hash_batch_new(LogicalTapeSet *lts, int tapenum, int setno, int64 input_tuples, int input_bits);
 static void hash_finish_initial_spills(AggState *aggstate);
 static void hash_spill_finish(AggState *aggstate, HashAggSpill *spill,
 							  int setno, int input_bits);
@@ -2462,7 +2464,7 @@ agg_refill_hash_table(AggState *aggstate)
 
 		CHECK_FOR_INTERRUPTS();
 
-		tuple = hash_read_spilled(batch->input_file, &hash);
+		tuple = hash_read_spilled(batch->lts, batch->input_tape, &hash);
 		if (tuple == NULL)
 			break;
 
@@ -2490,8 +2492,8 @@ agg_refill_hash_table(AggState *aggstate)
 								batch->input_tuples, aggstate->hashentrysize);
 			}
 
-			aggstate->hash_disk_used += hash_spill_tuple(
-				&batch->spill, batch->input_bits, slot, hash);
+			//aggstate->hash_disk_used +=
+			hash_spill_tuple(&batch->spill, batch->input_bits, slot, hash);
 		}
 
 		/* Advance the aggregates (or combine functions) */
@@ -2504,8 +2506,6 @@ agg_refill_hash_table(AggState *aggstate)
 		ResetExprContext(aggstate->tmpcontext);
 	}
 
-	BufFileClose(batch->input_file);
-
 	aggstate->current_phase = 0;
 	aggstate->phase = &aggstate->phases[aggstate->current_phase];
 
@@ -2690,6 +2690,9 @@ hash_spill_init(HashAggSpill *spill, int input_bits, uint64 input_groups,
 {
 	int     npartitions;
 	int     partition_bits;
+	int     i;
+	int     j;
+	int     old_npartitions;
 
 	npartitions = hash_choose_num_spill_partitions(input_groups,
 												   hashentrysize);
@@ -2702,10 +2705,33 @@ hash_spill_init(HashAggSpill *spill, int input_bits, uint64 input_groups,
 	/* number of partitions will be a power of two */
 	npartitions = 1L << partition_bits;
 
-	spill->partition_bits = partition_bits;
-	spill->n_partitions = npartitions;
-	spill->partitions = palloc0(sizeof(BufFile *) * npartitions);
-	spill->ntuples = palloc0(sizeof(int64) * npartitions);
+	if (spill->lts == NULL)
+	{
+		spill->partition_bits = partition_bits;
+		spill->n_partitions   = npartitions;
+		spill->partitions     = palloc0(sizeof(int) * npartitions);
+		for (i = 0; i < spill->n_partitions; ++i)
+		{
+			spill->partitions[i] = i;
+		}
+		spill->ntuples        = palloc0(sizeof(int64) * spill->n_partitions);
+		spill->lts            = LogicalTapeSetCreate(npartitions, NULL, NULL, 0); // TODO: worker is 0?
+	}
+	else // respill
+	{
+		old_npartitions = LogicalTapeGetNTapes(spill->lts);
+		spill->partition_bits = my_log2(npartitions);
+		spill->n_partitions   = (1L << spill->partition_bits);
+		spill->partitions     = palloc0(sizeof(int) * npartitions);
+		j = old_npartitions;
+		for (i = 0; i < spill->n_partitions; ++i)
+		{
+			spill->partitions[i] = j;
+			j++;
+		}
+		spill->ntuples        = palloc0(sizeof(int64) * spill->n_partitions);
+		spill->lts            = LogicalTapeSetExtend(spill->lts, spill->n_partitions);
+	}
 }
 
 /*
@@ -2720,8 +2746,6 @@ hash_spill_tuple(HashAggSpill *spill, int input_bits, TupleTableSlot *slot,
 {
 	int					 partition;
 	MinimalTuple		 tuple;
-	BufFile				*file;
-	int					 written;
 	int					 total_written = 0;
 	bool				 shouldFree;
 
@@ -2743,23 +2767,11 @@ hash_spill_tuple(HashAggSpill *spill, int input_bits, TupleTableSlot *slot,
 
 	spill->ntuples[partition]++;
 
-	if (spill->partitions[partition] == NULL)
-		spill->partitions[partition] = BufFileCreateTemp(false);
-	file = spill->partitions[partition];
-
-	written = BufFileWrite(file, (void *) &hash, sizeof(uint32));
-	if (written != sizeof(uint32))
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not write to HashAgg temporary file: %m")));
-	total_written += written;
+	LogicalTapeWrite(spill->lts, spill->partitions[partition], (void *) &hash, sizeof(uint32));
+	total_written += sizeof(uint32);
 
-	written = BufFileWrite(file, (void *) tuple, tuple->t_len);
-	if (written != tuple->t_len)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not write to HashAgg temporary file: %m")));
-	total_written += written;
+	LogicalTapeWrite(spill->lts, spill->partitions[partition], (void *) tuple, tuple->t_len);
+	total_written += tuple->t_len;
 
 	if (shouldFree)
 		pfree(tuple);
@@ -2772,38 +2784,37 @@ hash_spill_tuple(HashAggSpill *spill, int input_bits, TupleTableSlot *slot,
  * 		read the next tuple from a batch file.  Return NULL if no more.
  */
 static MinimalTuple
-hash_read_spilled(BufFile *file, uint32 *hashp)
+hash_read_spilled(LogicalTapeSet *lts, int tapenum, uint32 *hashp)
 {
 	MinimalTuple	tuple;
 	uint32			t_len;
 	size_t			nread;
 	uint32			hash;
 
-	nread = BufFileRead(file, &hash, sizeof(uint32));
+	nread = LogicalTapeRead(lts, tapenum, &hash, sizeof(uint32));
 	if (nread == 0)
 		return NULL;
 	if (nread != sizeof(uint32))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not read from HashAgg temporary file: %m")));
+				 errmsg("could not read the hash from HashAgg spilled tape: %m")));
 	if (hashp != NULL)
 		*hashp = hash;
 
-	nread = BufFileRead(file, &t_len, sizeof(t_len));
+	nread = LogicalTapeRead(lts, tapenum, &t_len, sizeof(t_len));
 	if (nread != sizeof(uint32))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not read from HashAgg temporary file: %m")));
+				 errmsg("could not read the t_len from HashAgg spilled tape: %m")));
 
 	tuple = (MinimalTuple) palloc(t_len);
 	tuple->t_len = t_len;
 
-	nread = BufFileRead(file, (void *)((char *)tuple + sizeof(uint32)),
-						t_len - sizeof(uint32));
+	nread = LogicalTapeRead(lts, tapenum, (void *)((char *)tuple + sizeof(uint32)), t_len - sizeof(uint32));
 	if (nread != t_len - sizeof(uint32))
 		ereport(ERROR,
 				(errcode_for_file_access(),
-				 errmsg("could not read from HashAgg temporary file: %m")));
+				 errmsg("could not read the data from HashAgg spilled tape: %m")));
 
 	return tuple;
 }
@@ -2815,15 +2826,17 @@ hash_read_spilled(BufFile *file, uint32 *hashp)
  * be done. Should be called in the aggregate's memory context.
  */
 static HashAggBatch *
-hash_batch_new(BufFile *input_file, int setno, int64 input_tuples,
+hash_batch_new(LogicalTapeSet *lts, int tapenum, int setno, int64 input_tuples,
 			   int input_bits)
 {
 	HashAggBatch *batch = palloc0(sizeof(HashAggBatch));
 
-	batch->input_file = input_file;
+	batch->input_tape = tapenum;
 	batch->input_bits = input_bits;
 	batch->input_tuples = input_tuples;
 	batch->setno = setno;
+	batch->lts = lts;
+	batch->spill.lts = lts; // share same logical tape set
 
 	/* batch->spill will be set only after spilling this batch */
 
@@ -2860,7 +2873,7 @@ hash_finish_initial_spills(AggState *aggstate)
 /*
  * hash_spill_finish
  *
- * Transform spill files into new batches.
+ * Transform spill files into new batches. // XXX so the partitions are empty and ready to be reused
  */
 static void
 hash_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno, int input_bits)
@@ -2872,28 +2885,20 @@ hash_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno, int input_
 
 	for (i = 0; i < spill->n_partitions; i++)
 	{
-		BufFile         *file = spill->partitions[i];
 		MemoryContext    oldContext;
 		HashAggBatch    *new_batch;
 
-		/* partition is empty */
-		if (file == NULL)
-			continue;
-
-		/* rewind file for reading */
-		if (BufFileSeek(file, 0, 0L, SEEK_SET))
-			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not rewind HashAgg temporary file: %m")));
-
 		oldContext = MemoryContextSwitchTo(aggstate->ss.ps.state->es_query_cxt);
-		new_batch = hash_batch_new(file, setno, spill->ntuples[i],
-								   spill->partition_bits + input_bits);
+		LogicalTapeRewindForRead(spill->lts, spill->partitions[i], 0);
+		new_batch = hash_batch_new(spill->lts, spill->partitions[i], setno, spill->ntuples[i],
+					   spill->partition_bits + input_bits);
 		aggstate->hash_batches = lappend(aggstate->hash_batches, new_batch);
 		aggstate->hash_batches_used++;
 		MemoryContextSwitchTo(oldContext);
 	}
 
+	if (!list_member_ptr(aggstate->lts_list, spill->lts))
+		aggstate->lts_list = lappend(aggstate->lts_list, spill->lts);
 	pfree(spill->ntuples);
 	pfree(spill->partitions);
 }
@@ -2904,13 +2909,10 @@ hash_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno, int input_
 static void
 hash_reset_spill(HashAggSpill *spill)
 {
-	int i;
-	for (i = 0; i < spill->n_partitions; i++)
+	if (spill->lts != NULL)
 	{
-		BufFile         *file = spill->partitions[i];
-
-		if (file != NULL)
-			BufFileClose(file);
+		LogicalTapeSetClose(spill->lts);
+		spill->lts = NULL;
 	}
 	if (spill->ntuples != NULL)
 		pfree(spill->ntuples);
@@ -2940,16 +2942,19 @@ hash_reset_spills(AggState *aggstate)
 	foreach(lc, aggstate->hash_batches)
 	{
 		HashAggBatch *batch = (HashAggBatch*) lfirst(lc);
-		if (batch->input_file != NULL)
-		{
-			BufFileClose(batch->input_file);
-			batch->input_file = NULL;
-		}
 		hash_reset_spill(&batch->spill);
 		pfree(batch);
 	}
 	list_free(aggstate->hash_batches);
 	aggstate->hash_batches = NIL;
+
+	foreach(lc, aggstate->lts_list)
+	{
+		LogicalTapeSet *lts = (LogicalTapeSet *) lfirst(lc);
+		LogicalTapeSetClose(lts);
+	}
+	list_free(aggstate->lts_list);
+	aggstate->lts_list = NIL;
 }
 
 
diff --git a/src/backend/utils/sort/logtape.c b/src/backend/utils/sort/logtape.c
index 8985b9e095..677b992743 100644
--- a/src/backend/utils/sort/logtape.c
+++ b/src/backend/utils/sort/logtape.c
@@ -202,7 +202,7 @@ struct LogicalTapeSet
 
 	/* The array of logical tapes. */
 	int			nTapes;			/* # of logical tapes in set */
-	LogicalTape tapes[FLEXIBLE_ARRAY_MEMBER];	/* has nTapes nentries */
+	LogicalTape *tapes;	/* has nTapes nentries */
 };
 
 static void ltsWriteBlock(LogicalTapeSet *lts, long blocknum, void *buffer);
@@ -518,8 +518,8 @@ LogicalTapeSetCreate(int ntapes, TapeShare *shared, SharedFileSet *fileset,
 	 * Create top-level struct including per-tape LogicalTape structs.
 	 */
 	Assert(ntapes > 0);
-	lts = (LogicalTapeSet *) palloc(offsetof(LogicalTapeSet, tapes) +
-									ntapes * sizeof(LogicalTape));
+	lts = (LogicalTapeSet *) palloc0(sizeof(LogicalTapeSet));
+	lts->tapes = (LogicalTape *)palloc0(ntapes * sizeof(LogicalTape));
 	lts->nBlocksAllocated = 0L;
 	lts->nBlocksWritten = 0L;
 	lts->nHoleBlocks = 0L;
@@ -577,6 +577,45 @@ LogicalTapeSetCreate(int ntapes, TapeShare *shared, SharedFileSet *fileset,
 	return lts;
 }
 
+LogicalTapeSet *
+LogicalTapeSetExtend(LogicalTapeSet *lts, int ntoextend)
+{
+	LogicalTape    *lt;
+	int            i;
+
+	/*
+	 * Create top-level struct including per-tape LogicalTape structs.
+	 */
+	Assert(ntoextend > 0);
+	lts->tapes = (LogicalTape *) repalloc(lts->tapes, (lts->nTapes + ntoextend) * sizeof(LogicalTape));
+	lts->nTapes           = lts->nTapes + ntoextend;
+
+	/*
+	 * Initialize per-tape structs.  Note we allocate the I/O buffer and the
+	 * first block for a tape only when it is first actually written to.  This
+	 * avoids wasting memory space when we overestimate the number of tapes needed.
+	 */
+	for (i = lts->nTapes - ntoextend; i < lts->nTapes; i++)
+	{
+		lt = &lts->tapes[i];
+		lt->writing           = true;
+		lt->frozen            = false;
+		lt->dirty             = false;
+		lt->firstBlockNumber  = -1L;
+		lt->curBlockNumber    = -1L;
+		lt->nextBlockNumber   = -1L;
+		lt->offsetBlockNumber = 0L;
+		lt->buffer            = NULL;
+		lt->buffer_size       = 0;
+		/* palloc() larger than MaxAllocSize would fail */
+		lt->max_size          = MaxAllocSize;
+		lt->pos               = 0;
+		lt->nbytes            = 0;
+	}
+
+	return lts;
+}
+
 /*
  * Close a logical tape set and release all resources.
  */
@@ -1083,3 +1122,9 @@ LogicalTapeSetBlocks(LogicalTapeSet *lts)
 {
 	return lts->nBlocksAllocated - lts->nHoleBlocks;
 }
+
+int
+LogicalTapeGetNTapes(LogicalTapeSet *lts)
+{
+	return lts->nTapes;
+}
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 8d4a36a353..d45473101c 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2083,6 +2083,7 @@ typedef struct AggState
 	uint64		hash_disk_used; /* bytes of disk space used */
 	int			hash_batches_used;	/* batches used during entire execution */
 	List	   *hash_batches;	/* hash batches remaining to be processed */
+	List	   *lts_list;
 
 	AggStatePerHash perhash;	/* array of per-hashtable data */
 	AggStatePerGroup *hash_pergroup;	/* grouping set indexed array of
diff --git a/src/include/utils/logtape.h b/src/include/utils/logtape.h
index 081b03880a..c2f5c72665 100644
--- a/src/include/utils/logtape.h
+++ b/src/include/utils/logtape.h
@@ -56,6 +56,7 @@ typedef struct TapeShare
 
 extern LogicalTapeSet *LogicalTapeSetCreate(int ntapes, TapeShare *shared,
 											SharedFileSet *fileset, int worker);
+extern LogicalTapeSet * LogicalTapeSetExtend(LogicalTapeSet *lts, int ntoextend);
 extern void LogicalTapeSetClose(LogicalTapeSet *lts);
 extern void LogicalTapeSetForgetFreeSpace(LogicalTapeSet *lts);
 extern size_t LogicalTapeRead(LogicalTapeSet *lts, int tapenum,
@@ -74,5 +75,6 @@ extern void LogicalTapeSeek(LogicalTapeSet *lts, int tapenum,
 extern void LogicalTapeTell(LogicalTapeSet *lts, int tapenum,
 							long *blocknum, int *offset);
 extern long LogicalTapeSetBlocks(LogicalTapeSet *lts);
+extern int LogicalTapeGetNTapes(LogicalTapeSet *lts);
 
 #endif							/* LOGTAPE_H */

--zYM0uCDKw75PZbzx--