9.5: Memory-bounded HashAgg

diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 510d1c5..6455864 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -296,9 +296,18 @@ typedef struct AggHashEntryData *AggHashEntry;
 
 typedef struct AggHashEntryData
 {
-	TupleHashEntryData shared;	/* common header for hash table entries */
-	/* per-aggregate transition status array - must be last! */
+
+	/* pointer to the next entry in the bucket */
+	AggHashEntry next;
+
+	/* hash computed from the group keys (stored in mintuple) */
+	uint32	hashvalue;
+
+	/* minimal tuple storing values for group keys */
+	MinimalTuple tuple;
+
 	AggStatePerGroupData pergroup[1];	/* VARIABLE LENGTH ARRAY */
+
 }	AggHashEntryData;	/* VARIABLE LENGTH STRUCT */
 
 
@@ -321,7 +330,7 @@ static void finalize_aggregate(AggState *aggstate,
 				   Datum *resultVal, bool *resultIsNull);
 static Bitmapset *find_unaggregated_cols(AggState *aggstate);
 static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
-static void build_hash_table(AggState *aggstate);
+static void build_hash_table(AggState *aggstate, Size tuple_width);
 static AggHashEntry lookup_hash_entry(AggState *aggstate,
 				  TupleTableSlot *inputslot);
 static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
@@ -329,6 +338,86 @@ static void agg_fill_hash_table(AggState *aggstate);
 static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
 static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
 
+static uint32 compute_hash_value(AggState * aggstate, TupleTableSlot * slot);
+static uint32 compute_bucket(AggState * aggstate, uint32 hashvalue);
+static bool groups_match(AggState * aggstate, TupleTableSlot *slot, AggHashEntry entry);
+static void increase_nbuckets(AggState * aggstate);
+static char * chunk_alloc(AggHashTable htab, int size);
+static void reset_hash_table(AggHashTable htab);
+
+static void IteratorReset(AggHashTable htab);
+static AggHashEntry IteratorGetNext(AggHashTable htab);
+
+/*
+ * The size of the chunks for dense allocation. This needs to be >8kB
+ * because the default (and only) memory context implementation uses
+ * 8kB as a boundary for keeping the blocks on a freelist. Which is
+ * exactly what we don't want here - we want to free the chunk when
+ * we don't need it (so that it can be reused for aggstate and so on).
+ *
+ * 16kB seems like a good default value.
+ */
+#define HASH_CHUNK_SIZE			(16*1024L)
+
+typedef struct HashChunkData
+{
+	int			ntuples;	/* number of tuples stored in this chunk */
+	Size		maxlen;		/* length of the buffer */
+	Size		used;		/* number of chunk bytes already used */
+
+	struct HashChunkData   *next;	/* pointer to the next chunk (linked list) */
+
+	char		data[1];	/* buffer allocated at the end */
+} HashChunkData;
+
+typedef struct HashChunkData* HashChunk;
+
+/*
+ * A simple hashtable, storing the data densely into larger chunks.
+ * Originally, HashAgg used dynahash (through methods in nodeGrouping.c)
+ * but that does not allow removing the entries and freeing memory. So
+ * this approach, already used in nodeHash.c was used here too (and
+ * wrapped a bit more nicely).
+ *
+ * The hash entries (containing the per-group data) and tuples (with
+ * keys of the group) are interleaved, i.e. the entry is always stored
+ * first, then the tuple (in a MinimalTuple format). The entries are
+ * always fixed size (either the aggregate state is passed by value and
+ * stored inline, or passed by reference and stored in a regularly
+ * palloced memory), the tuples are of arbitrary size.
+ */
+typedef struct AggHashTableData
+{
+
+	int	nentries;		/* number of hash table entries */
+	int	nbuckets;		/* current number of buckets */
+	int	nbuckets_max;	/* max number of buckets */
+
+	/* items copied from the TupleHashTable, because we still need them */
+	MemoryContext	tmpctx;		/* short-lived memory context (hash/eq funcs) */
+	AttrNumber 	   *keyColIdx;	/* attr numbers of key columns */
+	int				numCols;	/* number of columns */
+	TupleTableSlot *slot;		/* tuple slot for groups_match */
+	Size			entrysize;	/* size of hash table entry (no tuple) */
+
+	MemoryContext	htabctx;	/* memory context for the chunks */
+
+	/* buckets of the hash table */
+	AggHashEntry   *buckets;
+
+	/*
+	 * Used for iterating through the hash table - it keeps track of the
+	 * current chunk, and entry within the chunk. Use the provided
+	 * methods to initialize and advance the iterator.
+	 */
+	HashChunk		cur_chunk;
+	AggHashEntry	cur_entry;
+
+	/* list of chunks with dense-packed entries / minimal tuples */
+	HashChunk		chunks_hash;
+
+} AggHashTableData;
+
 
 /*
  * Initialize all aggregates for a new group of input values.
@@ -928,26 +1017,111 @@ find_unaggregated_cols_walker(Node *node, Bitmapset **colnos)
  * The hash table always lives in the aggcontext memory context.
  */
 static void
-build_hash_table(AggState *aggstate)
+build_hash_table(AggState *aggstate, Size tuple_width)
 {
 	Agg		   *node = (Agg *) aggstate->ss.ps.plan;
-	MemoryContext tmpmem = aggstate->tmpcontext->ecxt_per_tuple_memory;
-	Size		entrysize;
+	Size		entrysize;	/* size of entry in the hash table */
+	Size		groupsize;	/* space used by the group (includes bucket) */
+	AggHashTable	htab;
+
+	/* we assume 1024 buckets (i.e. 8kB of memory) is minimum */
+	int nbuckets     = 1024;
+	int nbuckets_max = 1024;
 
 	Assert(node->aggstrategy == AGG_HASHED);
 	Assert(node->numGroups > 0);
 
-	entrysize = sizeof(AggHashEntryData) +
-		(aggstate->numaggs - 1) * sizeof(AggStatePerGroupData);
-
-	aggstate->hashtable = BuildTupleHashTable(node->numCols,
-											  node->grpColIdx,
-											  aggstate->eqfunctions,
-											  aggstate->hashfunctions,
-											  node->numGroups,
-											  entrysize,
-											  aggstate->aggcontext,
-											  tmpmem);
+	/*
+	 * Compute size of the hash table entry (this is actual size, but it
+	 * does not include the MinimalTuple size, with values of the keys
+	 * for the group). There's only a pointer to the minimal tuple.
+	 */
+	entrysize = MAXALIGN(sizeof(AggHashEntryData) +
+				(aggstate->numaggs - 1) * sizeof(AggStatePerGroupData));
+
+	/*
+	 * Estimate the size of the group, so that we can estimate how many
+	 * of them fit into work_mem, and thus estimate what is the reasonable
+	 * max number of buckets that we can use. To do that we add the entry
+	 * size, a bucket (because we're shooting for <1 load factor), and
+	 * estimated tuple width (because we'll keep the first tuple for each
+	 * group because of group key values).
+	 *
+	 * XXX This does not include size of the aggregate states, passed by
+	 *     reference. First, we don't know how to determine that. However,
+	 *     if the states are small it won't make much difference and if
+	 *     the states get large the memory required for the buckets is
+	 *     going to be much less important.
+	 */
+	groupsize = entrysize + sizeof(AggHashEntry)
+				+ sizeof(MinimalTupleData) + tuple_width;
+
+	/*
+	 * determine maximum number of buckets that can fit into work_mem (along with
+	 * the entry)
+	 *
+	 * This assumes all the space is used by AggHashEntries, but many aggregates
+	 * are keeping state separate (e.g. as a "pass by reference" Datums), which
+	 * results in nbuckets_max values higher than possible in practice. But we
+	 * don't know that at this point, and we don't need to worry too much about
+	 * it because those aggregates do it to handle states that are significantly
+	 * larger than 8B, which makes the 8B per-bucket negligible.
+	 *
+	 * And of course, as mentioned above, this does not include the actual data
+	 * stored in the MinimalTuple.
+	 *
+	 * XXX We may re-evaluate this over time, as we'll know how many entries are
+	 *     there, and thus what is the average size of aggregate size. That is,
+	 *     as the state size grows, we may decrease the number of buckets. We'll
+	 *     save a bit of memory by that (although not much).
+	 */
+	while (nbuckets_max * groupsize <= work_mem * 1024L)
+		nbuckets_max *= 2;
+
+	/*
+	 * Update the initial number of buckets to match expected number of groups,
+	 * but don't grow over nbuckets_max because in that case we'll start with
+	 * the batching anyway.
+	 */
+	while ((nbuckets < node->numGroups) && (nbuckets < nbuckets_max))
+		nbuckets *= 2;
+
+	/*
+	 * XXX When batching, we might use (numGroups / nbuckets) as a starting
+	 * nbatch value, but maybe we can start with nbatch=1 with the assumption
+	 * that multiple tuples will be 'compressed' into the group (and thus
+	 * we'll write less data in total).
+	 */
+
+	htab = (AggHashTable)MemoryContextAllocZero(aggstate->aggcontext,
+											sizeof(AggHashTableData));
+
+	/* TODO create a memory context for the hash table */
+	htab->htabctx = AllocSetContextCreate(aggstate->aggcontext,
+											"HashAggHashTable",
+											ALLOCSET_DEFAULT_MINSIZE,
+											ALLOCSET_DEFAULT_INITSIZE,
+											ALLOCSET_DEFAULT_MAXSIZE);
+
+	/* buckets are just pointers to AggHashEntryData structures */
+	htab->buckets = (AggHashEntry*)MemoryContextAllocZero(htab->htabctx,
+									nbuckets * sizeof(AggHashEntry));
+
+	/* copy the column IDs from the node */
+	htab->keyColIdx = node->grpColIdx;
+
+	/* we'll use the per-tuple memory context for the hash/eq functions */
+	htab->tmpctx = aggstate->tmpcontext->ecxt_per_tuple_memory;
+
+	htab->nbuckets = nbuckets;
+	htab->nbuckets_max = nbuckets_max;
+	htab->nentries = 0;
+	htab->slot = NULL;
+	htab->numCols = node->numCols;
+	htab->entrysize = entrysize;
+
+	aggstate->hashtable = htab;
+
 }
 
 /*
@@ -1026,40 +1200,77 @@ hash_agg_entry_size(int numAggs)
 static AggHashEntry
 lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
 {
-	TupleTableSlot *hashslot = aggstate->hashslot;
-	ListCell   *l;
-	AggHashEntry entry;
-	bool		isnew;
 
-	/* if first time through, initialize hashslot by cloning input slot */
-	if (hashslot->tts_tupleDescriptor == NULL)
+	AggHashEntry entry = NULL;
+	uint32		hashvalue;
+	uint32		bucketno;
+	MinimalTuple mintuple;
+
+	hashvalue = compute_hash_value(aggstate, inputslot);
+	bucketno = compute_bucket(aggstate, hashvalue);
+
+	entry = aggstate->hashtable->buckets[bucketno];
+
+	/* try to find a matching entry in the hash table (in the bucket) */
+	while (entry != NULL)
 	{
-		ExecSetSlotDescriptor(hashslot, inputslot->tts_tupleDescriptor);
-		/* Make sure all unused columns are NULLs */
-		ExecStoreAllNullTuple(hashslot);
+
+		/* first check the hashes, only then check the keys (if hashes match) */
+		if ((entry->hashvalue == hashvalue) && (groups_match(aggstate, inputslot, entry)))
+			break;
+
+		/* these are not the entries you're looking for ... */
+		entry = entry->next;
 	}
 
-	/* transfer just the needed columns into hashslot */
-	slot_getsomeattrs(inputslot, linitial_int(aggstate->hash_needed));
-	foreach(l, aggstate->hash_needed)
+	/* There's not a maching entry in the bucket, so create a new one and
+	 * copy in data both for the aggregates, and the MinimalTuple containing
+	 * keys for the group columns. */
+	if (entry == NULL)
 	{
-		int			varNumber = lfirst_int(l) - 1;
 
-		hashslot->tts_values[varNumber] = inputslot->tts_values[varNumber];
-		hashslot->tts_isnull[varNumber] = inputslot->tts_isnull[varNumber];
-	}
+		MemoryContext old;
 
-	/* find or create the hashtable entry using the filtered tuple */
-	entry = (AggHashEntry) LookupTupleHashEntry(aggstate->hashtable,
-												hashslot,
-												&isnew);
+		/* only a reference to the mintuple - we'll copy it into a chunk */
+		mintuple = ExecFetchSlotMinimalTuple(inputslot);
+
+		/* FIXME probably create a separate context for the hash table, instead
+		 * of using aggcontext for everything ... */
+		old = MemoryContextSwitchTo(aggstate->aggcontext);
+
+		/* we need enough space for the entry and tuple with key values */
+		entry = (AggHashEntry) chunk_alloc(aggstate->hashtable,
+							aggstate->hashtable->entrysize + mintuple->t_len);
+
+		entry->hashvalue = hashvalue;
+
+		/* add to the proper bucket */
+		entry->next = aggstate->hashtable->buckets[bucketno];
+		aggstate->hashtable->buckets[bucketno] = entry;
+
+		/* the tuple is placed right after the entry (maxaligned) */
+		entry->tuple = (MinimalTuple)((char*)entry + aggstate->hashtable->entrysize);
+
+		/*
+		 * FIXME This seems to copy all the data, including columns that are not part
+		 * of the key (i.e. are there only as inputs for the aggregates - that may be
+		 * quite wasteful when there are many aggregates / the values are long etc.)
+		 */
+		memcpy(entry->tuple, mintuple, mintuple->t_len);
+
+		MemoryContextSwitchTo(old);
 
-	if (isnew)
-	{
 		/* initialize aggregates for new tuple group */
 		initialize_aggregates(aggstate, aggstate->peragg, entry->pergroup);
+
+		aggstate->hashtable->nentries += 1;
+
 	}
 
+	/* once we exceed 1 entry / bucket, increase number of buckets */
+	if (aggstate->hashtable->nentries > aggstate->hashtable->nbuckets)
+		increase_nbuckets(aggstate);
+
 	return entry;
 }
 
@@ -1363,8 +1574,10 @@ agg_fill_hash_table(AggState *aggstate)
 	}
 
 	aggstate->table_filled = true;
-	/* Initialize to walk the hash table */
-	ResetTupleHashIterator(aggstate->hashtable, &aggstate->hashiter);
+
+	/* Initialize for iteration through the table (first bucket / entry) */
+	IteratorReset(aggstate->hashtable);
+
 }
 
 /*
@@ -1381,6 +1594,7 @@ agg_retrieve_hash_table(AggState *aggstate)
 	AggHashEntry entry;
 	TupleTableSlot *firstSlot;
 	int			aggno;
+	AggHashTable	htab;
 
 	/*
 	 * get state info from node
@@ -1391,23 +1605,11 @@ agg_retrieve_hash_table(AggState *aggstate)
 	aggnulls = econtext->ecxt_aggnulls;
 	peragg = aggstate->peragg;
 	firstSlot = aggstate->ss.ss_ScanTupleSlot;
+	htab = aggstate->hashtable;
 
-	/*
-	 * We loop retrieving groups until we find one satisfying
-	 * aggstate->ss.ps.qual
-	 */
-	while (!aggstate->agg_done)
+	/* loop over entries in buckets */
+	while ((entry = IteratorGetNext(htab)) != NULL)
 	{
-		/*
-		 * Find the next entry in the hash table
-		 */
-		entry = (AggHashEntry) ScanTupleHashTable(&aggstate->hashiter);
-		if (entry == NULL)
-		{
-			/* No more entries in hashtable, so done */
-			aggstate->agg_done = TRUE;
-			return NULL;
-		}
 
 		/*
 		 * Clear the per-output-tuple context for each group
@@ -1419,19 +1621,19 @@ agg_retrieve_hash_table(AggState *aggstate)
 		ResetExprContext(econtext);
 
 		/*
-		 * Store the copied first input tuple in the tuple table slot reserved
-		 * for it, so that it can be used in ExecProject.
-		 */
-		ExecStoreMinimalTuple(entry->shared.firstTuple,
-							  firstSlot,
-							  false);
+		* Store the copied first input tuple in the tuple table slot reserved
+		* for it, so that it can be used in ExecProject.
+		*/
+		ExecStoreMinimalTuple(entry->tuple,
+							firstSlot,
+							false);
 
 		pergroup = entry->pergroup;
 
 		/*
-		 * Finalize each aggregate calculation, and stash results in the
-		 * per-output-tuple context.
-		 */
+		* Finalize each aggregate calculation, and stash results in the
+		* per-output-tuple context.
+		*/
 		for (aggno = 0; aggno < aggstate->numaggs; aggno++)
 		{
 			AggStatePerAgg peraggstate = &peragg[aggno];
@@ -1439,25 +1641,25 @@ agg_retrieve_hash_table(AggState *aggstate)
 
 			Assert(peraggstate->numSortCols == 0);
 			finalize_aggregate(aggstate, peraggstate, pergroupstate,
-							   &aggvalues[aggno], &aggnulls[aggno]);
+							&aggvalues[aggno], &aggnulls[aggno]);
 		}
 
 		/*
-		 * Use the representative input tuple for any references to
-		 * non-aggregated input columns in the qual and tlist.
-		 */
+		* Use the representative input tuple for any references to
+		* non-aggregated input columns in the qual and tlist.
+		*/
 		econtext->ecxt_outertuple = firstSlot;
 
 		/*
-		 * Check the qual (HAVING clause); if the group does not match, ignore
-		 * it and loop back to try to process another group.
-		 */
+		* Check the qual (HAVING clause); if the group does not match, ignore
+		* it and loop back to try to process another group.
+		*/
 		if (ExecQual(aggstate->ss.ps.qual, econtext, false))
 		{
 			/*
-			 * Form and return a projection tuple using the aggregate results
-			 * and the representative input tuple.
-			 */
+			* Form and return a projection tuple using the aggregate results
+			* and the representative input tuple.
+			*/
 			TupleTableSlot *result;
 			ExprDoneCond isDone;
 
@@ -1472,8 +1674,11 @@ agg_retrieve_hash_table(AggState *aggstate)
 		}
 		else
 			InstrCountFiltered1(aggstate, 1);
+
 	}
 
+	aggstate->agg_done = true;
+
 	/* No more groups */
 	return NULL;
 }
@@ -1515,7 +1720,6 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	aggstate->agg_done = false;
 	aggstate->pergroup = NULL;
 	aggstate->grp_firstTuple = NULL;
-	aggstate->hashtable = NULL;
 
 	/*
 	 * Create expression contexts.  We need two, one for per-input-tuple
@@ -1546,7 +1750,9 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	 */
 	ExecInitScanTupleSlot(estate, &aggstate->ss);
 	ExecInitResultTupleSlot(estate, &aggstate->ss.ps);
-	aggstate->hashslot = ExecInitExtraTupleSlot(estate);
+
+	/* FIXME maybe we could reuse this in groups_match for better efficiency (?) */
+	// aggstate->hashslot = ExecInitExtraTupleSlot(estate);
 
 	/*
 	 * initialize child expressions
@@ -1636,7 +1842,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 
 	if (node->aggstrategy == AGG_HASHED)
 	{
-		build_hash_table(aggstate);
+		build_hash_table(aggstate, outerPlan->plan_width);
 		aggstate->table_filled = false;
 		/* Compute the columns we actually need to hash on */
 		aggstate->hash_needed = find_hash_columns(aggstate);
@@ -2073,7 +2279,7 @@ ExecReScanAgg(AggState *node)
 		 */
 		if (node->ss.ps.lefttree->chgParam == NULL)
 		{
-			ResetTupleHashIterator(node->hashtable, &node->hashiter);
+			IteratorReset(node->hashtable);
 			return;
 		}
 	}
@@ -2112,8 +2318,9 @@ ExecReScanAgg(AggState *node)
 
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
+		Plan * outerPlan = outerPlan((Agg *) node->ss.ps.plan);
 		/* Rebuild an empty hash table */
-		build_hash_table(node);
+		build_hash_table(node, outerPlan->plan_width);
 		node->table_filled = false;
 	}
 	else
@@ -2269,3 +2476,385 @@ aggregate_dummy(PG_FUNCTION_ARGS)
 		 fcinfo->flinfo->fn_oid);
 	return (Datum) 0;			/* keep compiler quiet */
 }
+
+/*
+ * Computes a hash value from the group keys - this is pretty much the
+ * same as TupleHashTableHash, except that it's simplified a bit, and
+ * does not pass the tuples through an input etc.
+ */
+static uint32
+compute_hash_value(AggState * aggstate, TupleTableSlot * slot)
+{
+
+	uint32		hashkey = 0;
+	FmgrInfo   *hashfunctions = aggstate->hashfunctions;
+	int i = 0;
+
+	MemoryContext oldContext;
+
+	/* FIXME is it really OK to reset the per-tuple context here? */
+
+	/* Reset and switch into the temp context. */
+	MemoryContextReset(aggstate->hashtable->tmpctx);
+	oldContext = MemoryContextSwitchTo(aggstate->hashtable->tmpctx);
+
+	/* compute hash only from the needed column */
+	for (i = 0; i < aggstate->hashtable->numCols; i++)
+	{
+
+		AttrNumber	att = aggstate->hashtable->keyColIdx[i];
+		Datum		attr;
+		bool		isNull;
+
+		/* rotate hashkey left 1 bit at each step */
+		hashkey = (hashkey << 1) | ((hashkey & 0x80000000) ? 1 : 0);
+
+		attr = slot_getattr(slot, att, &isNull);
+
+		if (!isNull)			/* treat nulls as having hash key 0 */
+		{
+			uint32		hkey;
+
+			hkey = DatumGetUInt32(FunctionCall1(&hashfunctions[i],
+												attr));
+			hashkey ^= hkey;
+		}
+	}
+
+	MemoryContextSwitchTo(oldContext);
+
+	return hashkey;
+
+}
+
+/*
+ * Computes index of the bucket the group entry belongs to (same principles as
+ * in ExecHashGetBucketAndBatch in nodeHash.c)
+ */
+static uint32
+compute_bucket(AggState * aggstate, uint32 hashvalue)
+{
+	return hashvalue & (aggstate->hashtable->nbuckets - 1);
+}
+
+/*
+ * Compares that the group keys of the two groups actually match, using the
+ * equality functions. This is much more expensive than comparing uint32
+ * values (hashes), so always check hashes first.
+ */
+static bool
+groups_match(AggState * aggstate, TupleTableSlot *slot, AggHashEntry entry)
+{
+	bool		result;
+	FmgrInfo   *eqfunctions = aggstate->eqfunctions;
+	TupleDesc	tupdesc;
+	int i = 0;
+
+	MemoryContext oldContext;
+
+	/*
+	 * XXX Do we really need to do this slot gymnastics? can't we get the
+	 * info from the minimal tuple directly? It init happens only once,
+	 * so the overhead is not that bad, but it's annoying. And we still
+	 * have to call ExecStoreMinimalTuple every time.
+	 */
+	if (aggstate->hashtable->slot == NULL)
+	{
+		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
+		tupdesc = CreateTupleDescCopy(slot->tts_tupleDescriptor);
+		aggstate->hashtable->slot = MakeSingleTupleTableSlot(tupdesc);
+		MemoryContextSwitchTo(oldContext);
+	}
+
+	/* FIXME is it really OK to reset the per-tuple memory context here? */
+
+	/* Reset and switch into the temp context. */
+	MemoryContextReset(aggstate->hashtable->tmpctx);
+	oldContext = MemoryContextSwitchTo(aggstate->hashtable->tmpctx);
+
+	ExecStoreMinimalTuple(entry->tuple, aggstate->hashtable->slot, false);
+
+	/*
+	 * We cannot report a match without checking all the fields, but we can
+	 * report a non-match as soon as we find unequal fields.  So, start
+	 * comparing at the last field (least significant sort key). That's the
+	 * most likely to be different if we are dealing with sorted input.
+	 */
+	result = true;
+
+	for (i = aggstate->hashtable->numCols; --i >= 0;)
+	{
+
+		AttrNumber	att = aggstate->hashtable->keyColIdx[i];
+		Datum		attr1,
+					attr2;
+		bool		isNull1,
+					isNull2;
+
+		attr1 = slot_getattr(slot,  att, &isNull1);
+		attr2 = slot_getattr(aggstate->hashtable->slot, att, &isNull2);
+
+		if (isNull1 != isNull2)
+		{
+			result = false;		/* one null and one not; they aren't equal */
+			break;
+		}
+
+		if (isNull1)
+			continue;			/* both are null, treat as equal */
+
+		/* Apply the type-specific equality function */
+
+		if (!DatumGetBool(FunctionCall2(&eqfunctions[i],
+										attr1, attr2)))
+		{
+			result = false;		/* they aren't equal */
+			break;
+		}
+	}
+
+	MemoryContextSwitchTo(oldContext);
+
+	return result;
+}
+
+/*
+ * Resize the hash table for good performance. We're shooting for (nentries <= nbuckets)
+ * which should give us 1 group per bucket on average. We're working with groups and not
+ * tuples. And multiple tuples with the same hash are most likely in the same group, thus
+ * merged into a single entry. So we should not see many buckets with a long list of
+ * entries (which can happen in hashjoin quite easily).
+ */
+static void
+increase_nbuckets(AggState * aggstate)
+{
+
+	HashChunk chunk;
+	AggHashTable htab = aggstate->hashtable;
+
+	/* we've reached maximum number of buckets */
+	if (htab->nbuckets >= htab->nbuckets_max)
+		return;
+
+	htab->nbuckets *= 2;
+	htab->buckets
+		= (AggHashEntry*)repalloc(htab->buckets,
+								  htab->nbuckets * sizeof(AggHashEntry));
+	memset(htab->buckets, 0, htab->nbuckets * sizeof(AggHashEntry));
+
+	chunk = htab->chunks_hash;
+	while (chunk != NULL)
+	{
+
+		/* position within the buffer (up to chunk->used) */
+		size_t idx = 0;
+
+		/* we have a whole number of entries */
+		Assert(chunk->used % htab->entrysize == 0);
+
+		/* process all tuples stored in this chunk (and then free it) */
+		while (idx < chunk->used)
+		{
+
+			AggHashEntry entry = (AggHashEntry)(chunk->data + idx);
+
+			int bucketno = compute_bucket(aggstate, entry->hashvalue);
+
+			entry->next = htab->buckets[bucketno];
+			htab->buckets[bucketno] = entry;
+
+			/* bytes occupied in memory HJ tuple overhead + actual tuple length */
+			idx += htab->entrysize + entry->tuple->t_len;
+
+		}
+
+		/* proceed to the next chunk */
+		chunk = chunk->next;
+
+	}
+
+}
+
+static
+char * chunk_alloc(AggHashTable htab, int size)
+{
+	/* XXX maybe we should use MAXALIGN(size) here ... */
+
+	/* we need >8kB to get immediate free in aset.c */
+	Assert(HASH_CHUNK_SIZE > 8192);
+
+	/*
+	 * If the requested size is over 1/8 of chunk size, allocate a
+	 * separate chunk. of this size.
+	 *
+	 * XXXX This may be problematic, because chunks like this may get
+	 *      below 8kB, and thus be considered 'regular' blocks by aset.c
+	 *      (and put on freelist, instead of freeing immediately).
+	 */
+	if (size > (HASH_CHUNK_SIZE/8))
+	{
+
+		/*
+		 * Allocate new chunk and put it at the beginning of the list.
+		 *
+		 * There's no point in making this 2^N size, because blocks over
+		 * 8kB are handled as a special case in aset.c (exact size).
+		 */
+		HashChunk newChunk
+			= (HashChunk)MemoryContextAllocZero(htab->htabctx,
+								offsetof(HashChunkData, data) + size);
+
+		newChunk->maxlen = size;
+		newChunk->used = 0;
+		newChunk->ntuples = 0;
+
+		/*
+		 * If there already is a chunk, add the new one after it, so we
+		 * can still use the space in the existing one.
+		 */
+		if (htab->chunks_hash != NULL)
+		{
+			newChunk->next = htab->chunks_hash->next;
+			htab->chunks_hash->next = newChunk;
+		}
+		else
+		{
+			newChunk->next = htab->chunks_hash;
+			htab->chunks_hash = newChunk;
+		}
+
+		newChunk->used += size;
+		newChunk->ntuples += 1;
+
+		return newChunk->data;
+
+	}
+
+	/*
+	 * Requested size is less than 1/8 of a chunk, so place it in the
+	 * current chunk if there is enough free space. If not, allocate
+	 * a new chunk and add it there.
+	 */
+	if ((htab->chunks_hash == NULL) ||
+		(htab->chunks_hash->maxlen - htab->chunks_hash->used) < size)
+	{
+		/* allocate new chunk and put it at the beginning of the list */
+		HashChunk newChunk
+			= (HashChunk)MemoryContextAllocZero(htab->htabctx,
+						offsetof(HashChunkData, data) + HASH_CHUNK_SIZE);
+
+		newChunk->maxlen = HASH_CHUNK_SIZE;
+		newChunk->used = 0;
+		newChunk->ntuples = 0;
+
+		newChunk->next = htab->chunks_hash;
+		htab->chunks_hash = newChunk;
+	}
+
+	/* OK, we have enough space in the chunk, let's add the tuple */
+	htab->chunks_hash->used += size;
+	htab->chunks_hash->ntuples += 1;
+
+	/* allocate pointer to the start of the tuple memory */
+	return htab->chunks_hash->data + (htab->chunks_hash->used - size);
+
+}
+
+/*
+ * Resets the hash table iterator, so that it points to the first entry
+ * in the first chunk (the chunk created last, thus placed first in the
+ * list of chunks).
+ */
+static
+void IteratorReset(AggHashTable htab)
+{
+
+	htab->cur_chunk = htab->chunks_hash;
+
+	/* there may be no chunks at all (empty hash table) */
+	if (htab->cur_chunk != NULL)
+		htab->cur_entry = (AggHashEntry)htab->cur_chunk->data;
+	else
+		htab->cur_entry = NULL;
+
+}
+
+/*
+ * Returns the next hash table entry. Works by scanning the chunks, not
+ * by scanning the buckets etc. Returns NULL when there are no more
+ * entries.
+ */
+static
+AggHashEntry IteratorGetNext(AggHashTable htab)
+{
+
+	AggHashEntry	entry = NULL;
+	Size			len;
+
+	/* we've completed the last chunk (in the previous call) */
+	if (htab->cur_chunk == NULL)
+		return NULL;
+
+	/* we're not beyond the chunk data */
+	Assert((char*)htab->cur_entry < (htab->cur_chunk->data + htab->cur_chunk->used));
+
+	/*
+	 * We're still in the current chunk (otherwise the current chunk
+	 * would be set to NULL), so cur_entry points to a valid entry.
+	 * So compute how many bytes we need to skip to the next entry.
+	 */
+	entry = htab->cur_entry;
+	len = entry->tuple->t_len + htab->entrysize;
+
+	/*
+	 * Proceed to the next entry and check if we've reached end of this
+	 * chunk. If yes, skip to the next one and set the current entry
+	 * accordingly (chunk=NULL means there's no valid entry).
+	 */
+	htab->cur_entry = (AggHashEntry)((char*)entry + len);
+
+	if ((char*)htab->cur_entry >= (htab->cur_chunk->data + htab->cur_chunk->used))
+	{
+		htab->cur_chunk = htab->cur_chunk->next;
+		if (htab->cur_chunk != NULL)
+			htab->cur_entry = (AggHashEntry)htab->cur_chunk->data;
+		else
+			htab->cur_entry = NULL;
+	}
+
+	return entry;
+
+}
+
+/*
+ * Resets the contents of the hash table - removes all the entries and
+ * tuples, but keeps the 'size' of the hash table (nbuckets).
+ */
+static
+void reset_hash_table(AggHashTable htab) {
+
+	MemoryContext htabctx = htab->htabctx;
+	MemoryContext parent = htab->htabctx->parent;
+
+	htab->nentries = 0;
+	htab->chunks_hash = NULL;
+
+	/*
+	 * XXX If we could reset the context instead of recreating it
+	 *     from scratch, that'd be nice. However currently the reset
+	 *     often does not free a lot of memory because it keeps the
+	 *     blocks for future allocations.
+	 */
+	htab->htabctx = AllocSetContextCreateTracked(parent,
+											"HashAggHashTable",
+											ALLOCSET_DEFAULT_MINSIZE,
+											ALLOCSET_DEFAULT_INITSIZE,
+											ALLOCSET_DEFAULT_MAXSIZE,
+											true);
+
+	MemoryContextDelete(htabctx);
+
+	htab->buckets = (AggHashEntry*)MemoryContextAllocZero(htab->htabctx,
+								htab->nbuckets * sizeof(AggHashEntry));
+
+}
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index b271f21..995389b 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1701,6 +1701,8 @@ typedef struct GroupState
 /* these structs are private in nodeAgg.c: */
 typedef struct AggStatePerAggData *AggStatePerAgg;
 typedef struct AggStatePerGroupData *AggStatePerGroup;
+typedef struct AggHashEntryData *AggHashEntry;
+typedef struct AggHashTableData *AggHashTable;
 
 typedef struct AggState
 {
@@ -1714,15 +1716,16 @@ typedef struct AggState
 	ExprContext *tmpcontext;	/* econtext for input expressions */
 	AggStatePerAgg curperagg;	/* identifies currently active aggregate */
 	bool		agg_done;		/* indicates completion of Agg scan */
+
 	/* these fields are used in AGG_PLAIN and AGG_SORTED modes: */
 	AggStatePerGroup pergroup;	/* per-Aggref-per-group working state */
 	HeapTuple	grp_firstTuple; /* copy of first tuple of current group */
+
 	/* these fields are used in AGG_HASHED mode: */
-	TupleHashTable hashtable;	/* hash table with one entry per group */
-	TupleTableSlot *hashslot;	/* slot for loading hash table */
 	List	   *hash_needed;	/* list of columns needed in hash table */
 	bool		table_filled;	/* hash table filled yet? */
-	TupleHashIterator hashiter; /* for iterating through hash table */
+	AggHashTable	hashtable;	/* instance of the simple hash table */
+
 } AggState;
 
 /* ----------------

diff --git a/src/backend/utils/mmgr/aset.c b/src/backend/utils/mmgr/aset.c
index 743455e..d556f0b 100644
--- a/src/backend/utils/mmgr/aset.c
+++ b/src/backend/utils/mmgr/aset.c
@@ -242,6 +242,8 @@ typedef struct AllocChunkData
 #define AllocChunkGetPointer(chk)	\
 					((AllocPointer)(((char *)(chk)) + ALLOC_CHUNKHDRSZ))
 
+static void update_allocation(MemoryContext context, Size size);
+
 /*
  * These functions implement the MemoryContext API for AllocSet contexts.
  */
@@ -430,6 +432,9 @@ randomize_mem(char *ptr, size_t size)
  * minContextSize: minimum context size
  * initBlockSize: initial allocation block size
  * maxBlockSize: maximum allocation block size
+ *
+ * The flag determining whether this context tracks memory usage is inherited
+ * from the parent context.
  */
 MemoryContext
 AllocSetContextCreate(MemoryContext parent,
@@ -438,6 +443,26 @@ AllocSetContextCreate(MemoryContext parent,
 					  Size initBlockSize,
 					  Size maxBlockSize)
 {
+	return AllocSetContextCreateTracked(
+		parent, name, minContextSize, initBlockSize, maxBlockSize,
+		false);
+}
+
+/*
+ * AllocSetContextCreateTracked
+ *		Create a new AllocSet context.
+ *
+ * Implementation for AllocSetContextCreate, but also allows the caller to
+ * specify whether memory usage should be tracked or not.
+ */
+MemoryContext
+AllocSetContextCreateTracked(MemoryContext parent,
+							 const char *name,
+							 Size minContextSize,
+							 Size initBlockSize,
+							 Size maxBlockSize,
+							 bool track_mem)
+{
 	AllocSet	context;
 
 	/* Do the type-independent part of context creation */
@@ -445,7 +470,8 @@ AllocSetContextCreate(MemoryContext parent,
 											 sizeof(AllocSetContext),
 											 &AllocSetMethods,
 											 parent,
-											 name);
+											 name,
+											 track_mem);
 
 	/*
 	 * Make sure alloc parameters are reasonable, and save them.
@@ -500,6 +526,9 @@ AllocSetContextCreate(MemoryContext parent,
 					 errdetail("Failed while creating memory context \"%s\".",
 							   name)));
 		}
+
+		update_allocation((MemoryContext) context, blksize);
+
 		block->aset = context;
 		block->freeptr = ((char *) block) + ALLOC_BLOCKHDRSZ;
 		block->endptr = ((char *) block) + blksize;
@@ -590,6 +619,7 @@ AllocSetReset(MemoryContext context)
 		else
 		{
 			/* Normal case, release the block */
+			update_allocation(context, -(block->endptr - ((char*) block)));
 #ifdef CLOBBER_FREED_MEMORY
 			wipe_mem(block, block->freeptr - ((char *) block));
 #endif
@@ -616,6 +646,8 @@ AllocSetDelete(MemoryContext context)
 	AllocSet	set = (AllocSet) context;
 	AllocBlock	block = set->blocks;
 
+	MemoryAccounting accounting;
+
 	AssertArg(AllocSetIsValid(set));
 
 #ifdef MEMORY_CONTEXT_CHECKING
@@ -623,6 +655,17 @@ AllocSetDelete(MemoryContext context)
 	AllocSetCheck(context);
 #endif
 
+	if (context->accounting != NULL) {
+
+		accounting = context->accounting->parent;
+
+		while (accounting != NULL)
+		{
+			accounting->total_allocated -= context->accounting->total_allocated;
+			accounting = accounting->parent;
+		}
+	}
+
 	/* Make it look empty, just in case... */
 	MemSetAligned(set->freelist, 0, sizeof(set->freelist));
 	set->blocks = NULL;
@@ -678,6 +721,9 @@ AllocSetAlloc(MemoryContext context, Size size)
 					 errmsg("out of memory"),
 					 errdetail("Failed on request of size %zu.", size)));
 		}
+
+		update_allocation(context, blksize);
+
 		block->aset = set;
 		block->freeptr = block->endptr = ((char *) block) + blksize;
 
@@ -873,6 +919,8 @@ AllocSetAlloc(MemoryContext context, Size size)
 					 errdetail("Failed on request of size %zu.", size)));
 		}
 
+		update_allocation(context, blksize);
+
 		block->aset = set;
 		block->freeptr = ((char *) block) + ALLOC_BLOCKHDRSZ;
 		block->endptr = ((char *) block) + blksize;
@@ -976,6 +1024,7 @@ AllocSetFree(MemoryContext context, void *pointer)
 			set->blocks = block->next;
 		else
 			prevblock->next = block->next;
+		update_allocation(context, -(block->endptr - ((char*) block)));
 #ifdef CLOBBER_FREED_MEMORY
 		wipe_mem(block, block->freeptr - ((char *) block));
 #endif
@@ -1088,6 +1137,7 @@ AllocSetRealloc(MemoryContext context, void *pointer, Size size)
 		AllocBlock	prevblock = NULL;
 		Size		chksize;
 		Size		blksize;
+		Size		oldblksize;
 
 		while (block != NULL)
 		{
@@ -1105,6 +1155,8 @@ AllocSetRealloc(MemoryContext context, void *pointer, Size size)
 		/* Do the realloc */
 		chksize = MAXALIGN(size);
 		blksize = chksize + ALLOC_BLOCKHDRSZ + ALLOC_CHUNKHDRSZ;
+		oldblksize = block->endptr - ((char *)block);
+
 		block = (AllocBlock) realloc(block, blksize);
 		if (block == NULL)
 		{
@@ -1114,6 +1166,7 @@ AllocSetRealloc(MemoryContext context, void *pointer, Size size)
 					 errmsg("out of memory"),
 					 errdetail("Failed on request of size %zu.", size)));
 		}
+		update_allocation(context, blksize - oldblksize);
 		block->freeptr = block->endptr = ((char *) block) + blksize;
 
 		/* Update pointers since block has likely been moved */
@@ -1277,6 +1330,45 @@ AllocSetStats(MemoryContext context, int level)
 }
 
 
+/*
+ * update_allocation
+ *
+ * Track newly-allocated or newly-freed memory (freed memory should be
+ * negative).
+ */
+static void
+update_allocation(MemoryContext context, Size size)
+{
+
+	MemoryAccounting accounting = context->accounting;
+
+	if (accounting == NULL)
+		return;
+
+	/*
+	 * Update self_allocated only if this accounting info is specific
+	 * for this context (i.e. if tracking was requested for the context).
+	 */
+	if (context->track_mem)
+		accounting->self_allocated += size;
+
+	/*
+	 * Update total_allocated for all contexts up the accounting tree
+	 * (including this one).
+	 */
+	while (accounting != NULL) {
+
+		accounting->total_allocated += size;
+
+		Assert(accounting->self_allocated >= 0);
+		Assert(accounting->total_allocated >= accounting->self_allocated);
+
+		accounting = accounting->parent;
+
+	}
+
+}
+
 #ifdef MEMORY_CONTEXT_CHECKING
 
 /*
diff --git a/src/backend/utils/mmgr/mcxt.c b/src/backend/utils/mmgr/mcxt.c
index 4185a03..a70b296 100644
--- a/src/backend/utils/mmgr/mcxt.c
+++ b/src/backend/utils/mmgr/mcxt.c
@@ -202,7 +202,12 @@ MemoryContextDelete(MemoryContext context)
 	 */
 	MemoryContextSetParent(context, NULL);
 
+	/* pass the parent in case it's needed, however */
 	(*context->methods->delete_context) (context);
+
+	if (context->track_mem)
+		pfree(context->accounting);
+
 	VALGRIND_DESTROY_MEMPOOL(context);
 	pfree(context);
 }
@@ -324,6 +329,26 @@ MemoryContextAllowInCriticalSection(MemoryContext context, bool allow)
 }
 
 /*
+ * MemoryContextGetAllocated
+ *
+ * Return memory allocated by the system to this context. If total is true,
+ * include child contexts. Context must have track_mem set.
+ */
+Size
+MemoryContextGetAllocated(MemoryContext context, bool total)
+{
+	Assert(context->track_mem);
+
+	if (! context->track_mem)
+		return 0;
+
+	if (total)
+		return context->accounting->total_allocated;
+	else
+		return context->accounting->self_allocated;
+}
+
+/*
  * GetMemoryChunkSpace
  *		Given a currently-allocated chunk, determine the total space
  *		it occupies (including all memory-allocation overhead).
@@ -546,7 +571,8 @@ MemoryContext
 MemoryContextCreate(NodeTag tag, Size size,
 					MemoryContextMethods *methods,
 					MemoryContext parent,
-					const char *name)
+					const char *name,
+					bool track_mem)
 {
 	MemoryContext node;
 	Size		needed = size + strlen(name) + 1;
@@ -576,6 +602,8 @@ MemoryContextCreate(NodeTag tag, Size size,
 	node->firstchild = NULL;
 	node->nextchild = NULL;
 	node->isReset = true;
+	node->track_mem = track_mem;
+	node->accounting = NULL;
 	node->name = ((char *) node) + size;
 	strcpy(node->name, name);
 
@@ -596,6 +624,24 @@ MemoryContextCreate(NodeTag tag, Size size,
 #endif
 	}
 
+	/*
+	 * If accounting was requested for this context, create the struct
+	 * and link it to the accounting from parent context. Otherwise just
+	 * copy the accounting reference from parent.
+	 *
+	 * If both cases, accounting in parent may be NULL, which means
+	 * we don't need to update accounting in the upper memory contexts.
+	 */
+	if (track_mem)
+	{
+		node->accounting = (MemoryAccounting)MemoryContextAlloc(TopMemoryContext,
+												sizeof(MemoryAccountingData));
+		if (parent)
+			node->accounting->parent = parent->accounting;
+	} else if (parent) {
+		node->accounting = parent->accounting;
+	}
+
 	VALGRIND_CREATE_MEMPOOL(node, 0, false);
 
 	/* Return to type-specific creation routine to finish up */
diff --git a/src/include/nodes/memnodes.h b/src/include/nodes/memnodes.h
index ad77509..a6d9f8c 100644
--- a/src/include/nodes/memnodes.h
+++ b/src/include/nodes/memnodes.h
@@ -50,6 +50,18 @@ typedef struct MemoryContextMethods
 #endif
 } MemoryContextMethods;
 
+typedef struct MemoryAccountingData {
+
+	Size	total_allocated; /* including child contexts */
+	Size	self_allocated;  /* not including child contexts */
+
+	/* parent accounting (not parent context) */
+	struct MemoryAccountingData * parent;
+
+} MemoryAccountingData;
+
+typedef MemoryAccountingData * MemoryAccounting;
+
 
 typedef struct MemoryContextData
 {
@@ -60,6 +72,8 @@ typedef struct MemoryContextData
 	MemoryContext nextchild;	/* next child of same parent */
 	char	   *name;			/* context name (just for debugging) */
 	bool		isReset;		/* T = no space alloced since last reset */
+	bool		track_mem;		/* whether to track memory usage */
+	MemoryAccounting	accounting;
 #ifdef USE_ASSERT_CHECKING
 	bool		allowInCritSection;	/* allow palloc in critical section */
 #endif
diff --git a/src/include/utils/memutils.h b/src/include/utils/memutils.h
index 2fede86..b82f923 100644
--- a/src/include/utils/memutils.h
+++ b/src/include/utils/memutils.h
@@ -96,6 +96,7 @@ extern void MemoryContextDeleteChildren(MemoryContext context);
 extern void MemoryContextResetAndDeleteChildren(MemoryContext context);
 extern void MemoryContextSetParent(MemoryContext context,
 					   MemoryContext new_parent);
+extern Size MemoryContextGetAllocated(MemoryContext context, bool total);
 extern Size GetMemoryChunkSpace(void *pointer);
 extern MemoryContext GetMemoryChunkContext(void *pointer);
 extern MemoryContext MemoryContextGetParent(MemoryContext context);
@@ -117,7 +118,8 @@ extern bool MemoryContextContains(MemoryContext context, void *pointer);
 extern MemoryContext MemoryContextCreate(NodeTag tag, Size size,
 					MemoryContextMethods *methods,
 					MemoryContext parent,
-					const char *name);
+					const char *name,
+					bool track_mem);
 
 
 /*
@@ -130,6 +132,12 @@ extern MemoryContext AllocSetContextCreate(MemoryContext parent,
 					  Size minContextSize,
 					  Size initBlockSize,
 					  Size maxBlockSize);
+extern MemoryContext AllocSetContextCreateTracked(MemoryContext parent,
+					  const char *name,
+					  Size minContextSize,
+					  Size initBlockSize,
+					  Size maxBlockSize,
+					  bool track_mem);
 
 /*
  * Recommended default alloc parameters, suitable for "ordinary" contexts

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 49547ee..b651858 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -2884,6 +2884,21 @@ include_dir 'conf.d'
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-enable-hashagg-disk" xreflabel="enable_hashagg_disk">
+      <term><varname>enable_hashagg_disk</varname> (<type>boolean</type>)
+      <indexterm>
+       <primary><varname>enable_hashagg_disk</> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Enables or disables the query planner's use of hashed aggregation plan
+        types when the planner expects the hash table size to exceed
+        <varname>work_mem</varname>. The default is <literal>on</>.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-enable-hashjoin" xreflabel="enable_hashjoin">
       <term><varname>enable_hashjoin</varname> (<type>boolean</type>)
       <indexterm>
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 6455864..3ae9583 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -108,6 +108,7 @@
 #include "optimizer/tlist.h"
 #include "parser/parse_agg.h"
 #include "parser/parse_coerce.h"
+#include "storage/buffile.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
@@ -115,7 +116,11 @@
 #include "utils/syscache.h"
 #include "utils/tuplesort.h"
 #include "utils/datum.h"
+#include "utils/dynahash.h"
 
+#define HASH_DISK_MIN_PARTITIONS		1
+#define HASH_DISK_DEFAULT_PARTITIONS	4
+#define HASH_DISK_MAX_PARTITIONS		256
 
 /*
  * AggStatePerAggData - per-aggregate working state for the Agg scan
@@ -310,6 +315,24 @@ typedef struct AggHashEntryData
 
 }	AggHashEntryData;	/* VARIABLE LENGTH STRUCT */
 
+/*
+ * Used as a unit of work when batching. After reaching work_mem, no new
+ * groups are added to the hash table, and the tuples are divided into
+ * multiple batches (using a range of bits from the hash value).
+ * 
+ * At the end, each output partition (represented by a temporary file)
+ * is converted into a new HashWork item and the process is repeated.
+ */
+typedef struct HashWork
+{
+	BufFile		 *input_file;	/* input partition, NULL for outer plan */
+	int			  input_bits;	/* number of bits for input partition mask */
+
+	int			  n_output_partitions; /* number of output partitions */
+	BufFile		**output_partitions; /* output partition files */
+	int			 *output_ntuples; /* number of tuples in each partition */
+	int			  output_bits; /* log2(n_output_partitions) + input_bits */
+} HashWork;
 
 static void initialize_aggregates(AggState *aggstate,
 					  AggStatePerAgg peragg,
@@ -331,10 +354,11 @@ static void finalize_aggregate(AggState *aggstate,
 static Bitmapset *find_unaggregated_cols(AggState *aggstate);
 static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
 static void build_hash_table(AggState *aggstate, Size tuple_width);
-static AggHashEntry lookup_hash_entry(AggState *aggstate,
-				  TupleTableSlot *inputslot);
+static AggHashEntry lookup_hash_entry(AggState *aggstate, HashWork * work,
+					uint32 hashvalue, TupleTableSlot *inputslot);
+static HashWork *hash_work(BufFile *input_file, int input_bits);
 static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
-static void agg_fill_hash_table(AggState *aggstate);
+static bool agg_fill_hash_table(AggState *aggstate);
 static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
 static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
 
@@ -348,6 +372,12 @@ static void reset_hash_table(AggHashTable htab);
 static void IteratorReset(AggHashTable htab);
 static AggHashEntry IteratorGetNext(AggHashTable htab);
 
+static TupleTableSlot *
+read_saved_tuple(BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot);
+static void
+save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
+		   uint32 hashvalue);
+
 /*
  * The size of the chunks for dense allocation. This needs to be >8kB
  * because the default (and only) memory context implementation uses
@@ -412,6 +442,7 @@ typedef struct AggHashTableData
 	 */
 	HashChunk		cur_chunk;
 	AggHashEntry	cur_entry;
+	int				niterated;
 
 	/* list of chunks with dense-packed entries / minimal tuples */
 	HashChunk		chunks_hash;
@@ -1096,12 +1127,15 @@ build_hash_table(AggState *aggstate, Size tuple_width)
 	htab = (AggHashTable)MemoryContextAllocZero(aggstate->aggcontext,
 											sizeof(AggHashTableData));
 
+	htab->niterated = 0;
+
 	/* TODO create a memory context for the hash table */
-	htab->htabctx = AllocSetContextCreate(aggstate->aggcontext,
+	htab->htabctx = AllocSetContextCreateTracked(aggstate->aggcontext,
 											"HashAggHashTable",
 											ALLOCSET_DEFAULT_MINSIZE,
 											ALLOCSET_DEFAULT_INITSIZE,
-											ALLOCSET_DEFAULT_MAXSIZE);
+											ALLOCSET_DEFAULT_MAXSIZE,
+											true);
 
 	/* buckets are just pointers to AggHashEntryData structures */
 	htab->buckets = (AggHashEntry*)MemoryContextAllocZero(htab->htabctx,
@@ -1198,15 +1232,14 @@ hash_agg_entry_size(int numAggs)
  * When called, CurrentMemoryContext should be the per-query context.
  */
 static AggHashEntry
-lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
+lookup_hash_entry(AggState *aggstate, HashWork * work, uint32 hashvalue,
+TupleTableSlot *inputslot)
 {
 
 	AggHashEntry entry = NULL;
-	uint32		hashvalue;
 	uint32		bucketno;
 	MinimalTuple mintuple;
 
-	hashvalue = compute_hash_value(aggstate, inputslot);
 	bucketno = compute_bucket(aggstate, hashvalue);
 
 	entry = aggstate->hashtable->buckets[bucketno];
@@ -1223,10 +1256,13 @@ lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
 		entry = entry->next;
 	}
 
-	/* There's not a maching entry in the bucket, so create a new one and
-	 * copy in data both for the aggregates, and the MinimalTuple containing
-	 * keys for the group columns. */
-	if (entry == NULL)
+	/*
+	 * There's not a maching entry in the bucket (and we've not reached the
+	 * work_mem limit, so create a new one and copy in data both for the
+	 * aggregates, and the MinimalTuple containing keys for the group columns.
+	 */
+	if ((entry == NULL) &&
+		(MemoryContextGetAllocated(aggstate->hashtable->htabctx, true) < work_mem * 1024L))
 	{
 
 		MemoryContext old;
@@ -1318,9 +1354,16 @@ ExecAgg(AggState *node)
 	/* Dispatch based on strategy */
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
-		if (!node->table_filled)
-			agg_fill_hash_table(node);
-		return agg_retrieve_hash_table(node);
+		TupleTableSlot *slot = NULL;
+
+		while (slot == NULL)
+		{
+			if (!node->table_filled)
+				if (! agg_fill_hash_table(node))
+					break;	/* no more HashWork items to process */
+			slot = agg_retrieve_hash_table(node);
+		}
+		return slot;
 	}
 	else
 		return agg_retrieve_direct(node);
@@ -1536,13 +1579,15 @@ agg_retrieve_direct(AggState *aggstate)
 /*
  * ExecAgg for hashed case: phase 1, read input and build hash table
  */
-static void
+static bool
 agg_fill_hash_table(AggState *aggstate)
 {
 	PlanState  *outerPlan;
 	ExprContext *tmpcontext;
 	AggHashEntry entry;
-	TupleTableSlot *outerslot;
+	TupleTableSlot *outerslot = NULL;
+	HashWork   *work;
+	int			i;
 
 	/*
 	 * get state info from node
@@ -1551,33 +1596,120 @@ agg_fill_hash_table(AggState *aggstate)
 	/* tmpcontext is the per-input-tuple expression context */
 	tmpcontext = aggstate->tmpcontext;
 
+	/* if there's no HashWork item, we're done */
+	if (aggstate->hash_work == NIL)
+	{
+		aggstate->agg_done = true;
+		return false;
+	}
+
+	work = linitial(aggstate->hash_work);
+	aggstate->hash_work = list_delete_first(aggstate->hash_work);
+
+	/* if not the first time through, reinitialize */
+	if (!aggstate->hash_init_state)
+	{
+		/* FIXME get rid of all the previous aggregate states somehow.
+		 *       Either reset the aggcontext, or clear the hash table
+		 *       somehow. Resetting the context seems better. */
+
+		/* reset the hash table (free the chunks, zero buckets) */
+		reset_hash_table(aggstate->hashtable);
+	}
+
+	/* reinitialize on the next item */
+	aggstate->hash_init_state = false;
+
 	/*
 	 * Process each outer-plan tuple, and then fetch the next one, until we
 	 * exhaust the outer plan.
 	 */
 	for (;;)
 	{
-		outerslot = ExecProcNode(outerPlan);
-		if (TupIsNull(outerslot))
-			break;
-		/* set up for advance_aggregates call */
-		tmpcontext->ecxt_outertuple = outerslot;
+
+		uint32 hashvalue;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* the first HashWork item means we need to fetch the tuples */
+		if (work->input_file == NULL)
+		{
+			outerslot = ExecProcNode(outerPlan);
+			if (TupIsNull(outerslot))
+				break;
+
+			hashvalue = compute_hash_value(aggstate, outerslot);
+		}
+		else
+		{
+			/* first time through this HashWork item */
+			if (outerslot == NULL)
+				outerslot = MakeSingleTupleTableSlot(aggstate->hashtable->slot->tts_tupleDescriptor);
+
+			outerslot = read_saved_tuple(work->input_file, &hashvalue, outerslot);
+			if (TupIsNull(outerslot))
+			{
+				BufFileClose(work->input_file);
+				work->input_file = NULL;
+				break;
+			}
+		}
 
 		/* Find or build hashtable entry for this tuple's group */
-		entry = lookup_hash_entry(aggstate, outerslot);
+		entry = lookup_hash_entry(aggstate, work, hashvalue, outerslot);
 
-		/* Advance the aggregates */
-		advance_aggregates(aggstate, entry->pergroup);
+		if (entry != NULL) {
 
-		/* Reset per-input-tuple context after each tuple */
-		ResetExprContext(tmpcontext);
+			/* set up for advance_aggregates call */
+			tmpcontext->ecxt_outertuple = outerslot;
+
+			/* Advance the aggregates */
+			advance_aggregates(aggstate, entry->pergroup);
+
+			/* Reset per-input-tuple context after each tuple */
+			ResetExprContext(tmpcontext);
+
+		} else {
+
+			/* no entry for this tuple, and we've reached work_mem */
+			save_tuple(aggstate, work, outerslot, hashvalue);
+
+		}
 	}
 
+	/* add each output partition as a new work item */
+	for (i = 0; i < work->n_output_partitions; i++)
+	{
+		BufFile			*file = work->output_partitions[i];
+		MemoryContext	 oldContext;
+
+		/* partition is empty */
+		if (work->output_ntuples[i] == 0)
+			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->aggcontext);
+		aggstate->hash_work = lappend(aggstate->hash_work,
+									  hash_work(file,
+												work->output_bits + work->input_bits));
+		MemoryContextSwitchTo(oldContext);
+	}
+
+	pfree(work);
+
 	aggstate->table_filled = true;
 
 	/* Initialize for iteration through the table (first bucket / entry) */
 	IteratorReset(aggstate->hashtable);
 
+	/* ready to return groups from this hash table */
+	return true;
+
 }
 
 /*
@@ -1620,6 +1752,8 @@ agg_retrieve_hash_table(AggState *aggstate)
 		 */
 		ResetExprContext(econtext);
 
+		htab->niterated += 1;
+
 		/*
 		* Store the copied first input tuple in the tuple table slot reserved
 		* for it, so that it can be used in ExecProject.
@@ -1677,7 +1811,8 @@ agg_retrieve_hash_table(AggState *aggstate)
 
 	}
 
-	aggstate->agg_done = true;
+	/* No more entries in hashtable, so done with this batch */
+	aggstate->table_filled = false;
 
 	/* No more groups */
 	return NULL;
@@ -1739,11 +1874,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	 * recover no-longer-wanted space.
 	 */
 	aggstate->aggcontext =
-		AllocSetContextCreate(CurrentMemoryContext,
+		AllocSetContextCreateTracked(CurrentMemoryContext,
 							  "AggContext",
 							  ALLOCSET_DEFAULT_MINSIZE,
 							  ALLOCSET_DEFAULT_INITSIZE,
-							  ALLOCSET_DEFAULT_MAXSIZE);
+							  ALLOCSET_DEFAULT_MAXSIZE, true);
 
 	/*
 	 * tuple table initialization
@@ -1842,10 +1977,22 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 
 	if (node->aggstrategy == AGG_HASHED)
 	{
+		MemoryContext oldContext;
+
 		build_hash_table(aggstate, outerPlan->plan_width);
 		aggstate->table_filled = false;
+		aggstate->hash_init_state = true;
+		aggstate->hash_disk = false;
+
 		/* Compute the columns we actually need to hash on */
 		aggstate->hash_needed = find_hash_columns(aggstate);
+
+		/* prime with initial work item to read from outer plan */
+		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
+		aggstate->hash_work = lappend(aggstate->hash_work,
+									  hash_work(NULL, 0));
+		MemoryContextSwitchTo(oldContext);
+
 	}
 	else
 	{
@@ -2264,22 +2411,23 @@ ExecReScanAgg(AggState *node)
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
 		/*
-		 * In the hashed case, if we haven't yet built the hash table then we
-		 * can just return; nothing done yet, so nothing to undo. If subnode's
-		 * chgParam is not NULL then it will be re-scanned by ExecProcNode,
-		 * else no reason to re-scan it at all.
+		 * In the hashed case, if we haven't done any execution work yet, we
+		 * can just return; nothing to undo. If subnode's chgParam is not NULL
+		 * then it will be re-scanned by ExecProcNode, else no reason to
+		 * re-scan it at all.
 		 */
-		if (!node->table_filled)
+		if (node->hash_init_state)
 			return;
 
 		/*
-		 * If we do have the hash table and the subplan does not have any
-		 * parameter changes, then we can just rescan the existing hash table;
-		 * no need to build it again.
+		 * If we do have the hash table, it never went to disk, and the
+		 * subplan does not have any parameter changes, then we can just
+		 * rescan the existing hash table; no need to build it again.
 		 */
-		if (node->ss.ps.lefttree->chgParam == NULL)
+		if (node->ss.ps.lefttree->chgParam == NULL && !node->hash_disk)
 		{
 			IteratorReset(node->hashtable);
+			node->table_filled = true;
 			return;
 		}
 	}
@@ -2318,10 +2466,21 @@ ExecReScanAgg(AggState *node)
 
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
+		MemoryContext oldContext;
 		Plan * outerPlan = outerPlan((Agg *) node->ss.ps.plan);
+
 		/* Rebuild an empty hash table */
 		build_hash_table(node, outerPlan->plan_width);
+		node->hash_init_state = true;
 		node->table_filled = false;
+		node->hash_disk = false;
+		node->hash_work = NIL;
+
+		/* prime with initial work item to read from outer plan */
+		oldContext = MemoryContextSwitchTo(node->aggcontext);
+		node->hash_work = lappend(node->hash_work,
+								  hash_work(NULL, 0));
+		MemoryContextSwitchTo(oldContext);
 	}
 	else
 	{
@@ -2827,6 +2986,144 @@ AggHashEntry IteratorGetNext(AggHashTable htab)
 }
 
 /*
+ * hash_work
+ *
+ * Construct a HashWork item, which represents one iteration of HashAgg to be
+ * done. Should be called in the aggregate's memory context.
+ */
+static HashWork *
+hash_work(BufFile *input_file, int input_bits)
+{
+	HashWork *work = palloc(sizeof(HashWork));
+
+	work->input_file = input_file;
+	work->input_bits = input_bits;
+
+	/*
+	 * Will be set only if we run out of memory and need to partition an
+	 * additional level.
+	 */
+	work->n_output_partitions = 0;
+	work->output_partitions = NULL;
+	work->output_ntuples = NULL;
+	work->output_bits = 0;
+
+	return work;
+}
+
+/*
+ * save_tuple
+ *
+ * Not enough memory to add tuple as new entry in hash table. Save for later
+ * in the appropriate partition.
+ */
+static void
+save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
+		   uint32 hashvalue)
+{
+	int					 partition;
+	MinimalTuple		 tuple;
+	BufFile				*file;
+	int					 written;
+
+	if (work->output_partitions == NULL)
+	{
+		int npartitions = HASH_DISK_DEFAULT_PARTITIONS; //TODO choose
+		int partition_bits;
+		int i;
+
+		if (npartitions < HASH_DISK_MIN_PARTITIONS)
+			npartitions = HASH_DISK_MIN_PARTITIONS;
+		if (npartitions > HASH_DISK_MAX_PARTITIONS)
+			npartitions = HASH_DISK_MAX_PARTITIONS;
+
+		partition_bits = my_log2(npartitions);
+
+		/* make sure that we don't exhaust the hash bits */
+		if (partition_bits + work->input_bits >= 32)
+			partition_bits = 32 - work->input_bits;
+
+		/* number of partitions will be a power of two */
+		npartitions = 1L << partition_bits;
+
+		work->output_bits = partition_bits;
+		work->n_output_partitions = npartitions;
+		work->output_partitions = palloc(sizeof(BufFile *) * npartitions);
+		work->output_ntuples = palloc0(sizeof(int) * npartitions);
+
+		for (i = 0; i < npartitions; i++)
+			work->output_partitions[i] = BufFileCreateTemp(false);
+	}
+
+	if (work->output_bits == 0)
+		partition = 0;
+	else
+		partition = (hashvalue << work->input_bits) >>
+			(32 - work->output_bits);
+
+	work->output_ntuples[partition]++;
+	file = work->output_partitions[partition];
+	tuple = ExecFetchSlotMinimalTuple(slot);
+
+	written = BufFileWrite(file, (void *) &hashvalue, sizeof(uint32));
+	if (written != sizeof(uint32))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not write to HashAgg temporary file: %m")));
+
+	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")));
+}
+
+
+/*
+ * read_saved_tuple
+ *		read the next tuple from a batch file.  Return NULL if no more.
+ *
+ * On success, *hashvalue is set to the tuple's hash value, and the tuple
+ * itself is stored in the given slot.
+ *
+ * Copied with minor modifications from ExecHashJoinGetSavedTuple.
+ */
+static TupleTableSlot *
+read_saved_tuple(BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot)
+{
+	uint32		header[2];
+	size_t		nread;
+	MinimalTuple tuple;
+
+	/*
+	 * Since both the hash value and the MinimalTuple length word are uint32,
+	 * we can read them both in one BufFileRead() call without any type
+	 * cheating.
+	 */
+	nread = BufFileRead(file, (void *) header, sizeof(header));
+	if (nread == 0)				/* end of file */
+	{
+		ExecClearTuple(tupleSlot);
+		return NULL;
+	}
+	if (nread != sizeof(header))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+	*hashvalue = header[0];
+	tuple = (MinimalTuple) palloc(header[1]);
+	tuple->t_len = header[1];
+	nread = BufFileRead(file,
+						(void *) ((char *) tuple + sizeof(uint32)),
+						header[1] - sizeof(uint32));
+	if (nread != header[1] - sizeof(uint32))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+	return ExecStoreMinimalTuple(tuple, tupleSlot, true);
+}
+
+/*
  * Resets the contents of the hash table - removes all the entries and
  * tuples, but keeps the 'size' of the hash table (nbuckets).
  */
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 0cdb790..926abad 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -113,6 +113,7 @@ bool		enable_bitmapscan = true;
 bool		enable_tidscan = true;
 bool		enable_sort = true;
 bool		enable_hashagg = true;
+bool		enable_hashagg_disk = true;
 bool		enable_nestloop = true;
 bool		enable_material = true;
 bool		enable_mergejoin = true;
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index e1480cd..7b8135d 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -2741,7 +2741,8 @@ choose_hashed_grouping(PlannerInfo *root,
 	/* plus the per-hash-entry overhead */
 	hashentrysize += hash_agg_entry_size(agg_costs->numAggs);
 
-	if (hashentrysize * dNumGroups > work_mem * 1024L)
+	if (!enable_hashagg_disk &&
+		hashentrysize * dNumGroups > work_mem * 1024L)
 		return false;
 
 	/*
@@ -2907,7 +2908,8 @@ choose_hashed_distinct(PlannerInfo *root,
 	/* plus the per-hash-entry overhead */
 	hashentrysize += hash_agg_entry_size(0);
 
-	if (hashentrysize * dNumDistinctRows > work_mem * 1024L)
+	if (!enable_hashagg_disk &&
+		hashentrysize * dNumDistinctRows > work_mem * 1024L)
 		return false;
 
 	/*
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 8c57803..5128e20 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -749,6 +749,15 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 	{
+		{"enable_hashagg_disk", PGC_USERSET, QUERY_TUNING_METHOD,
+			gettext_noop("Enables the planner's use of disk-based hashed aggregation plans."),
+			NULL
+		},
+		&enable_hashagg_disk,
+		true,
+		NULL, NULL, NULL
+	},
+	{
 		{"enable_material", PGC_USERSET, QUERY_TUNING_METHOD,
 			gettext_noop("Enables the planner's use of materialization."),
 			NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index df98b02..8f5b73b 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -266,6 +266,7 @@
 
 #enable_bitmapscan = on
 #enable_hashagg = on
+#enable_hashagg_disk = on
 #enable_hashjoin = on
 #enable_indexscan = on
 #enable_indexonlyscan = on
diff --git a/src/backend/utils/mmgr/mcxt.c b/src/backend/utils/mmgr/mcxt.c
index a70b296..97034f1 100644
--- a/src/backend/utils/mmgr/mcxt.c
+++ b/src/backend/utils/mmgr/mcxt.c
@@ -634,7 +634,7 @@ MemoryContextCreate(NodeTag tag, Size size,
 	 */
 	if (track_mem)
 	{
-		node->accounting = (MemoryAccounting)MemoryContextAlloc(TopMemoryContext,
+		node->accounting = (MemoryAccounting)MemoryContextAllocZero(TopMemoryContext,
 												sizeof(MemoryAccountingData));
 		if (parent)
 			node->accounting->parent = parent->accounting;
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 995389b..1a61ac7 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1726,6 +1726,11 @@ typedef struct AggState
 	bool		table_filled;	/* hash table filled yet? */
 	AggHashTable	hashtable;	/* instance of the simple hash table */
 
+	/* simple batching */
+	bool		hash_init_state; /* in initial state before execution? */
+	bool		hash_disk;		/* have we exceeded memory yet? */
+	List	   *hash_work;		/* remaining work to be done */
+
 } AggState;
 
 /* ----------------
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index 75e2afb..d363e65 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -57,6 +57,7 @@ extern bool enable_bitmapscan;
 extern bool enable_tidscan;
 extern bool enable_sort;
 extern bool enable_hashagg;
+extern bool enable_hashagg_disk;
 extern bool enable_nestloop;
 extern bool enable_material;
 extern bool enable_mergejoin;
diff --git a/src/test/regress/expected/rangefuncs.out b/src/test/regress/expected/rangefuncs.out
index 774e75e..e88c83c 100644
--- a/src/test/regress/expected/rangefuncs.out
+++ b/src/test/regress/expected/rangefuncs.out
@@ -3,6 +3,7 @@ SELECT name, setting FROM pg_settings WHERE name LIKE 'enable%';
 ----------------------+---------
  enable_bitmapscan    | on
  enable_hashagg       | on
+ enable_hashagg_disk  | on
  enable_hashjoin      | on
  enable_indexonlyscan | on
  enable_indexscan     | on
@@ -12,7 +13,7 @@ SELECT name, setting FROM pg_settings WHERE name LIKE 'enable%';
  enable_seqscan       | on
  enable_sort          | on
  enable_tidscan       | on
-(11 rows)
+(12 rows)
 
 CREATE TABLE foo2(fooid int, f2 int);
 INSERT INTO foo2 VALUES(1, 11);

diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 781a736..54caade 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -78,6 +78,8 @@ static void show_merge_append_keys(MergeAppendState *mstate, List *ancestors,
 					   ExplainState *es);
 static void show_agg_keys(AggState *astate, List *ancestors,
 			  ExplainState *es);
+static void show_agg_batching(AggState *astate, List *ancestors,
+			  ExplainState *es);
 static void show_group_keys(GroupState *gstate, List *ancestors,
 				ExplainState *es);
 static void show_sort_group_keys(PlanState *planstate, const char *qlabel,
@@ -1391,6 +1393,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
 										   planstate, es);
 			break;
 		case T_Agg:
+			show_agg_batching((AggState *) planstate, ancestors, es);
 			show_agg_keys((AggState *) planstate, ancestors, es);
 			show_upper_qual(plan->qual, "Filter", planstate, ancestors, es);
 			if (plan->qual)
@@ -1790,6 +1793,36 @@ show_agg_keys(AggState *astate, List *ancestors,
 }
 
 /*
+ * Show the batching info for an Agg node.
+ */
+static void
+show_agg_batching(AggState *astate, List *ancestors,
+			  ExplainState *es)
+{
+	Agg		   *plan = (Agg *) astate->ss.ps.plan;
+
+	if ((es->analyze) && (plan->aggstrategy == AGG_HASHED))
+	{
+		if (es->format == EXPLAIN_FORMAT_TEXT)
+		{
+			appendStringInfoSpaces(es->str, es->indent * 2);
+			appendStringInfo(es->str, "Batch Count: %d  Rebatches: %d  Smallest: %ldkB  Largest: %ldkB Rescanned: %.0f%%\n",
+							 astate->batch_count, astate->rebatch_count,
+							 astate->batch_min_size / 1024,
+							 astate->batch_max_size / 1024,
+							 astate->ntuples_rescanned * 100.0 / astate->ntuples_scanned);
+		}
+		else
+		{
+			ExplainPropertyLong("Batch Count", astate->batch_count, es);
+			ExplainPropertyLong("Batch Smallest", astate->batch_min_size/1024, es);
+			ExplainPropertyLong("Batch Largest", astate->batch_max_size/1024, es);
+			ExplainPropertyLong("Batch Rescan Rate", (astate->ntuples_rescanned * 100) / astate->ntuples_scanned, es);
+		}
+	}
+}
+
+/*
  * Show the grouping keys for a Group node.
  */
 static void
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 3ae9583..1266faf 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -118,7 +118,7 @@
 #include "utils/datum.h"
 #include "utils/dynahash.h"
 
-#define HASH_DISK_MIN_PARTITIONS		1
+#define HASH_DISK_MIN_PARTITIONS		2
 #define HASH_DISK_DEFAULT_PARTITIONS	4
 #define HASH_DISK_MAX_PARTITIONS		256
 
@@ -328,6 +328,7 @@ typedef struct HashWork
 	BufFile		 *input_file;	/* input partition, NULL for outer plan */
 	int			  input_bits;	/* number of bits for input partition mask */
 
+	double		  ntuples_expected; /* number of tuples expected on input */
 	int			  n_output_partitions; /* number of output partitions */
 	BufFile		**output_partitions; /* output partition files */
 	int			 *output_ntuples; /* number of tuples in each partition */
@@ -356,7 +357,7 @@ static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
 static void build_hash_table(AggState *aggstate, Size tuple_width);
 static AggHashEntry lookup_hash_entry(AggState *aggstate, HashWork * work,
 					uint32 hashvalue, TupleTableSlot *inputslot);
-static HashWork *hash_work(BufFile *input_file, int input_bits);
+static HashWork *hash_work(BufFile *input_file, int input_bits, double ntuples_expected);
 static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
 static bool agg_fill_hash_table(AggState *aggstate);
 static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
@@ -376,7 +377,7 @@ static TupleTableSlot *
 read_saved_tuple(BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot);
 static void
 save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
-		   uint32 hashvalue);
+		   uint32 hashvalue, int64 ntuples_current);
 
 /*
  * The size of the chunks for dense allocation. This needs to be >8kB
@@ -1588,6 +1589,8 @@ agg_fill_hash_table(AggState *aggstate)
 	TupleTableSlot *outerslot = NULL;
 	HashWork   *work;
 	int			i;
+	int64		ntuples = 0;
+	Size		allocated;
 
 	/*
 	 * get state info from node
@@ -1639,6 +1642,7 @@ agg_fill_hash_table(AggState *aggstate)
 				break;
 
 			hashvalue = compute_hash_value(aggstate, outerslot);
+			aggstate->ntuples_scanned += 1;
 		}
 		else
 		{
@@ -1653,8 +1657,12 @@ agg_fill_hash_table(AggState *aggstate)
 				work->input_file = NULL;
 				break;
 			}
+
+			aggstate->ntuples_rescanned += 1;
 		}
 
+		ntuples++;
+
 		/* Find or build hashtable entry for this tuple's group */
 		entry = lookup_hash_entry(aggstate, work, hashvalue, outerslot);
 
@@ -1672,11 +1680,81 @@ agg_fill_hash_table(AggState *aggstate)
 		} else {
 
 			/* no entry for this tuple, and we've reached work_mem */
-			save_tuple(aggstate, work, outerslot, hashvalue);
+			save_tuple(aggstate, work, outerslot, hashvalue, ntuples);
 
 		}
 	}
 
+	/*
+	 * XXX Idea on estimating the number of partitions necessary in the
+	 *     next step, based on estimating group state size etc.
+	 *
+	 * See how much memory is allocated in the memory context, so that
+	 * we can use it to compute average state size (including ovehead).
+	 * When doing the initial partitioning, we did it when reaching
+	 * work_mem, but the states may have grown further.
+	 *
+	 * Assuming the groups are not of wildly different size, we can
+	 * optimize the partitioning in the following work items like this:
+	 *
+	 * 1) computing average group size
+	 *
+	 *   avg_group_size = allocated_bytes / nentries
+	 *
+	 * 2) computing number of groups that fit into work_mem
+	 *
+	 *   groups_work_mem = (work_mem * 1024L) / avg_group_size
+	 *
+	 * 3) computing tuples per group
+	 *
+	 *   ntuples_per_group = (ntuples / nentries)
+	 *
+	 * 4) computing ntuples_in_partition
+	 *
+	 *   optimal_partition_tuples = ntuples_per_group * groups_work_mem
+	 *
+	 * 5) we know how many tuples we wrote into each partition - we can
+	 *    either compute it as (ntuples/npartitions) which is easy, or
+	 *    track the number per partition (more correct), so we can
+	 *    decide into how many partitions should we split it in the next
+	 *    step
+	 *
+	 *     npartitions = ntuples_per_partition / optimal_partition_tuples
+	 *
+	 *    and we can do this immediately at the beginning, using the
+	 *    hash value (assuming it's a power of 2).
+	 *
+	 * This should minimize the number of times a tuple is read from the
+	 * temporary file, only to be written again because there's not enough
+	 * free memory.
+	 *
+	 * The question is how this will deal with exceptionally large
+	 * groups. Technically, all partitions should receive about the
+	 * same number of groups, but if there's a very frequent group the
+	 * partition may be much larger (many more tuples, belonging to the
+	 * very large group). What we need to prevent is splitting the data
+	 * into needlessly small partitions.
+	 */
+
+	allocated = MemoryContextGetAllocated(aggstate->hashtable->htabctx, true);
+
+	/* keep track of the largest/smallest batch size */
+	if (aggstate->batch_count == 1)
+	{
+		aggstate->batch_min_size = allocated;
+		aggstate->batch_max_size = allocated;
+	}
+	else
+	{
+		if (allocated < aggstate->batch_min_size)
+			aggstate->batch_min_size = allocated;
+		if (allocated > aggstate->batch_max_size)
+			aggstate->batch_max_size = allocated;
+	}
+
+	if (work->n_output_partitions > 0)
+		aggstate->rebatch_count += 1;
+
 	/* add each output partition as a new work item */
 	for (i = 0; i < work->n_output_partitions; i++)
 	{
@@ -1693,11 +1771,27 @@ agg_fill_hash_table(AggState *aggstate)
 					(errcode_for_file_access(),
 					 errmsg("could not rewind HashAgg temporary file: %m")));
 
+		/*
+		 * XXX This assumes all the batches are equally sized, but that
+		 * can easily not be the case - imagine a frequent group. All the
+		 * tuples will get saved into the same batch, thus making it
+		 * much larger than the rest.
+		 *
+		 * At this point, we can also estimate the average group size as
+		 * (allocated memory / nentries), and number of tuples per group
+		 * (ntuples / nentries). Granted, those are pretty rough estimates
+		 * that can go wrong in many ways, but it's better than nothing.
+		 *
+		 * TODO Keep track of number of tuples saved in each group.
+		 */
 		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
-		aggstate->hash_work = lappend(aggstate->hash_work,
-									  hash_work(file,
-												work->output_bits + work->input_bits));
+		aggstate->hash_work
+			= lappend(aggstate->hash_work,
+					  hash_work(file,
+					  work->output_bits + work->input_bits,
+					  work->output_ntuples[i]));
 		MemoryContextSwitchTo(oldContext);
+		aggstate->batch_count += 1;
 	}
 
 	pfree(work);
@@ -1984,13 +2078,19 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 		aggstate->hash_init_state = true;
 		aggstate->hash_disk = false;
 
+		/* explain (analyze) counters */
+		aggstate->batch_count = 1;
+		aggstate->rebatch_count = 0;
+		aggstate->ntuples_scanned = 0;
+		aggstate->ntuples_rescanned = 0;
+
 		/* Compute the columns we actually need to hash on */
 		aggstate->hash_needed = find_hash_columns(aggstate);
 
 		/* prime with initial work item to read from outer plan */
 		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
 		aggstate->hash_work = lappend(aggstate->hash_work,
-									  hash_work(NULL, 0));
+									  hash_work(NULL, 0, outerPlan->plan_rows));
 		MemoryContextSwitchTo(oldContext);
 
 	}
@@ -2467,7 +2567,7 @@ ExecReScanAgg(AggState *node)
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
 		MemoryContext oldContext;
-		Plan * outerPlan = outerPlan((Agg *) node->ss.ps.plan);
+		Plan *outerPlan = outerPlan((Agg *) node->ss.ps.plan);
 
 		/* Rebuild an empty hash table */
 		build_hash_table(node, outerPlan->plan_width);
@@ -2476,10 +2576,16 @@ ExecReScanAgg(AggState *node)
 		node->hash_disk = false;
 		node->hash_work = NIL;
 
+		/* explain (analyze) counters */
+		node->batch_count = 1;
+		node->rebatch_count = 0;
+		node->ntuples_scanned = 0;
+		node->ntuples_rescanned = 0;
+
 		/* prime with initial work item to read from outer plan */
 		oldContext = MemoryContextSwitchTo(node->aggcontext);
 		node->hash_work = lappend(node->hash_work,
-								  hash_work(NULL, 0));
+								  hash_work(NULL, 0, outerPlan->plan_rows));
 		MemoryContextSwitchTo(oldContext);
 	}
 	else
@@ -2992,12 +3098,13 @@ AggHashEntry IteratorGetNext(AggHashTable htab)
  * done. Should be called in the aggregate's memory context.
  */
 static HashWork *
-hash_work(BufFile *input_file, int input_bits)
+hash_work(BufFile *input_file, int input_bits, double ntuples_expected)
 {
 	HashWork *work = palloc(sizeof(HashWork));
 
 	work->input_file = input_file;
 	work->input_bits = input_bits;
+	work->ntuples_expected = ntuples_expected;
 
 	/*
 	 * Will be set only if we run out of memory and need to partition an
@@ -3019,7 +3126,7 @@ hash_work(BufFile *input_file, int input_bits)
  */
 static void
 save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
-		   uint32 hashvalue)
+		   uint32 hashvalue, int64 ntuples_current)
 {
 	int					 partition;
 	MinimalTuple		 tuple;
@@ -3028,7 +3135,35 @@ save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
 
 	if (work->output_partitions == NULL)
 	{
-		int npartitions = HASH_DISK_DEFAULT_PARTITIONS; //TODO choose
+		/*
+		 * We will choose the number of partitions based on when we
+		 * reached full work_mem. We expect to get work->ntuples in
+		 * total, and we've reached work_mem at ntuples, so we expect
+		 * to get (work->ntuples - ntuples) additional tuples.
+		 * 
+		 * By assuming to reach work_mem every 'ntuples' tuples, we can
+		 * estimate the number of batches as
+		 * 
+		 *    ceil((work->ntuples - ntuples)/ntuples)
+		 * 
+		 * which is actually
+		 * 
+		 *    (int)(1 + (work->ntuples - ntuples)/ntuples)
+		 * 
+		 * and that's just (work->ntuples/ntuples). We'll impose some
+		 * basic safety numbers on those values.
+		 * 
+		 * It's probably better to over-estimate here, under-estimation
+		 * means we'll have to read the tuples and write then again,
+		 * into another set of batches. Not really efficient.
+		 *
+		 * Also, even if there's a huge group in this batch, the state
+		 * size usually grows along with the number of tuples passed
+		 * to the transition function. So even in this case it should
+		 * be a good estimate (e.g. the batching should not be triggered
+		 * too early).
+		 */
+		int npartitions = (work->ntuples_expected / ntuples_current);
 		int partition_bits;
 		int i;
 
@@ -3037,7 +3172,9 @@ save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
 		if (npartitions > HASH_DISK_MAX_PARTITIONS)
 			npartitions = HASH_DISK_MAX_PARTITIONS;
 
+		/* make it a power of 2 */
 		partition_bits = my_log2(npartitions);
+		npartitions = (1 << partition_bits);
 
 		/* make sure that we don't exhaust the hash bits */
 		if (partition_bits + work->input_bits >= 32)
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 1a61ac7..9b2558e 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1731,6 +1731,20 @@ typedef struct AggState
 	bool		hash_disk;		/* have we exceeded memory yet? */
 	List	   *hash_work;		/* remaining work to be done */
 
+	/* counters used mostly for explain (analyze) */
+	int			batch_count;		/* number of batches generated in total */
+	int			rebatch_count;		/* number of rebatches (splitting of a batch) */
+	Size		batch_min_size;		/* minimum batch size (bytes) */
+	Size		batch_max_size;		/* maximum batch size (bytes) */
+
+	/*
+	 * These two counters allow evaluation of how many times the tuples
+	 * were saved/read. With no batching, rescanned=0. With a single
+	 * level of batching (rescanned/scanned < 1.00) and with multi-level
+	 * batching it may happen that (rescanned/scanned > 1.00). */
+	int64		ntuples_scanned;	/* number of input tuples scanned */
+	int64		ntuples_rescanned;	/* number of tuples saved/read */
+
 } AggState;
 
 /* ----------------

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 49547ee..b651858 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -2884,6 +2884,21 @@ include_dir 'conf.d'
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-enable-hashagg-disk" xreflabel="enable_hashagg_disk">
+      <term><varname>enable_hashagg_disk</varname> (<type>boolean</type>)
+      <indexterm>
+       <primary><varname>enable_hashagg_disk</> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Enables or disables the query planner's use of hashed aggregation plan
+        types when the planner expects the hash table size to exceed
+        <varname>work_mem</varname>. The default is <literal>on</>.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-enable-hashjoin" xreflabel="enable_hashjoin">
       <term><varname>enable_hashjoin</varname> (<type>boolean</type>)
       <indexterm>
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 781a736..ca9f026 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -78,6 +78,8 @@ static void show_merge_append_keys(MergeAppendState *mstate, List *ancestors,
 					   ExplainState *es);
 static void show_agg_keys(AggState *astate, List *ancestors,
 			  ExplainState *es);
+static void show_agg_batching(AggState *astate, List *ancestors,
+			  ExplainState *es);
 static void show_group_keys(GroupState *gstate, List *ancestors,
 				ExplainState *es);
 static void show_sort_group_keys(PlanState *planstate, const char *qlabel,
@@ -1391,6 +1393,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
 										   planstate, es);
 			break;
 		case T_Agg:
+			show_agg_batching((AggState *) planstate, ancestors, es);
 			show_agg_keys((AggState *) planstate, ancestors, es);
 			show_upper_qual(plan->qual, "Filter", planstate, ancestors, es);
 			if (plan->qual)
@@ -1790,6 +1793,37 @@ show_agg_keys(AggState *astate, List *ancestors,
 }
 
 /*
+ * Show the batching info for an Agg node.
+ */
+static void
+show_agg_batching(AggState *astate, List *ancestors,
+			  ExplainState *es)
+{
+	Agg		   *plan = (Agg *) astate->ss.ps.plan;
+
+	if ((es->analyze) && (plan->aggstrategy == AGG_HASHED))
+	{
+		if (es->format == EXPLAIN_FORMAT_TEXT)
+		{
+			appendStringInfoSpaces(es->str, es->indent * 2);
+			appendStringInfo(es->str, "Batch Count: %d  Original: %d  Smallest: %ldkB  Largest: %ldkB Rescanned: %.0f%%\n",
+							 astate->nbatch, astate->nbatch_original,
+							 astate->batch_min_size / 1024,
+							 astate->batch_max_size / 1024,
+							 astate->ntuples_rescanned * 100.0 / astate->ntuples_scanned);
+		}
+		else
+		{
+			ExplainPropertyLong("Batch Count", astate->nbatch, es);
+			ExplainPropertyLong("Batch Count Original", astate->nbatch_original, es);
+			ExplainPropertyLong("Batch Smallest", astate->batch_min_size/1024, es);
+			ExplainPropertyLong("Batch Largest", astate->batch_max_size/1024, es);
+			ExplainPropertyLong("Batch Rescan Rate", (astate->ntuples_rescanned * 100) / astate->ntuples_scanned, es);
+		}
+	}
+}
+
+/*
  * Show the grouping keys for a Group node.
  */
 static void
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index 6455864..d0e30b1 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -96,6 +96,8 @@
 
 #include "postgres.h"
 
+#include <limits.h>
+
 #include "access/htup_details.h"
 #include "catalog/objectaccess.h"
 #include "catalog/pg_aggregate.h"
@@ -108,6 +110,7 @@
 #include "optimizer/tlist.h"
 #include "parser/parse_agg.h"
 #include "parser/parse_coerce.h"
+#include "storage/buffile.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
@@ -115,7 +118,11 @@
 #include "utils/syscache.h"
 #include "utils/tuplesort.h"
 #include "utils/datum.h"
+#include "utils/dynahash.h"
 
+#define HASH_DISK_MIN_PARTITIONS		2
+#define HASH_DISK_DEFAULT_PARTITIONS	4
+#define HASH_DISK_MAX_PARTITIONS		256
 
 /*
  * AggStatePerAggData - per-aggregate working state for the Agg scan
@@ -310,7 +317,6 @@ typedef struct AggHashEntryData
 
 }	AggHashEntryData;	/* VARIABLE LENGTH STRUCT */
 
-
 static void initialize_aggregates(AggState *aggstate,
 					  AggStatePerAgg peragg,
 					  AggStatePerGroup pergroup);
@@ -332,22 +338,44 @@ static Bitmapset *find_unaggregated_cols(AggState *aggstate);
 static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
 static void build_hash_table(AggState *aggstate, Size tuple_width);
 static AggHashEntry lookup_hash_entry(AggState *aggstate,
-				  TupleTableSlot *inputslot);
+					uint32 hashvalue, TupleTableSlot *inputslot);
+static void create_hash_entry(AggState *aggstate, AggHashEntry entry);
 static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
-static void agg_fill_hash_table(AggState *aggstate);
+static bool agg_fill_hash_table(AggState *aggstate);
 static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
 static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
 
 static uint32 compute_hash_value(AggState * aggstate, TupleTableSlot * slot);
 static uint32 compute_bucket(AggState * aggstate, uint32 hashvalue);
+static uint32 compute_batchno(AggState * aggstate, uint32 hashvalue);
 static bool groups_match(AggState * aggstate, TupleTableSlot *slot, AggHashEntry entry);
 static void increase_nbuckets(AggState * aggstate);
+static void increase_nbatches(AggState * aggstate);
 static char * chunk_alloc(AggHashTable htab, int size);
 static void reset_hash_table(AggHashTable htab);
 
+static int choose_nbatch(AggState *aggstate, int nbuckets, Size tuple_width);
+static void init_batch_files(AggState * aggstate);
+static void close_batch_files(AggState * aggstate);
+static void reinit_batch_files(AggState * aggstate);
+
 static void IteratorReset(AggHashTable htab);
 static AggHashEntry IteratorGetNext(AggHashTable htab);
 
+static TupleTableSlot *
+read_saved_tuple(BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot);
+AggHashEntry
+read_saved_group(AggState * aggstate, BufFile *file, AggHashEntry entry);
+
+static void
+save_tuple(AggState *aggstate, int batchno, TupleTableSlot *slot,
+		   uint32 hashvalue);
+static void
+save_group(AggState *aggstate, int batchno, AggHashEntry entry);
+
+static bool
+batching_supported(AggState * aggstate);
+
 /*
  * The size of the chunks for dense allocation. This needs to be >8kB
  * because the default (and only) memory context implementation uses
@@ -392,6 +420,7 @@ typedef struct AggHashTableData
 	int	nentries;		/* number of hash table entries */
 	int	nbuckets;		/* current number of buckets */
 	int	nbuckets_max;	/* max number of buckets */
+	int	nbuckets_bits;	/* bits for nbuckets_max (used for batching) */
 
 	/* items copied from the TupleHashTable, because we still need them */
 	MemoryContext	tmpctx;		/* short-lived memory context (hash/eq funcs) */
@@ -412,6 +441,7 @@ typedef struct AggHashTableData
 	 */
 	HashChunk		cur_chunk;
 	AggHashEntry	cur_entry;
+	int				niterated;
 
 	/* list of chunks with dense-packed entries / minimal tuples */
 	HashChunk		chunks_hash;
@@ -1027,6 +1057,7 @@ build_hash_table(AggState *aggstate, Size tuple_width)
 	/* we assume 1024 buckets (i.e. 8kB of memory) is minimum */
 	int nbuckets     = 1024;
 	int nbuckets_max = 1024;
+	int nbuckets_bits = 10;
 
 	Assert(node->aggstrategy == AGG_HASHED);
 	Assert(node->numGroups > 0);
@@ -1076,7 +1107,10 @@ build_hash_table(AggState *aggstate, Size tuple_width)
 	 *     save a bit of memory by that (although not much).
 	 */
 	while (nbuckets_max * groupsize <= work_mem * 1024L)
+	{
 		nbuckets_max *= 2;
+		nbuckets_bits += 1;
+	}
 
 	/*
 	 * Update the initial number of buckets to match expected number of groups,
@@ -1096,12 +1130,15 @@ build_hash_table(AggState *aggstate, Size tuple_width)
 	htab = (AggHashTable)MemoryContextAllocZero(aggstate->aggcontext,
 											sizeof(AggHashTableData));
 
+	htab->niterated = 0;
+
 	/* TODO create a memory context for the hash table */
-	htab->htabctx = AllocSetContextCreate(aggstate->aggcontext,
+	htab->htabctx = AllocSetContextCreateTracked(aggstate->aggcontext,
 											"HashAggHashTable",
 											ALLOCSET_DEFAULT_MINSIZE,
 											ALLOCSET_DEFAULT_INITSIZE,
-											ALLOCSET_DEFAULT_MAXSIZE);
+											ALLOCSET_DEFAULT_MAXSIZE,
+											true);
 
 	/* buckets are just pointers to AggHashEntryData structures */
 	htab->buckets = (AggHashEntry*)MemoryContextAllocZero(htab->htabctx,
@@ -1115,6 +1152,7 @@ build_hash_table(AggState *aggstate, Size tuple_width)
 
 	htab->nbuckets = nbuckets;
 	htab->nbuckets_max = nbuckets_max;
+	htab->nbuckets_bits = nbuckets_bits;
 	htab->nentries = 0;
 	htab->slot = NULL;
 	htab->numCols = node->numCols;
@@ -1198,15 +1236,14 @@ hash_agg_entry_size(int numAggs)
  * When called, CurrentMemoryContext should be the per-query context.
  */
 static AggHashEntry
-lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
+lookup_hash_entry(AggState *aggstate, uint32 hashvalue,
+TupleTableSlot *inputslot)
 {
 
 	AggHashEntry entry = NULL;
-	uint32		hashvalue;
 	uint32		bucketno;
 	MinimalTuple mintuple;
 
-	hashvalue = compute_hash_value(aggstate, inputslot);
 	bucketno = compute_bucket(aggstate, hashvalue);
 
 	entry = aggstate->hashtable->buckets[bucketno];
@@ -1223,9 +1260,11 @@ lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
 		entry = entry->next;
 	}
 
-	/* There's not a maching entry in the bucket, so create a new one and
-	 * copy in data both for the aggregates, and the MinimalTuple containing
-	 * keys for the group columns. */
+	/*
+	 * There's not a maching entry in the bucket, create a new one and
+	 * copy in data both for the aggregates, and the MinimalTuple
+	 * containing keys for the group columns.
+	 */
 	if (entry == NULL)
 	{
 
@@ -1265,16 +1304,48 @@ lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
 
 		aggstate->hashtable->nentries += 1;
 
-	}
+		/* once we exceed 1 entry / bucket, increase number of buckets */
+		if (aggstate->hashtable->nentries > aggstate->hashtable->nbuckets)
+			increase_nbuckets(aggstate);
 
-	/* once we exceed 1 entry / bucket, increase number of buckets */
-	if (aggstate->hashtable->nentries > aggstate->hashtable->nbuckets)
-		increase_nbuckets(aggstate);
+	}
 
 	return entry;
 }
 
 /*
+ * Creates a new hash entry in the hash table, containing the provided
+ * data. This assumes there's not a matching entry (this is not checked,
+ * and it's expected the caller not to break this).
+ * 
+ * This is used when adding entries with aggregate states read from
+ * a batch file.
+ */
+static void create_hash_entry(AggState *aggstate, AggHashEntry entry)
+{
+	AggHashEntry entry_new
+		= (AggHashEntry) chunk_alloc(aggstate->hashtable,
+					aggstate->hashtable->entrysize + entry->tuple->t_len);
+
+	AggHashTable htab = aggstate->hashtable;
+
+	int bucketno = compute_bucket(aggstate, entry->hashvalue);
+
+	Assert((bucketno >= 0) && (bucketno < htab->nbuckets));
+	Assert(aggstate->cur_batch == compute_batchno(aggstate, entry->hashvalue));
+
+	memcpy(entry_new, entry, htab->entrysize);
+
+	entry_new->tuple = (MinimalTuple)((char*)entry_new + htab->entrysize);
+
+	memcpy(entry_new->tuple, entry->tuple, entry->tuple->t_len);
+
+	entry_new->next = htab->buckets[bucketno];
+	htab->buckets[bucketno] = entry_new;
+
+}
+
+/*
  * ExecAgg -
  *
  *	  ExecAgg receives tuples from its outer subplan and aggregates over
@@ -1318,9 +1389,16 @@ ExecAgg(AggState *node)
 	/* Dispatch based on strategy */
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
-		if (!node->table_filled)
-			agg_fill_hash_table(node);
-		return agg_retrieve_hash_table(node);
+		TupleTableSlot *slot = NULL;
+
+		while (slot == NULL)
+		{
+			if (!node->table_filled)
+				if (! agg_fill_hash_table(node))
+					break;	/* no more batches to process */
+			slot = agg_retrieve_hash_table(node);
+		}
+		return slot;
 	}
 	else
 		return agg_retrieve_direct(node);
@@ -1536,13 +1614,16 @@ agg_retrieve_direct(AggState *aggstate)
 /*
  * ExecAgg for hashed case: phase 1, read input and build hash table
  */
-static void
+static bool
 agg_fill_hash_table(AggState *aggstate)
 {
 	PlanState  *outerPlan;
 	ExprContext *tmpcontext;
 	AggHashEntry entry;
-	TupleTableSlot *outerslot;
+	TupleTableSlot *outerslot = NULL;
+	int64		ntuples = 0;
+	Size		allocated;
+	BufFile	   *infile = NULL;
 
 	/*
 	 * get state info from node
@@ -1551,33 +1632,172 @@ agg_fill_hash_table(AggState *aggstate)
 	/* tmpcontext is the per-input-tuple expression context */
 	tmpcontext = aggstate->tmpcontext;
 
+	/* if there're no more batches, we're done */
+	if (aggstate->cur_batch == aggstate->nbatch)
+	{
+		aggstate->agg_done = true;
+		return false;
+	}
+
+	/* if not the first time through, reinitialize */
+	if (aggstate->cur_batch > 0)
+	{
+
+		BufFile *file_groups = aggstate->batched_groups[aggstate->cur_batch];
+
+		/* used for all the read_saved_group calls, to minimize palloc
+		 * overhead (and released in the last one automatically) */
+		AggHashEntry entry = (AggHashEntry)palloc0(aggstate->hashtable->entrysize);
+
+		/* reset the hash table (free the chunks, zero buckets, ...) */
+		reset_hash_table(aggstate->hashtable);
+
+		/* read all the aggregate states and either insert them into the
+		 * hash table, or move them to the proper batch */
+		BufFileSeek(file_groups, 0, 0L, SEEK_SET);
+
+		while ((entry = read_saved_group(aggstate, file_groups, entry)) != NULL)
+		{
+			/* XXX hashjoin uses a single call to compute both bucket
+			 * and batch, maybe we could do the same here (and pass
+			 * bucketno to create_hash_entry) */
+			int batchno = compute_batchno(aggstate, entry->hashvalue);
+
+			if (batchno == aggstate->cur_batch)
+				/* keep in the current batch */
+				create_hash_entry(aggstate, entry);
+			else
+				/* move to the proper batch */
+				save_group(aggstate, batchno, entry);
+		}
+
+		/* we're done with the temp file */
+		BufFileClose(file_groups);
+		aggstate->batched_groups[aggstate->cur_batch] = NULL;
+
+		/* prepare to read the saved tuples */
+		BufFileSeek(aggstate->batched_tuples[aggstate->cur_batch], 0, 0L, SEEK_SET);
+		infile = aggstate->batched_tuples[aggstate->cur_batch];
+	}
+
 	/*
 	 * Process each outer-plan tuple, and then fetch the next one, until we
 	 * exhaust the outer plan.
 	 */
 	for (;;)
 	{
-		outerslot = ExecProcNode(outerPlan);
-		if (TupIsNull(outerslot))
-			break;
-		/* set up for advance_aggregates call */
-		tmpcontext->ecxt_outertuple = outerslot;
 
-		/* Find or build hashtable entry for this tuple's group */
-		entry = lookup_hash_entry(aggstate, outerslot);
+		uint32 hashvalue;
+		int batchno = 0;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* the first batch means we need to fetch the tuples */
+		if (aggstate->cur_batch == 0)
+		{
+			outerslot = ExecProcNode(outerPlan);
+
+			if (TupIsNull(outerslot))
+				break;
+
+			/* copy the tuple descriptor for the following batches */
+			if (aggstate->tupdesc == NULL)
+			{
+				MemoryContext oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
+				aggstate->tupdesc = CreateTupleDescCopy(outerslot->tts_tupleDescriptor);
+				MemoryContextSwitchTo(oldContext);
+			}
+
+			hashvalue = compute_hash_value(aggstate, outerslot);
+			aggstate->ntuples_scanned += 1;
+		}
+		else
+		{
+
+			if (outerslot == NULL)
+				outerslot = MakeSingleTupleTableSlot(aggstate->tupdesc);
+
+			outerslot = read_saved_tuple(infile, &hashvalue, outerslot);
+			if (TupIsNull(outerslot))
+			{
+				BufFileClose(infile);
+				aggstate->batched_tuples[aggstate->cur_batch] = NULL;
+				break;
+			}
+
+			aggstate->ntuples_rescanned += 1;
+		}
+
+		ntuples++;
+
+		batchno = compute_batchno(aggstate, hashvalue);
+
+		Assert(batchno >= aggstate->cur_batch);
+
+		if (batchno == aggstate->cur_batch) {
+
+			/* Find or build hashtable entry for this tuple's group */
+			entry = lookup_hash_entry(aggstate, hashvalue, outerslot);
+
+			/* set up for advance_aggregates call */
+			tmpcontext->ecxt_outertuple = outerslot;
 
-		/* Advance the aggregates */
-		advance_aggregates(aggstate, entry->pergroup);
+			/* Advance the aggregates */
+			advance_aggregates(aggstate, entry->pergroup);
 
-		/* Reset per-input-tuple context after each tuple */
-		ResetExprContext(tmpcontext);
+			/* Reset per-input-tuple context after each tuple */
+			ResetExprContext(tmpcontext);
+
+			/* have we exceeded work_mem? if yes, increase number of batches
+			 * 
+			 * FIXME This uses htabctx, which is OK for states using
+			 *       pass-by-value types, but it's not really correct
+			 *       in general (use aggcontext instead).
+			 */
+			if (MemoryContextGetAllocated(aggstate->hashtable->htabctx, true) >= work_mem * 1024L)
+				increase_nbatches(aggstate);
+
+		} else {
+
+			/* no entry for this tuple, and we've reached work_mem */
+			save_tuple(aggstate, batchno, outerslot, hashvalue);
+
+		}
 	}
 
+	allocated = MemoryContextGetAllocated(aggstate->hashtable->htabctx, true);
+
+	/* keep track of the largest/smallest batch size */
+	if (aggstate->cur_batch == 0)
+	{
+		aggstate->batch_min_size = allocated;
+		aggstate->batch_max_size = allocated;
+	}
+	else
+	{
+		if (allocated < aggstate->batch_min_size)
+			aggstate->batch_min_size = allocated;
+		if (allocated > aggstate->batch_max_size)
+			aggstate->batch_max_size = allocated;
+	}
+
+	/* if we're in the first batch, and there were 0 tuples, we're done */
+	if ((aggstate->cur_batch == 0) && (aggstate->ntuples_scanned == 0))
+	{
+		aggstate->agg_done = true;
+		return false;
+	}
+
+	/* the hash table is filled, and we're ready for the next batch */
 	aggstate->table_filled = true;
+	aggstate->cur_batch += 1;
 
 	/* Initialize for iteration through the table (first bucket / entry) */
 	IteratorReset(aggstate->hashtable);
 
+	/* ready to return groups from this hash table */
+	return true;
+
 }
 
 /*
@@ -1620,6 +1840,8 @@ agg_retrieve_hash_table(AggState *aggstate)
 		 */
 		ResetExprContext(econtext);
 
+		htab->niterated += 1;
+
 		/*
 		* Store the copied first input tuple in the tuple table slot reserved
 		* for it, so that it can be used in ExecProject.
@@ -1677,7 +1899,8 @@ agg_retrieve_hash_table(AggState *aggstate)
 
 	}
 
-	aggstate->agg_done = true;
+	/* No more entries in hashtable, so done with this batch */
+	aggstate->table_filled = false;
 
 	/* No more groups */
 	return NULL;
@@ -1739,11 +1962,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	 * recover no-longer-wanted space.
 	 */
 	aggstate->aggcontext =
-		AllocSetContextCreate(CurrentMemoryContext,
+		AllocSetContextCreateTracked(CurrentMemoryContext,
 							  "AggContext",
 							  ALLOCSET_DEFAULT_MINSIZE,
 							  ALLOCSET_DEFAULT_INITSIZE,
-							  ALLOCSET_DEFAULT_MAXSIZE);
+							  ALLOCSET_DEFAULT_MAXSIZE, true);
 
 	/*
 	 * tuple table initialization
@@ -1842,10 +2065,29 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 
 	if (node->aggstrategy == AGG_HASHED)
 	{
+		MemoryContext oldContext;
+
 		build_hash_table(aggstate, outerPlan->plan_width);
 		aggstate->table_filled = false;
+
+		aggstate->tupdesc = NULL;
+		aggstate->nbatch = choose_nbatch(aggstate, aggstate->hashtable->nbuckets, outerPlan->plan_width);
+		aggstate->nbatch_original = aggstate->nbatch;
+		aggstate->cur_batch = 0;
+
+		/* initialize temporary files for batched tuples/groups */
+		init_batch_files(aggstate);
+
+		/* explain (analyze) counters */
+		aggstate->ntuples_scanned = 0;
+		aggstate->ntuples_rescanned = 0;
+
 		/* Compute the columns we actually need to hash on */
 		aggstate->hash_needed = find_hash_columns(aggstate);
+
+		/* prime with initial work item to read from outer plan */
+		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
+		MemoryContextSwitchTo(oldContext);
 	}
 	else
 	{
@@ -2198,6 +2440,18 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
 	/* Update numaggs to match number of unique aggregates found */
 	aggstate->numaggs = aggno + 1;
 
+	/* check whether we can do batching */
+	aggstate->batching_enabled = batching_supported(aggstate);
+
+	/*
+	 * If in hashed mode, with no batching, disable nbuckets_max limit,
+	 * because if we're gonna exhaust memory, there's not much
+	 * difference between doing that fast and slow. It's equally bad
+	 * either way :-/
+	 */
+	if ((node->aggstrategy == AGG_HASHED) && (! aggstate->batching_enabled))
+		aggstate->hashtable->nbuckets_max = INT_MAX/2;
+
 	return aggstate;
 }
 
@@ -2245,6 +2499,10 @@ ExecEndAgg(AggState *node)
 	/* clean up tuple table */
 	ExecClearTuple(node->ss.ss_ScanTupleSlot);
 
+	/* properly close the batch files (in batching mode) */
+	if (node->nbatch != 0)
+		close_batch_files(node);
+
 	MemoryContextDelete(node->aggcontext);
 
 	outerPlan = outerPlanState(node);
@@ -2264,22 +2522,24 @@ ExecReScanAgg(AggState *node)
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
 		/*
-		 * In the hashed case, if we haven't yet built the hash table then we
-		 * can just return; nothing done yet, so nothing to undo. If subnode's
-		 * chgParam is not NULL then it will be re-scanned by ExecProcNode,
-		 * else no reason to re-scan it at all.
+		 * In the hashed case, if we haven't done any execution work yet, we
+		 * can just return; nothing to undo. If subnode's chgParam is not NULL
+		 * then it will be re-scanned by ExecProcNode, else no reason to
+		 * re-scan it at all.
 		 */
-		if (!node->table_filled)
-			return;
+		// FIXME maybe it was a good idea to have hash_init_state ...
+		// if (node->hash_init_state)
+		//	return;
 
 		/*
-		 * If we do have the hash table and the subplan does not have any
-		 * parameter changes, then we can just rescan the existing hash table;
-		 * no need to build it again.
+		 * If we do have the hash table, it never went to disk, and the
+		 * subplan does not have any parameter changes, then we can just
+		 * rescan the existing hash table; no need to build it again.
 		 */
-		if (node->ss.ps.lefttree->chgParam == NULL)
+		if (node->ss.ps.lefttree->chgParam == NULL && (node->nbatch == 1))
 		{
 			IteratorReset(node->hashtable);
+			node->table_filled = true;
 			return;
 		}
 	}
@@ -2318,10 +2578,43 @@ ExecReScanAgg(AggState *node)
 
 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
 	{
-		Plan * outerPlan = outerPlan((Agg *) node->ss.ps.plan);
+		Plan *outerPlan = outerPlan((Agg *) node->ss.ps.plan);
+
 		/* Rebuild an empty hash table */
 		build_hash_table(node, outerPlan->plan_width);
 		node->table_filled = false;
+		node->tupdesc = NULL;
+
+		/* reset the current batch, but remember the nbatch */
+		node->cur_batch = 0;
+
+		/* XXX The way we work with the temporary files right now is that
+		 * on rescan we throw them away, and start over. The problem is
+		 * that when the rescan triggers after somewhere after the initial
+		 * batch and before completing all the batches, we don't know
+		 * which groups/tuples were already moved (copied) to the following
+		 * batches (so we can't just move them again). Also, we close the
+		 * files as soon as we complete reading them.
+		 * 
+		 * We could however improve this by keeping the files open until
+		 * ExecEndAgg, and remembering which tuples / groups we've
+		 * already moved to the appropriate batch (a batchno/tupleno
+		 * pair should be enough), and only move the tuples after that.
+		 * 
+		 * The problem is with the initial batch, which is only in memory
+		 * by default. We could serialize this to a file once it's
+		 * complete (only the groups, should be less than work_mem),
+		 * but that's likely to impact even plans that don't require
+		 * the rescan. Not sure if it's know in advance whether a rescan
+		 * is likely to happen.
+		 */
+
+		/* reinitialize the files with batched tuples/groups */
+		reinit_batch_files(node);
+
+		/* explain (analyze) counters */
+		node->ntuples_scanned = 0;
+		node->ntuples_rescanned = 0;
 	}
 	else
 	{
@@ -2649,9 +2942,6 @@ increase_nbuckets(AggState * aggstate)
 		/* position within the buffer (up to chunk->used) */
 		size_t idx = 0;
 
-		/* we have a whole number of entries */
-		Assert(chunk->used % htab->entrysize == 0);
-
 		/* process all tuples stored in this chunk (and then free it) */
 		while (idx < chunk->used)
 		{
@@ -2675,6 +2965,120 @@ increase_nbuckets(AggState * aggstate)
 
 }
 
+
+/*
+ * Increase the number of batches - we'll double it by default, but we
+ * may grow faster if needed. Contrary to increasing the number of
+ * buckets, this needs to remove ~50% of the entries (when doubling
+ * the number of batches).
+ * 
+ * We keep the number of buckets etc. because we expect the table to
+ * grow further.
+ */
+static void
+increase_nbatches(AggState * aggstate)
+{
+
+	HashChunk chunk, chunk_prev;
+	AggHashTable htab = aggstate->hashtable;
+	int i, nbatch_old = aggstate->nbatch;
+	MemoryContext oldctx;
+
+	/* remember the old chunks (and reset to NULL, to allocate new ones) */
+	HashChunk oldchunks = htab->chunks_hash;
+	htab->chunks_hash = NULL;
+
+	aggstate->nbatch      *= 2;
+	aggstate->hashtable->nentries = 0;
+
+	oldctx = MemoryContextSwitchTo(aggstate->aggcontext);
+
+	/* initialize appropriate number of temporary files */
+	if (aggstate->nbatch == 2)
+	{
+		aggstate->batched_groups = (BufFile**)palloc0(2*sizeof(BufFile*));
+		aggstate->batched_tuples = (BufFile**)palloc0(2*sizeof(BufFile*));
+	}
+	else
+	{
+		aggstate->batched_groups = (BufFile**)repalloc(aggstate->batched_groups, aggstate->nbatch * sizeof(BufFile*));
+		aggstate->batched_tuples = (BufFile**)repalloc(aggstate->batched_tuples, aggstate->nbatch * sizeof(BufFile*));
+	}
+
+	for (i = nbatch_old; i < aggstate->nbatch; i++)
+	{
+		aggstate->batched_groups[i] = BufFileCreateTemp(false);
+		aggstate->batched_tuples[i] = BufFileCreateTemp(false);
+	}
+
+	MemoryContextSwitchTo(oldctx);
+
+	/* reset the buckets (we'll rebuild them from scratch) */
+	memset(htab->buckets, 0, htab->nbuckets * sizeof(AggHashEntry));
+
+	chunk = oldchunks;
+	while (chunk != NULL)
+	{
+
+		/* position within the buffer (up to chunk->used) */
+		size_t idx = 0;
+
+		/* current chunk (so that we can pfree it at the end) */
+		chunk_prev = chunk;
+
+		/* process all tuples stored in this chunk (and then free it) */
+		while (idx < chunk->used)
+		{
+
+			AggHashEntry entry = (AggHashEntry)(chunk->data + idx);
+
+			/* this already uses the updated nbatch values */
+			int batchno = compute_batchno(aggstate, entry->hashvalue);
+
+			Assert(batchno >= aggstate->cur_batch);
+
+			if (batchno == aggstate->cur_batch) {
+
+				/*
+				 * If the batch number is still cur_batch, copy it to
+				 * a new chunk, and put it into the proper bucket.
+				 */
+
+				int bucketno = compute_bucket(aggstate, entry->hashvalue);
+
+				AggHashEntry entry_new = (AggHashEntry)chunk_alloc(htab,
+								htab->entrysize + entry->tuple->t_len);
+
+				memcpy(entry_new, entry, htab->entrysize + entry->tuple->t_len);
+
+				entry_new->tuple = (MinimalTuple)((char*)entry_new + htab->entrysize);
+
+				/* fine, just put the entry into */
+				entry_new->next = htab->buckets[bucketno];
+				htab->buckets[bucketno] = entry_new;
+
+				aggstate->hashtable->nentries += 1;
+
+			} else {
+
+				/* different batch - save the group */
+				save_group(aggstate, batchno, entry);
+
+			}
+
+			/* bytes occupied in memory HJ tuple overhead + actual tuple length */
+			idx += htab->entrysize + entry->tuple->t_len;
+
+		}
+
+		/* proceed to the next chunk */
+		chunk = chunk->next;
+
+		pfree(chunk_prev);
+	}
+
+}
+
 static
 char * chunk_alloc(AggHashTable htab, int size)
 {
@@ -2804,6 +3208,7 @@ AggHashEntry IteratorGetNext(AggHashTable htab)
 	 * So compute how many bytes we need to skip to the next entry.
 	 */
 	entry = htab->cur_entry;
+
 	len = entry->tuple->t_len + htab->entrysize;
 
 	/*
@@ -2826,6 +3231,166 @@ AggHashEntry IteratorGetNext(AggHashTable htab)
 
 }
 
+
+/*
+ * save_tuple
+ *
+ * Not enough memory to add tuple as new entry in hash table. Save for later
+ * in the appropriate partition.
+ */
+static void
+save_tuple(AggState *aggstate, int batchno, TupleTableSlot *slot,
+		   uint32 hashvalue)
+{
+	MinimalTuple		 tuple;
+	BufFile				*file;
+	int					 written;
+
+	Assert(batchno > aggstate->cur_batch);
+
+	file = aggstate->batched_tuples[batchno];
+
+	tuple = ExecFetchSlotMinimalTuple(slot);
+
+	written = BufFileWrite(file, (void *) &hashvalue, sizeof(uint32));
+	if (written != sizeof(uint32))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not write to HashAgg temporary file: %m")));
+
+	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")));
+}
+
+
+/*
+ * save_group
+ *
+ */
+static void
+save_group(AggState *aggstate, int batchno, AggHashEntry entry)
+{
+	MinimalTuple		 tuple;
+	BufFile				*file;
+	int					 written;
+
+	Assert(batchno > aggstate->cur_batch);
+
+	file = aggstate->batched_groups[batchno];
+	tuple = entry->tuple;
+
+	written = BufFileWrite(file, (void *) entry, aggstate->hashtable->entrysize);
+	if (written != aggstate->hashtable->entrysize)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not write to HashAgg temporary file: %m")));
+
+	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")));
+}
+
+
+/*
+ * read_saved_tuple
+ *		read the next tuple from a batch file.  Return NULL if no more.
+ *
+ * On success, *hashvalue is set to the tuple's hash value, and the tuple
+ * itself is stored in the given slot.
+ *
+ * Copied with minor modifications from ExecHashJoinGetSavedTuple.
+ */
+static TupleTableSlot *
+read_saved_tuple(BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot)
+{
+	uint32		header[2];
+	size_t		nread;
+	MinimalTuple tuple;
+
+	/*
+	 * Since both the hash value and the MinimalTuple length word are uint32,
+	 * we can read them both in one BufFileRead() call without any type
+	 * cheating.
+	 */
+	nread = BufFileRead(file, (void *) header, sizeof(header));
+	if (nread == 0)				/* end of file */
+	{
+		ExecClearTuple(tupleSlot);
+		return NULL;
+	}
+	if (nread != sizeof(header))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+	*hashvalue = header[0];
+	tuple = (MinimalTuple) palloc(header[1]);
+	tuple->t_len = header[1];
+	nread = BufFileRead(file,
+						(void *) ((char *) tuple + sizeof(uint32)),
+						header[1] - sizeof(uint32));
+	if (nread != header[1] - sizeof(uint32))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+	return ExecStoreMinimalTuple(tuple, tupleSlot, true);
+}
+
+/*
+ * read_saved_group
+ */
+AggHashEntry
+read_saved_group(AggState * aggstate, BufFile *file, AggHashEntry entry)
+{
+	uint32		tlen;
+	size_t		nread;
+
+	/* we know the size of the entry, we don't know the tuple size yet */
+
+	Assert(entry != NULL);
+
+	/* always release the tuple (well, maybe we could keep track of the
+	 * allocated space and reuse the tuple buffer) */
+	if (entry->tuple != NULL)
+		pfree(entry->tuple);
+
+	nread = BufFileRead(file, (void *)entry, aggstate->hashtable->entrysize);
+	if (nread == 0)				/* end of file */
+	{
+		pfree(entry);
+		return NULL;
+	}
+
+	if (nread != aggstate->hashtable->entrysize)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+
+	/* now, we need to read the actual tuple - first, read the length */
+	nread = BufFileRead(file, (void *)&tlen, sizeof(uint32));
+	if (nread != sizeof(int32))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+
+	/* now, allocate space for the tuple and read the rest */
+	entry->tuple = (MinimalTuple) palloc(tlen);
+	entry->tuple->t_len = tlen;
+	nread = BufFileRead(file,
+						(void *) ((char *) entry->tuple + sizeof(uint32)),
+						tlen - sizeof(uint32));
+	if (nread != tlen - sizeof(uint32))
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read from HashAgg temporary file: %m")));
+
+	return entry;
+}
+
 /*
  * Resets the contents of the hash table - removes all the entries and
  * tuples, but keeps the 'size' of the hash table (nbuckets).
@@ -2858,3 +3423,114 @@ void reset_hash_table(AggHashTable htab) {
 								htab->nbuckets * sizeof(AggHashEntry));
 
 }
+
+static uint32 compute_batchno(AggState * aggstate, uint32 hashvalue) {
+	if ((! aggstate->batching_enabled) || (aggstate->nbatch == 1))
+		return 0;
+	else
+		/*
+		 * When computing the batch number, skip the bits that might be
+		 * used for buckets.
+		 * 
+		 * XXX We should probably make sure that we don't exceed the 32
+		 *     bits we have available in the hash. This is pretty much
+		 *     the same issue as in hash join.
+		 */
+		return (hashvalue >> aggstate->hashtable->nbuckets_bits) & (aggstate->nbatch - 1);
+}
+
+static bool
+batching_supported(AggState * aggstate)
+{
+	int aggno;
+
+	/* check that all the aggregates use state passed by value */
+	for (aggno = 0; aggno < aggstate->numaggs; aggno++)
+		if (! aggstate->peragg[aggno].transtypeByVal)
+			return false;
+
+	return true;
+}
+
+static void
+init_batch_files(AggState * aggstate)
+{
+
+	int i;
+
+	if (aggstate->nbatch > 1) {
+		aggstate->batched_groups = (BufFile**)palloc0(aggstate->nbatch * sizeof(BufFile*));
+		aggstate->batched_tuples = (BufFile**)palloc0(aggstate->nbatch * sizeof(BufFile*));
+	}
+
+	for (i = 1; i < aggstate->nbatch; i++) {
+		aggstate->batched_groups[i] = BufFileCreateTemp(false);
+		aggstate->batched_tuples[i] = BufFileCreateTemp(false);
+	}
+
+}
+
+static void
+close_batch_files(AggState * aggstate)
+{
+	int i;
+
+	for (i = 1; i < aggstate->nbatch; i++) {
+
+		/* if we're halfway through the batches, the files might be
+		 * already closed (and set to NULL) */
+		if (aggstate->batched_groups[i] != NULL)
+			BufFileClose(aggstate->batched_groups[i]);
+
+		if (aggstate->batched_tuples[i] != NULL)
+			BufFileClose(aggstate->batched_tuples[i]);
+
+		aggstate->batched_groups[i] = NULL;
+		aggstate->batched_tuples[i] = NULL;
+
+	}
+}
+
+static void
+reinit_batch_files(AggState * aggstate)
+{
+	int i;
+
+	/* make sure all the files are properly closed */
+	close_batch_files(aggstate);
+
+	/* reinit all the files (skip the first one, which is batchno=0) */
+	for (i = 1; i < aggstate->nbatch; i++)
+	{
+		aggstate->batched_groups[i] = BufFileCreateTemp(false);
+		aggstate->batched_tuples[i] = BufFileCreateTemp(false);
+	}
+}
+
+static int
+choose_nbatch(AggState *aggstate, int nbuckets, Size tuple_width)
+{
+	Agg		   *node = (Agg *) aggstate->ss.ps.plan;
+
+	/* space used by the group (includes bucket) */
+	Size		groupsize, bucketssize, groupssize;
+	int nbatch = 1;
+
+	Assert(node->aggstrategy == AGG_HASHED);
+	Assert(node->numGroups > 0);
+
+	/* XXX see how build_hash_table estimates entrysize and groupsize */
+	groupsize = MAXALIGN(sizeof(AggHashEntryData) +
+			(aggstate->numaggs - 1) * sizeof(AggStatePerGroupData)) +
+			+ MAXALIGN(sizeof(MinimalTupleData)) + tuple_width;
+
+	bucketssize = nbuckets * sizeof(AggHashEntry);
+	groupssize = groupsize * node->numGroups;
+
+	/* double nbatch until we're expected to fit in work_mem */
+	while (groupssize / nbatch + bucketssize >= work_mem * 1024L)
+		nbatch *= 2;
+
+	return nbatch;
+
+}
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index 0cdb790..926abad 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -113,6 +113,7 @@ bool		enable_bitmapscan = true;
 bool		enable_tidscan = true;
 bool		enable_sort = true;
 bool		enable_hashagg = true;
+bool		enable_hashagg_disk = true;
 bool		enable_nestloop = true;
 bool		enable_material = true;
 bool		enable_mergejoin = true;
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index e1480cd..7b8135d 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -2741,7 +2741,8 @@ choose_hashed_grouping(PlannerInfo *root,
 	/* plus the per-hash-entry overhead */
 	hashentrysize += hash_agg_entry_size(agg_costs->numAggs);
 
-	if (hashentrysize * dNumGroups > work_mem * 1024L)
+	if (!enable_hashagg_disk &&
+		hashentrysize * dNumGroups > work_mem * 1024L)
 		return false;
 
 	/*
@@ -2907,7 +2908,8 @@ choose_hashed_distinct(PlannerInfo *root,
 	/* plus the per-hash-entry overhead */
 	hashentrysize += hash_agg_entry_size(0);
 
-	if (hashentrysize * dNumDistinctRows > work_mem * 1024L)
+	if (!enable_hashagg_disk &&
+		hashentrysize * dNumDistinctRows > work_mem * 1024L)
 		return false;
 
 	/*
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 8c57803..5128e20 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -749,6 +749,15 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 	{
+		{"enable_hashagg_disk", PGC_USERSET, QUERY_TUNING_METHOD,
+			gettext_noop("Enables the planner's use of disk-based hashed aggregation plans."),
+			NULL
+		},
+		&enable_hashagg_disk,
+		true,
+		NULL, NULL, NULL
+	},
+	{
 		{"enable_material", PGC_USERSET, QUERY_TUNING_METHOD,
 			gettext_noop("Enables the planner's use of materialization."),
 			NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index df98b02..8f5b73b 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -266,6 +266,7 @@
 
 #enable_bitmapscan = on
 #enable_hashagg = on
+#enable_hashagg_disk = on
 #enable_hashjoin = on
 #enable_indexscan = on
 #enable_indexonlyscan = on
diff --git a/src/backend/utils/mmgr/mcxt.c b/src/backend/utils/mmgr/mcxt.c
index a70b296..97034f1 100644
--- a/src/backend/utils/mmgr/mcxt.c
+++ b/src/backend/utils/mmgr/mcxt.c
@@ -634,7 +634,7 @@ MemoryContextCreate(NodeTag tag, Size size,
 	 */
 	if (track_mem)
 	{
-		node->accounting = (MemoryAccounting)MemoryContextAlloc(TopMemoryContext,
+		node->accounting = (MemoryAccounting)MemoryContextAllocZero(TopMemoryContext,
 												sizeof(MemoryAccountingData));
 		if (parent)
 			node->accounting->parent = parent->accounting;
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 995389b..f2286fe 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -22,7 +22,7 @@
 #include "utils/reltrigger.h"
 #include "utils/sortsupport.h"
 #include "utils/tuplestore.h"
-
+#include "storage/buffile.h"
 
 /* ----------------
  *	  IndexInfo information
@@ -1725,6 +1725,31 @@ typedef struct AggState
 	List	   *hash_needed;	/* list of columns needed in hash table */
 	bool		table_filled;	/* hash table filled yet? */
 	AggHashTable	hashtable;	/* instance of the simple hash table */
+	TupleDesc	tupdesc;
+
+	/* simple batching */
+	bool		batching_enabled;	/* can we serialize group states? */
+	int			nbatch;				/* number of batches */
+	int			nbatch_bits;		/* bits from the hash */
+	int			nbatch_original;	/* number of batches (original) */
+	int			cur_batch;			/* current batch (up to nbatch-1) */
+
+	/* temporary files with data for further batches */
+	BufFile		**batched_groups;	/* serialized aggregate states */
+	BufFile		**batched_tuples;	/* serialized tuples */
+
+	/* counters for explain (analyze) */
+	Size		batch_min_size;		/* minimum batch size (bytes) */
+	Size		batch_max_size;		/* maximum batch size (bytes) */
+
+	/*
+	 * These two counters allow evaluation of how many times the tuples
+	 * were saved/read. With no batching, rescanned=0. With a single
+	 * level of batching (rescanned/scanned < 1.00) and with multi-level
+	 * batching it may happen that (rescanned/scanned > 1.00).
+	 */
+	int64		ntuples_scanned;	/* number of input tuples scanned */
+	int64		ntuples_rescanned;	/* number of tuples saved/read */
 
 } AggState;
 
diff --git a/src/include/optimizer/cost.h b/src/include/optimizer/cost.h
index 75e2afb..d363e65 100644
--- a/src/include/optimizer/cost.h
+++ b/src/include/optimizer/cost.h
@@ -57,6 +57,7 @@ extern bool enable_bitmapscan;
 extern bool enable_tidscan;
 extern bool enable_sort;
 extern bool enable_hashagg;
+extern bool enable_hashagg_disk;
 extern bool enable_nestloop;
 extern bool enable_material;
 extern bool enable_mergejoin;
diff --git a/src/test/regress/expected/rangefuncs.out b/src/test/regress/expected/rangefuncs.out
index 774e75e..e88c83c 100644
--- a/src/test/regress/expected/rangefuncs.out
+++ b/src/test/regress/expected/rangefuncs.out
@@ -3,6 +3,7 @@ SELECT name, setting FROM pg_settings WHERE name LIKE 'enable%';
 ----------------------+---------
  enable_bitmapscan    | on
  enable_hashagg       | on
+ enable_hashagg_disk  | on
  enable_hashjoin      | on
  enable_indexonlyscan | on
  enable_indexscan     | on
@@ -12,7 +13,7 @@ SELECT name, setting FROM pg_settings WHERE name LIKE 'enable%';
  enable_seqscan       | on
  enable_sort          | on
  enable_tidscan       | on
-(11 rows)
+(12 rows)
 
 CREATE TABLE foo2(fooid int, f2 int);
 INSERT INTO foo2 VALUES(1, 11);

*** a/doc/src/sgml/config.sgml
--- b/doc/src/sgml/config.sgml
***************
*** 3017,3022 **** include_dir 'conf.d'
--- 3017,3037 ----
        </listitem>
       </varlistentry>
  
+      <varlistentry id="guc-enable-hashagg-disk" xreflabel="enable_hashagg_disk">
+       <term><varname>enable_hashagg_disk</varname> (<type>boolean</type>)
+       <indexterm>
+        <primary><varname>enable_hashagg_disk</> configuration parameter</primary>
+       </indexterm>
+       </term>
+       <listitem>
+        <para>
+         Enables or disables the query planner's use of hashed aggregation plan
+         types when the planner expects the hash table size to exceed
+         <varname>work_mem</varname>. The default is <literal>on</>.
+        </para>
+       </listitem>
+      </varlistentry>
+ 
       <varlistentry id="guc-enable-hashjoin" xreflabel="enable_hashjoin">
        <term><varname>enable_hashjoin</varname> (<type>boolean</type>)
        <indexterm>
*** a/src/backend/commands/explain.c
--- b/src/backend/commands/explain.c
***************
*** 86,91 **** static void show_sort_group_keys(PlanState *planstate, const char *qlabel,
--- 86,92 ----
  					 List *ancestors, ExplainState *es);
  static void show_sort_info(SortState *sortstate, ExplainState *es);
  static void show_hash_info(HashState *hashstate, ExplainState *es);
+ static void show_hashagg_info(AggState *hashstate, ExplainState *es);
  static void show_tidbitmap_info(BitmapHeapScanState *planstate,
  					ExplainState *es);
  static void show_instrumentation_count(const char *qlabel, int which,
***************
*** 1423,1428 **** ExplainNode(PlanState *planstate, List *ancestors,
--- 1424,1430 ----
  		case T_Agg:
  			show_agg_keys((AggState *) planstate, ancestors, es);
  			show_upper_qual(plan->qual, "Filter", planstate, ancestors, es);
+ 			show_hashagg_info((AggState *) planstate, es);
  			if (plan->qual)
  				show_instrumentation_count("Rows Removed by Filter", 1,
  										   planstate, es);
***************
*** 1913,1918 **** show_sort_info(SortState *sortstate, ExplainState *es)
--- 1915,1956 ----
  }
  
  /*
+  * Show information on hash aggregate buckets and batches
+  */
+ static void
+ show_hashagg_info(AggState *aggstate, ExplainState *es)
+ {
+ 	Agg *agg = (Agg *)aggstate->ss.ps.plan;
+ 
+ 	Assert(IsA(aggstate, AggState));
+ 
+ 	if (agg->aggstrategy != AGG_HASHED)
+ 		return;
+ 
+ 	if (!aggstate->hash_init_state)
+ 	{
+ 		long	memPeakKb = (aggstate->hash_mem_peak + 1023) / 1024;
+ 		long	diskKb	  = (aggstate->hash_disk + 1023) / 1024;
+ 
+ 		if (es->format == EXPLAIN_FORMAT_TEXT)
+ 		{
+ 			appendStringInfoSpaces(es->str, es->indent * 2);
+ 			appendStringInfo(
+ 				es->str,
+ 				"Batches: %d  Memory Usage: %ldkB  Disk Usage:%ldkB\n",
+ 				aggstate->hash_num_batches, memPeakKb, diskKb);
+ 		}
+ 		else
+ 		{
+ 			ExplainPropertyLong("HashAgg Batches",
+ 								aggstate->hash_num_batches, es);
+ 			ExplainPropertyLong("Peak Memory Usage", memPeakKb, es);
+ 			ExplainPropertyLong("Disk Usage", diskKb, es);
+ 		}
+ 	}
+ }
+ 
+ /*
   * Show information on hash buckets/batches.
   */
  static void
*** a/src/backend/executor/execGrouping.c
--- b/src/backend/executor/execGrouping.c
***************
*** 310,316 **** BuildTupleHashTable(int numCols, AttrNumber *keyColIdx,
  	hash_ctl.hcxt = tablecxt;
  	hashtable->hashtab = hash_create("TupleHashTable", nbuckets,
  									 &hash_ctl,
! 					HASH_ELEM | HASH_FUNCTION | HASH_COMPARE | HASH_CONTEXT);
  
  	return hashtable;
  }
--- 310,317 ----
  	hash_ctl.hcxt = tablecxt;
  	hashtable->hashtab = hash_create("TupleHashTable", nbuckets,
  									 &hash_ctl,
! 									 HASH_ELEM | HASH_FUNCTION | HASH_COMPARE |
! 									 HASH_CONTEXT | HASH_NOCHILDCXT);
  
  	return hashtable;
  }
***************
*** 331,336 **** TupleHashEntry
--- 332,386 ----
  LookupTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot,
  					 bool *isnew)
  {
+ 	uint32 hashvalue;
+ 
+ 	hashvalue = TupleHashEntryHash(hashtable, slot);
+ 	return LookupTupleHashEntryHash(hashtable, slot, hashvalue, isnew);
+ }
+ 
+ /*
+  * TupleHashEntryHash
+  *
+  * Calculate the hash value of the tuple.
+  */
+ uint32
+ TupleHashEntryHash(TupleHashTable hashtable, TupleTableSlot *slot)
+ {
+ 	TupleHashEntryData	dummy;
+ 	TupleHashTable		saveCurHT;
+ 	uint32				hashvalue;
+ 
+ 	/*
+ 	 * Set up data needed by hash function.
+ 	 *
+ 	 * We save and restore CurTupleHashTable just in case someone manages to
+ 	 * invoke this code re-entrantly.
+ 	 */
+ 	hashtable->inputslot = slot;
+ 	hashtable->in_hash_funcs = hashtable->tab_hash_funcs;
+ 	hashtable->cur_eq_funcs = hashtable->tab_eq_funcs;
+ 
+ 	saveCurHT = CurTupleHashTable;
+ 	CurTupleHashTable = hashtable;
+ 
+ 	dummy.firstTuple = NULL;	/* flag to reference inputslot */
+ 	hashvalue = TupleHashTableHash(&dummy, sizeof(TupleHashEntryData));
+ 
+ 	CurTupleHashTable = saveCurHT;
+ 
+ 	return hashvalue;
+ }
+ 
+ /*
+  * LookupTupleHashEntryHash
+  *
+  * Like LookupTupleHashEntry, but allows the caller to specify the tuple's
+  * hash value, to avoid recalculating it.
+  */
+ TupleHashEntry
+ LookupTupleHashEntryHash(TupleHashTable hashtable, TupleTableSlot *slot,
+ 						 uint32 hashvalue, bool *isnew)
+ {
  	TupleHashEntry entry;
  	MemoryContext oldContext;
  	TupleHashTable saveCurHT;
***************
*** 371,380 **** LookupTupleHashEntry(TupleHashTable hashtable, TupleTableSlot *slot,
  
  	/* Search the hash table */
  	dummy.firstTuple = NULL;	/* flag to reference inputslot */
! 	entry = (TupleHashEntry) hash_search(hashtable->hashtab,
! 										 &dummy,
! 										 isnew ? HASH_ENTER : HASH_FIND,
! 										 &found);
  
  	if (isnew)
  	{
--- 421,429 ----
  
  	/* Search the hash table */
  	dummy.firstTuple = NULL;	/* flag to reference inputslot */
! 	entry = (TupleHashEntry) hash_search_with_hash_value(
! 		hashtable->hashtab, &dummy, hashvalue, isnew ? HASH_ENTER : HASH_FIND,
! 		&found);
  
  	if (isnew)
  	{
*** a/src/backend/executor/nodeAgg.c
--- b/src/backend/executor/nodeAgg.c
***************
*** 96,101 ****
--- 96,103 ----
  
  #include "postgres.h"
  
+ #include <math.h>
+ 
  #include "access/htup_details.h"
  #include "catalog/objectaccess.h"
  #include "catalog/pg_aggregate.h"
***************
*** 108,121 ****
--- 110,127 ----
  #include "optimizer/tlist.h"
  #include "parser/parse_agg.h"
  #include "parser/parse_coerce.h"
+ #include "storage/buffile.h"
  #include "utils/acl.h"
  #include "utils/builtins.h"
+ #include "utils/dynahash.h"
  #include "utils/lsyscache.h"
  #include "utils/memutils.h"
  #include "utils/syscache.h"
  #include "utils/tuplesort.h"
  #include "utils/datum.h"
  
+ #define HASH_DISK_MIN_PARTITIONS		1
+ #define HASH_DISK_MAX_PARTITIONS		256
  
  /*
   * AggStatePerAggData - per-aggregate working state for the Agg scan
***************
*** 301,306 **** typedef struct AggHashEntryData
--- 307,323 ----
  	AggStatePerGroupData pergroup[1];	/* VARIABLE LENGTH ARRAY */
  }	AggHashEntryData;	/* VARIABLE LENGTH STRUCT */
  
+ typedef struct HashWork
+ {
+ 	BufFile		 *input_file;	/* input partition, NULL for outer plan */
+ 	int			  input_bits;	/* number of bits for input partition mask */
+ 	int64		  input_groups; /* estimated number of input groups */
+ 
+ 	int			  n_output_partitions; /* number of output partitions */
+ 	BufFile		**output_partitions; /* output partition files */
+ 	int64		 *output_ntuples; /* number of tuples in each partition */
+ 	int			  output_bits; /* log2(n_output_partitions) + input_bits */
+ } HashWork;
  
  static void initialize_aggregates(AggState *aggstate,
  					  AggStatePerAgg peragg,
***************
*** 321,331 **** static void finalize_aggregate(AggState *aggstate,
  				   Datum *resultVal, bool *resultIsNull);
  static Bitmapset *find_unaggregated_cols(AggState *aggstate);
  static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
! static void build_hash_table(AggState *aggstate);
! static AggHashEntry lookup_hash_entry(AggState *aggstate,
! 				  TupleTableSlot *inputslot);
  static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
! static void agg_fill_hash_table(AggState *aggstate);
  static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
  static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
  
--- 338,352 ----
  				   Datum *resultVal, bool *resultIsNull);
  static Bitmapset *find_unaggregated_cols(AggState *aggstate);
  static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
! static void build_hash_table(AggState *aggstate, long nbuckets);
! static AggHashEntry lookup_hash_entry(AggState *aggstate, HashWork *work,
! 				   TupleTableSlot *inputslot, uint32 hashvalue);
! static HashWork *hash_work(BufFile *input_file, int64 input_groups,
! 						   int input_bits);
! static void save_tuple(AggState *aggstate, HashWork *work,
! 					   TupleTableSlot *slot, uint32 hashvalue);
  static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
! static bool agg_fill_hash_table(AggState *aggstate);
  static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
  static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
  
***************
*** 923,942 **** find_unaggregated_cols_walker(Node *node, Bitmapset **colnos)
  }
  
  /*
   * Initialize the hash table to empty.
   *
   * The hash table always lives in the aggcontext memory context.
   */
  static void
! build_hash_table(AggState *aggstate)
  {
  	Agg		   *node = (Agg *) aggstate->ss.ps.plan;
  	MemoryContext tmpmem = aggstate->tmpcontext->ecxt_per_tuple_memory;
  	Size		entrysize;
  
  	Assert(node->aggstrategy == AGG_HASHED);
  	Assert(node->numGroups > 0);
  
  	entrysize = sizeof(AggHashEntryData) +
  		(aggstate->numaggs - 1) * sizeof(AggStatePerGroupData);
  
--- 944,989 ----
  }
  
  /*
+  * Estimate all memory used by a group in the hash table.
+  */
+ Size
+ hash_group_size(int numAggs, int inputWidth, Size transitionSpace)
+ {
+ 	Size size;
+ 
+ 	/* tuple overhead */
+ 	size = MAXALIGN(sizeof(MinimalTupleData));
+ 	/* group key */
+ 	size += MAXALIGN(inputWidth);
+ 	/* hash table overhead */
+ 	size += hash_agg_entry_size(numAggs);
+ 	/* by-ref transition space */
+ 	size += transitionSpace;
+ 
+ 	return size;
+ }
+ 
+ /*
   * Initialize the hash table to empty.
   *
   * The hash table always lives in the aggcontext memory context.
   */
  static void
! build_hash_table(AggState *aggstate, long nbuckets)
  {
  	Agg		   *node = (Agg *) aggstate->ss.ps.plan;
  	MemoryContext tmpmem = aggstate->tmpcontext->ecxt_per_tuple_memory;
  	Size		entrysize;
+ 	Size		hash_group_mem = hash_group_size(aggstate->numaggs,
+ 												 node->plan_width,
+ 												 node->transitionSpace);
  
  	Assert(node->aggstrategy == AGG_HASHED);
  	Assert(node->numGroups > 0);
  
+ 	/* don't exceed work_mem */
+ 	nbuckets = Min(nbuckets, (long) ((work_mem * 1024L) / hash_group_mem));
+ 
  	entrysize = sizeof(AggHashEntryData) +
  		(aggstate->numaggs - 1) * sizeof(AggStatePerGroupData);
  
***************
*** 944,953 **** build_hash_table(AggState *aggstate)
  											  node->grpColIdx,
  											  aggstate->eqfunctions,
  											  aggstate->hashfunctions,
! 											  node->numGroups,
  											  entrysize,
! 											  aggstate->aggcontext,
  											  tmpmem);
  }
  
  /*
--- 991,1006 ----
  											  node->grpColIdx,
  											  aggstate->eqfunctions,
  											  aggstate->hashfunctions,
! 											  nbuckets,
  											  entrysize,
! 											  aggstate->hashcontext,
  											  tmpmem);
+ 
+ 	aggstate->hash_mem_min = MemoryContextMemAllocated(
+ 		aggstate->hashcontext, true);
+ 
+ 	if (aggstate->hash_mem_min > aggstate->hash_mem_peak)
+ 		aggstate->hash_mem_peak = aggstate->hash_mem_min;
  }
  
  /*
***************
*** 1024,1035 **** hash_agg_entry_size(int numAggs)
   * When called, CurrentMemoryContext should be the per-query context.
   */
  static AggHashEntry
! lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
  {
  	TupleTableSlot *hashslot = aggstate->hashslot;
  	ListCell   *l;
  	AggHashEntry entry;
! 	bool		isnew;
  
  	/* if first time through, initialize hashslot by cloning input slot */
  	if (hashslot->tts_tupleDescriptor == NULL)
--- 1077,1091 ----
   * When called, CurrentMemoryContext should be the per-query context.
   */
  static AggHashEntry
! lookup_hash_entry(AggState *aggstate, HashWork *work,
! 				  TupleTableSlot *inputslot, uint32 hashvalue)
  {
  	TupleTableSlot *hashslot = aggstate->hashslot;
  	ListCell   *l;
  	AggHashEntry entry;
! 	int64		hash_mem;
! 	bool		isnew = false;
! 	bool	   *p_isnew;
  
  	/* if first time through, initialize hashslot by cloning input slot */
  	if (hashslot->tts_tupleDescriptor == NULL)
***************
*** 1049,1058 **** lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
  		hashslot->tts_isnull[varNumber] = inputslot->tts_isnull[varNumber];
  	}
  
  	/* find or create the hashtable entry using the filtered tuple */
! 	entry = (AggHashEntry) LookupTupleHashEntry(aggstate->hashtable,
! 												hashslot,
! 												&isnew);
  
  	if (isnew)
  	{
--- 1105,1124 ----
  		hashslot->tts_isnull[varNumber] = inputslot->tts_isnull[varNumber];
  	}
  
+ 	hash_mem = MemoryContextMemAllocated(aggstate->hashcontext, true);
+ 	if (hash_mem > aggstate->hash_mem_peak)
+ 		aggstate->hash_mem_peak = hash_mem;
+ 
+ 	if (hash_mem <= aggstate->hash_mem_min ||
+ 		hash_mem < work_mem * 1024L)
+ 		p_isnew = &isnew;
+ 	else
+ 		p_isnew = NULL;
+ 
  	/* find or create the hashtable entry using the filtered tuple */
! 	entry = (AggHashEntry) LookupTupleHashEntryHash(aggstate->hashtable,
! 													hashslot, hashvalue,
! 													p_isnew);
  
  	if (isnew)
  	{
***************
*** 1060,1068 **** lookup_hash_entry(AggState *aggstate, TupleTableSlot *inputslot)
--- 1126,1291 ----
  		initialize_aggregates(aggstate, aggstate->peragg, entry->pergroup);
  	}
  
+ 	if (entry == NULL)
+ 		save_tuple(aggstate, work, inputslot, hashvalue);
+ 
  	return entry;
  }
  
+ 
+ /*
+  * hash_work
+  *
+  * Construct a HashWork item, which represents one iteration of HashAgg to be
+  * done. Should be called in the aggregate's memory context.
+  */
+ static HashWork *
+ hash_work(BufFile *input_file, int64 input_groups, int input_bits)
+ {
+ 	HashWork *work = palloc(sizeof(HashWork));
+ 
+ 	work->input_file = input_file;
+ 	work->input_bits = input_bits;
+ 	work->input_groups = input_groups;
+ 
+ 	/*
+ 	 * Will be set only if we run out of memory and need to partition an
+ 	 * additional level.
+ 	 */
+ 	work->n_output_partitions = 0;
+ 	work->output_partitions = NULL;
+ 	work->output_ntuples = NULL;
+ 	work->output_bits = 0;
+ 
+ 	return work;
+ }
+ 
+ /*
+  * save_tuple
+  *
+  * Not enough memory to add tuple as new entry in hash table. Save for later
+  * in the appropriate partition.
+  */
+ static void
+ save_tuple(AggState *aggstate, HashWork *work, TupleTableSlot *slot,
+ 		   uint32 hashvalue)
+ {
+ 	int					 partition;
+ 	MinimalTuple		 tuple;
+ 	BufFile				*file;
+ 	int					 written;
+ 
+ 	if (work->output_partitions == NULL)
+ 	{
+ 		Agg		*agg = (Agg *) aggstate->ss.ps.plan;
+ 		Size	 group_size = hash_group_size(aggstate->numaggs,
+ 											  agg->plan_width,
+ 											  agg->transitionSpace);
+ 		double	 total_size = group_size * work->input_groups;
+ 		int		 npartitions;
+ 		int		 partition_bits;
+ 
+ 		/*
+ 		 * Try to make enough partitions so that each one fits in work_mem,
+ 		 * with a little slop.
+ 		 */
+ 		npartitions = ceil ( (1.5 * total_size) / (work_mem * 1024L) );
+ 
+ 		if (npartitions < HASH_DISK_MIN_PARTITIONS)
+ 			npartitions = HASH_DISK_MIN_PARTITIONS;
+ 		if (npartitions > HASH_DISK_MAX_PARTITIONS)
+ 			npartitions = HASH_DISK_MAX_PARTITIONS;
+ 
+ 		partition_bits = my_log2(npartitions);
+ 
+ 		/* make sure that we don't exhaust the hash bits */
+ 		if (partition_bits + work->input_bits >= 32)
+ 			partition_bits = 32 - work->input_bits;
+ 
+ 		/* number of partitions will be a power of two */
+ 		npartitions = 1L << partition_bits;
+ 
+ 		work->output_bits = partition_bits;
+ 		work->n_output_partitions = npartitions;
+ 		work->output_partitions = palloc0(sizeof(BufFile *) * npartitions);
+ 		work->output_ntuples = palloc0(sizeof(int64) * npartitions);
+ 	}
+ 
+ 	if (work->output_bits == 0)
+ 		partition = 0;
+ 	else
+ 		partition = (hashvalue << work->input_bits) >>
+ 			(32 - work->output_bits);
+ 
+ 	work->output_ntuples[partition]++;
+ 
+ 	if (work->output_partitions[partition] == NULL)
+ 		work->output_partitions[partition] = BufFileCreateTemp(false);
+ 	file = work->output_partitions[partition];
+ 
+ 	tuple = ExecFetchSlotMinimalTuple(slot);
+ 
+ 	written = BufFileWrite(file, (void *) &hashvalue, sizeof(uint32));
+ 	if (written != sizeof(uint32))
+ 		ereport(ERROR,
+ 				(errcode_for_file_access(),
+ 				 errmsg("could not write to HashAgg temporary file: %m")));
+ 	aggstate->hash_disk += written;
+ 
+ 	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")));
+ 	aggstate->hash_disk += written;
+ }
+ 
+ 
+ /*
+  * read_saved_tuple
+  *		read the next tuple from a batch file.  Return NULL if no more.
+  *
+  * On success, *hashvalue is set to the tuple's hash value, and the tuple
+  * itself is stored in the given slot.
+  *
+  * Copied with minor modifications from ExecHashJoinGetSavedTuple.
+  */
+ static TupleTableSlot *
+ read_saved_tuple(BufFile *file, uint32 *hashvalue, TupleTableSlot *tupleSlot)
+ {
+ 	uint32		header[2];
+ 	size_t		nread;
+ 	MinimalTuple tuple;
+ 
+ 	/*
+ 	 * Since both the hash value and the MinimalTuple length word are uint32,
+ 	 * we can read them both in one BufFileRead() call without any type
+ 	 * cheating.
+ 	 */
+ 	nread = BufFileRead(file, (void *) header, sizeof(header));
+ 	if (nread == 0)				/* end of file */
+ 	{
+ 		ExecClearTuple(tupleSlot);
+ 		return NULL;
+ 	}
+ 	if (nread != sizeof(header))
+ 		ereport(ERROR,
+ 				(errcode_for_file_access(),
+ 				 errmsg("could not read from HashAgg temporary file: %m")));
+ 	*hashvalue = header[0];
+ 	tuple = (MinimalTuple) palloc(header[1]);
+ 	tuple->t_len = header[1];
+ 	nread = BufFileRead(file,
+ 						(void *) ((char *) tuple + sizeof(uint32)),
+ 						header[1] - sizeof(uint32));
+ 	if (nread != header[1] - sizeof(uint32))
+ 		ereport(ERROR,
+ 				(errcode_for_file_access(),
+ 				 errmsg("could not read from HashAgg temporary file: %m")));
+ 	return ExecStoreMinimalTuple(tuple, tupleSlot, true);
+ }
+ 
+ 
  /*
   * ExecAgg -
   *
***************
*** 1107,1115 **** ExecAgg(AggState *node)
  	/* Dispatch based on strategy */
  	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
  	{
! 		if (!node->table_filled)
! 			agg_fill_hash_table(node);
! 		return agg_retrieve_hash_table(node);
  	}
  	else
  		return agg_retrieve_direct(node);
--- 1330,1345 ----
  	/* Dispatch based on strategy */
  	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
  	{
! 		TupleTableSlot *slot = NULL;
! 
! 		while (slot == NULL)
! 		{
! 			if (!node->table_filled)
! 				if (!agg_fill_hash_table(node))
! 					break;
! 			slot = agg_retrieve_hash_table(node);
! 		}
! 		return slot;
  	}
  	else
  		return agg_retrieve_direct(node);
***************
*** 1325,1337 **** agg_retrieve_direct(AggState *aggstate)
  /*
   * ExecAgg for hashed case: phase 1, read input and build hash table
   */
! static void
  agg_fill_hash_table(AggState *aggstate)
  {
  	PlanState  *outerPlan;
  	ExprContext *tmpcontext;
  	AggHashEntry entry;
  	TupleTableSlot *outerslot;
  
  	/*
  	 * get state info from node
--- 1555,1569 ----
  /*
   * ExecAgg for hashed case: phase 1, read input and build hash table
   */
! static bool
  agg_fill_hash_table(AggState *aggstate)
  {
  	PlanState  *outerPlan;
  	ExprContext *tmpcontext;
  	AggHashEntry entry;
  	TupleTableSlot *outerslot;
+ 	HashWork	*work;
+ 	int			 i;
  
  	/*
  	 * get state info from node
***************
*** 1340,1359 **** agg_fill_hash_table(AggState *aggstate)
  	/* tmpcontext is the per-input-tuple expression context */
  	tmpcontext = aggstate->tmpcontext;
  
  	/*
  	 * Process each outer-plan tuple, and then fetch the next one, until we
  	 * exhaust the outer plan.
  	 */
  	for (;;)
  	{
! 		outerslot = ExecProcNode(outerPlan);
! 		if (TupIsNull(outerslot))
! 			break;
  		/* set up for advance_aggregates call */
  		tmpcontext->ecxt_outertuple = outerslot;
  
  		/* Find or build hashtable entry for this tuple's group */
! 		entry = lookup_hash_entry(aggstate, outerslot);
  
  		/* Advance the aggregates */
  		advance_aggregates(aggstate, entry->pergroup);
--- 1572,1640 ----
  	/* tmpcontext is the per-input-tuple expression context */
  	tmpcontext = aggstate->tmpcontext;
  
+ 	if (aggstate->hash_work == NIL)
+ 	{
+ 		aggstate->agg_done = true;
+ 		return false;
+ 	}
+ 
+ 	work = linitial(aggstate->hash_work);
+ 	aggstate->hash_work = list_delete_first(aggstate->hash_work);
+ 
+ 	/* if not the first time through, reinitialize */
+ 	if (!aggstate->hash_init_state)
+ 	{
+ 		long	 nbuckets;
+ 		Agg		*node = (Agg *) aggstate->ss.ps.plan;
+ 
+ 		MemoryContextResetAndDeleteChildren(aggstate->hashcontext);
+ 
+ 		/*
+ 		 * If this table will hold only a partition of the input, then use a
+ 		 * proportionally smaller estimate for nbuckets.
+ 		 */
+ 		nbuckets = node->numGroups >> work->input_bits;
+ 
+ 		build_hash_table(aggstate, nbuckets);
+ 	}
+ 
+ 	aggstate->hash_init_state = false;
+ 
  	/*
  	 * Process each outer-plan tuple, and then fetch the next one, until we
  	 * exhaust the outer plan.
  	 */
  	for (;;)
  	{
! 		uint32 hashvalue;
! 
! 		CHECK_FOR_INTERRUPTS();
! 
! 		if (work->input_file == NULL)
! 		{
! 			outerslot = ExecProcNode(outerPlan);
! 			if (TupIsNull(outerslot))
! 				break;
! 
! 			hashvalue = TupleHashEntryHash(aggstate->hashtable, outerslot);
! 		}
! 		else
! 		{
! 			outerslot = read_saved_tuple(work->input_file, &hashvalue,
! 										 aggstate->hashslot);
! 			if (TupIsNull(outerslot))
! 			{
! 				BufFileClose(work->input_file);
! 				work->input_file = NULL;
! 				break;
! 			}
! 		}
! 
  		/* set up for advance_aggregates call */
  		tmpcontext->ecxt_outertuple = outerslot;
  
  		/* Find or build hashtable entry for this tuple's group */
! 		entry = lookup_hash_entry(aggstate, work, outerslot, hashvalue);
  
  		/* Advance the aggregates */
  		advance_aggregates(aggstate, entry->pergroup);
***************
*** 1362,1370 **** agg_fill_hash_table(AggState *aggstate)
--- 1643,1697 ----
  		ResetExprContext(tmpcontext);
  	}
  
+ 	if (work->input_file)
+ 		BufFileClose(work->input_file);
+ 
+ 	/* add each output partition as a new work item */
+ 	for (i = 0; i < work->n_output_partitions; i++)
+ 	{
+ 		BufFile			*file = work->output_partitions[i];
+ 		MemoryContext	 oldContext;
+ 		HashWork		*new_work;
+ 		int64			 input_ngroups;
+ 
+ 		/* 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")));
+ 
+ 		/*
+ 		 * Estimate the number of input groups for this new work item as the
+ 		 * total number of tuples in its input file. Although that's a worst
+ 		 * case, it's not bad here for two reasons: (1) overestimating is
+ 		 * better than underestimating; and (2) we've already scanned the
+ 		 * relation once, so it's likely that we've already finalized many of
+ 		 * the common values.
+ 		 */
+ 		input_ngroups = work->output_ntuples[i];
+ 
+ 		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
+ 		new_work = hash_work(file,
+ 							 input_ngroups,
+ 							 work->output_bits + work->input_bits);
+ 		aggstate->hash_work = lappend(
+ 			aggstate->hash_work,
+ 			new_work);
+ 		aggstate->hash_num_batches++;
+ 		MemoryContextSwitchTo(oldContext);
+ 	}
+ 
+ 	pfree(work);
+ 
  	aggstate->table_filled = true;
  	/* Initialize to walk the hash table */
  	ResetTupleHashIterator(aggstate->hashtable, &aggstate->hashiter);
+ 
+ 	return true;
  }
  
  /*
***************
*** 1396,1411 **** agg_retrieve_hash_table(AggState *aggstate)
  	 * We loop retrieving groups until we find one satisfying
  	 * aggstate->ss.ps.qual
  	 */
! 	while (!aggstate->agg_done)
  	{
  		/*
  		 * Find the next entry in the hash table
  		 */
  		entry = (AggHashEntry) ScanTupleHashTable(&aggstate->hashiter);
  		if (entry == NULL)
  		{
! 			/* No more entries in hashtable, so done */
! 			aggstate->agg_done = TRUE;
  			return NULL;
  		}
  
--- 1723,1740 ----
  	 * We loop retrieving groups until we find one satisfying
  	 * aggstate->ss.ps.qual
  	 */
! 	for (;;)
  	{
+ 		CHECK_FOR_INTERRUPTS();
+ 
  		/*
  		 * Find the next entry in the hash table
  		 */
  		entry = (AggHashEntry) ScanTupleHashTable(&aggstate->hashiter);
  		if (entry == NULL)
  		{
! 			/* No more entries in hashtable, so done with this batch */
! 			aggstate->table_filled = false;
  			return NULL;
  		}
  
***************
*** 1636,1645 **** ExecInitAgg(Agg *node, EState *estate, int eflags)
  
  	if (node->aggstrategy == AGG_HASHED)
  	{
! 		build_hash_table(aggstate);
  		aggstate->table_filled = false;
  		/* Compute the columns we actually need to hash on */
  		aggstate->hash_needed = find_hash_columns(aggstate);
  	}
  	else
  	{
--- 1965,1997 ----
  
  	if (node->aggstrategy == AGG_HASHED)
  	{
! 		MemoryContext oldContext;
! 
! 		aggstate->hash_mem_min = 0;
! 		aggstate->hash_mem_peak = 0;
! 		aggstate->hash_num_batches = 0;
! 		aggstate->hash_init_state = true;
  		aggstate->table_filled = false;
+ 		aggstate->hash_disk = 0;
+ 
+ 		aggstate->hashcontext =
+ 			AllocSetContextCreate(aggstate->aggcontext,
+ 								  "HashAgg Hash Table Context",
+ 								  ALLOCSET_DEFAULT_MINSIZE,
+ 								  ALLOCSET_DEFAULT_INITSIZE,
+ 								  ALLOCSET_DEFAULT_MAXSIZE);
+ 
+ 		build_hash_table(aggstate, node->numGroups);
+ 
  		/* Compute the columns we actually need to hash on */
  		aggstate->hash_needed = find_hash_columns(aggstate);
+ 
+ 		/* prime with initial work item to read from outer plan */
+ 		oldContext = MemoryContextSwitchTo(aggstate->aggcontext);
+ 		aggstate->hash_work = lappend(aggstate->hash_work,
+ 									  hash_work(NULL, node->numGroups, 0));
+ 		aggstate->hash_num_batches++;
+ 		MemoryContextSwitchTo(oldContext);
  	}
  	else
  	{
***************
*** 2048,2079 **** ExecEndAgg(AggState *node)
  void
  ExecReScanAgg(AggState *node)
  {
  	ExprContext *econtext = node->ss.ps.ps_ExprContext;
! 	int			aggno;
  
  	node->agg_done = false;
  
  	node->ss.ps.ps_TupFromTlist = false;
  
! 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
  	{
  		/*
! 		 * In the hashed case, if we haven't yet built the hash table then we
! 		 * can just return; nothing done yet, so nothing to undo. If subnode's
! 		 * chgParam is not NULL then it will be re-scanned by ExecProcNode,
! 		 * else no reason to re-scan it at all.
  		 */
! 		if (!node->table_filled)
  			return;
  
  		/*
! 		 * If we do have the hash table and the subplan does not have any
! 		 * parameter changes, then we can just rescan the existing hash table;
! 		 * no need to build it again.
  		 */
! 		if (node->ss.ps.lefttree->chgParam == NULL)
  		{
  			ResetTupleHashIterator(node->hashtable, &node->hashiter);
  			return;
  		}
  	}
--- 2400,2433 ----
  void
  ExecReScanAgg(AggState *node)
  {
+ 	Agg			*agg = (Agg *) node->ss.ps.plan;
  	ExprContext *econtext = node->ss.ps.ps_ExprContext;
! 	int			 aggno;
  
  	node->agg_done = false;
  
  	node->ss.ps.ps_TupFromTlist = false;
  
! 	if (agg->aggstrategy == AGG_HASHED)
  	{
  		/*
! 		 * In the hashed case, if we haven't done any execution work yet, we
! 		 * can just return; nothing to undo. If subnode's chgParam is not NULL
! 		 * then it will be re-scanned by ExecProcNode, else no reason to
! 		 * re-scan it at all.
  		 */
! 		if (node->hash_init_state)
  			return;
  
  		/*
! 		 * If we do have the hash table, it never went to disk, and the
! 		 * subplan does not have any parameter changes, then we can just
! 		 * rescan the existing hash table; no need to build it again.
  		 */
! 		if (node->ss.ps.lefttree->chgParam == NULL && node->hash_disk == 0)
  		{
  			ResetTupleHashIterator(node->hashtable, &node->hashiter);
+ 			node->table_filled = true;
  			return;
  		}
  	}
***************
*** 2110,2120 **** ExecReScanAgg(AggState *node)
  	 */
  	MemoryContextResetAndDeleteChildren(node->aggcontext);
  
! 	if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
  	{
  		/* Rebuild an empty hash table */
! 		build_hash_table(node);
  		node->table_filled = false;
  	}
  	else
  	{
--- 2464,2493 ----
  	 */
  	MemoryContextResetAndDeleteChildren(node->aggcontext);
  
! 	if (agg->aggstrategy == AGG_HASHED)
  	{
+ 		MemoryContext oldContext;
+ 
+ 		node->hashcontext =
+ 			AllocSetContextCreate(node->aggcontext,
+ 								  "HashAgg Hash Table Context",
+ 								  ALLOCSET_DEFAULT_MINSIZE,
+ 								  ALLOCSET_DEFAULT_INITSIZE,
+ 								  ALLOCSET_DEFAULT_MAXSIZE);
+ 
  		/* Rebuild an empty hash table */
! 		build_hash_table(node, agg->numGroups);
! 		node->hash_init_state = true;
  		node->table_filled = false;
+ 		node->hash_disk = 0;
+ 		node->hash_work = NIL;
+ 
+ 		/* prime with initial work item to read from outer plan */
+ 		oldContext = MemoryContextSwitchTo(node->aggcontext);
+ 		node->hash_work = lappend(node->hash_work,
+ 								  hash_work(NULL, agg->numGroups, 0));
+ 		node->hash_num_batches++;
+ 		MemoryContextSwitchTo(oldContext);
  	}
  	else
  	{
*** a/src/backend/optimizer/path/costsize.c
--- b/src/backend/optimizer/path/costsize.c
***************
*** 75,80 ****
--- 75,81 ----
  
  #include "access/htup_details.h"
  #include "executor/executor.h"
+ #include "executor/nodeAgg.h"
  #include "executor/nodeHash.h"
  #include "miscadmin.h"
  #include "nodes/nodeFuncs.h"
***************
*** 113,118 **** bool		enable_bitmapscan = true;
--- 114,120 ----
  bool		enable_tidscan = true;
  bool		enable_sort = true;
  bool		enable_hashagg = true;
+ bool		enable_hashagg_disk = true;
  bool		enable_nestloop = true;
  bool		enable_material = true;
  bool		enable_mergejoin = true;
***************
*** 1468,1474 **** cost_agg(Path *path, PlannerInfo *root,
  		 AggStrategy aggstrategy, const AggClauseCosts *aggcosts,
  		 int numGroupCols, double numGroups,
  		 Cost input_startup_cost, Cost input_total_cost,
! 		 double input_tuples)
  {
  	double		output_tuples;
  	Cost		startup_cost;
--- 1470,1476 ----
  		 AggStrategy aggstrategy, const AggClauseCosts *aggcosts,
  		 int numGroupCols, double numGroups,
  		 Cost input_startup_cost, Cost input_total_cost,
! 		 int input_width, double input_tuples)
  {
  	double		output_tuples;
  	Cost		startup_cost;
***************
*** 1531,1536 **** cost_agg(Path *path, PlannerInfo *root,
--- 1533,1542 ----
  	else
  	{
  		/* must be AGG_HASHED */
+ 		double	group_size = hash_group_size(aggcosts->numAggs,
+ 											 input_width,
+ 											 aggcosts->transitionSpace);
+ 
  		startup_cost = input_total_cost;
  		startup_cost += aggcosts->transCost.startup;
  		startup_cost += aggcosts->transCost.per_tuple * input_tuples;
***************
*** 1538,1543 **** cost_agg(Path *path, PlannerInfo *root,
--- 1544,1578 ----
  		total_cost = startup_cost;
  		total_cost += aggcosts->finalCost * numGroups;
  		total_cost += cpu_tuple_cost * numGroups;
+ 
+ 		if (group_size * numGroups > (work_mem * 1024L))
+ 		{
+ 			double groups_per_batch = (work_mem * 1024L) / group_size;
+ 
+ 			/* first batch doesn't go to disk */
+ 			double groups_disk = numGroups - groups_per_batch;
+ 
+ 			/*
+ 			 * Assume that the groups that go to disk are of an average number
+ 			 * of tuples. This is pessimistic -- the largest groups are more
+ 			 * likely to be processed in the first pass and never go to disk.
+ 			 */
+ 			double tuples_disk = groups_disk * (input_tuples / numGroups);
+ 
+ 			int tuple_size = sizeof(uint32) /* stored hash value */
+ 				+ MAXALIGN(sizeof(MinimalTupleData))
+ 				+ MAXALIGN(input_width);
+ 			double pages_to_disk = (tuples_disk * tuple_size) / BLCKSZ;
+ 
+ 			/*
+ 			 * Write and then read back the data that's not processed in the
+ 			 * first pass. Data could be read and written more times than that
+ 			 * if not enough partitions are created, but the depth will be a
+ 			 * very small number even for a very large amount of data, so
+ 			 * ignore it here.
+ 			 */
+ 			total_cost += seq_page_cost * 2 * pages_to_disk;
+ 		}
  		output_tuples = numGroups;
  	}
  
*** a/src/backend/optimizer/plan/createplan.c
--- b/src/backend/optimizer/plan/createplan.c
***************
*** 4369,4374 **** make_agg(PlannerInfo *root, List *tlist, List *qual,
--- 4369,4377 ----
  	node->grpColIdx = grpColIdx;
  	node->grpOperators = grpOperators;
  	node->numGroups = numGroups;
+ 	if (aggcosts != NULL)
+ 		node->transitionSpace = aggcosts->transitionSpace;
+ 	node->plan_width = lefttree->plan_width;
  
  	copy_plan_costsize(plan, lefttree); /* only care about copying size */
  	cost_agg(&agg_path, root,
***************
*** 4376,4381 **** make_agg(PlannerInfo *root, List *tlist, List *qual,
--- 4379,4385 ----
  			 numGroupCols, numGroups,
  			 lefttree->startup_cost,
  			 lefttree->total_cost,
+ 			 lefttree->plan_width,
  			 lefttree->plan_rows);
  	plan->startup_cost = agg_path.startup_cost;
  	plan->total_cost = agg_path.total_cost;
*** a/src/backend/optimizer/plan/planagg.c
--- b/src/backend/optimizer/plan/planagg.c
***************
*** 234,240 **** optimize_minmax_aggregates(PlannerInfo *root, List *tlist,
  	cost_agg(&agg_p, root, AGG_PLAIN, aggcosts,
  			 0, 0,
  			 best_path->startup_cost, best_path->total_cost,
! 			 best_path->parent->rows);
  
  	if (total_cost > agg_p.total_cost)
  		return NULL;			/* too expensive */
--- 234,240 ----
  	cost_agg(&agg_p, root, AGG_PLAIN, aggcosts,
  			 0, 0,
  			 best_path->startup_cost, best_path->total_cost,
! 			 best_path->parent->width, best_path->parent->rows);
  
  	if (total_cost > agg_p.total_cost)
  		return NULL;			/* too expensive */
*** a/src/backend/optimizer/plan/planner.c
--- b/src/backend/optimizer/plan/planner.c
***************
*** 2744,2750 **** choose_hashed_grouping(PlannerInfo *root,
  	/* plus the per-hash-entry overhead */
  	hashentrysize += hash_agg_entry_size(agg_costs->numAggs);
  
! 	if (hashentrysize * dNumGroups > work_mem * 1024L)
  		return false;
  
  	/*
--- 2744,2751 ----
  	/* plus the per-hash-entry overhead */
  	hashentrysize += hash_agg_entry_size(agg_costs->numAggs);
  
! 	if (!enable_hashagg_disk &&
! 		hashentrysize * dNumGroups > work_mem * 1024L)
  		return false;
  
  	/*
***************
*** 2779,2785 **** choose_hashed_grouping(PlannerInfo *root,
  	cost_agg(&hashed_p, root, AGG_HASHED, agg_costs,
  			 numGroupCols, dNumGroups,
  			 cheapest_path->startup_cost, cheapest_path->total_cost,
! 			 path_rows);
  	/* Result of hashed agg is always unsorted */
  	if (target_pathkeys)
  		cost_sort(&hashed_p, root, target_pathkeys, hashed_p.total_cost,
--- 2780,2786 ----
  	cost_agg(&hashed_p, root, AGG_HASHED, agg_costs,
  			 numGroupCols, dNumGroups,
  			 cheapest_path->startup_cost, cheapest_path->total_cost,
! 			 path_width, path_rows);
  	/* Result of hashed agg is always unsorted */
  	if (target_pathkeys)
  		cost_sort(&hashed_p, root, target_pathkeys, hashed_p.total_cost,
***************
*** 2810,2816 **** choose_hashed_grouping(PlannerInfo *root,
  		cost_agg(&sorted_p, root, AGG_SORTED, agg_costs,
  				 numGroupCols, dNumGroups,
  				 sorted_p.startup_cost, sorted_p.total_cost,
! 				 path_rows);
  	else
  		cost_group(&sorted_p, root, numGroupCols, dNumGroups,
  				   sorted_p.startup_cost, sorted_p.total_cost,
--- 2811,2817 ----
  		cost_agg(&sorted_p, root, AGG_SORTED, agg_costs,
  				 numGroupCols, dNumGroups,
  				 sorted_p.startup_cost, sorted_p.total_cost,
! 				 path_width, path_rows);
  	else
  		cost_group(&sorted_p, root, numGroupCols, dNumGroups,
  				   sorted_p.startup_cost, sorted_p.total_cost,
***************
*** 2910,2916 **** choose_hashed_distinct(PlannerInfo *root,
  	/* plus the per-hash-entry overhead */
  	hashentrysize += hash_agg_entry_size(0);
  
! 	if (hashentrysize * dNumDistinctRows > work_mem * 1024L)
  		return false;
  
  	/*
--- 2911,2918 ----
  	/* plus the per-hash-entry overhead */
  	hashentrysize += hash_agg_entry_size(0);
  
! 	if (!enable_hashagg_disk &&
! 		hashentrysize * dNumDistinctRows > work_mem * 1024L)
  		return false;
  
  	/*
***************
*** 2929,2935 **** choose_hashed_distinct(PlannerInfo *root,
  	cost_agg(&hashed_p, root, AGG_HASHED, NULL,
  			 numDistinctCols, dNumDistinctRows,
  			 cheapest_startup_cost, cheapest_total_cost,
! 			 path_rows);
  
  	/*
  	 * Result of hashed agg is always unsorted, so if ORDER BY is present we
--- 2931,2937 ----
  	cost_agg(&hashed_p, root, AGG_HASHED, NULL,
  			 numDistinctCols, dNumDistinctRows,
  			 cheapest_startup_cost, cheapest_total_cost,
! 			 path_width, path_rows);
  
  	/*
  	 * Result of hashed agg is always unsorted, so if ORDER BY is present we
*** a/src/backend/optimizer/prep/prepunion.c
--- b/src/backend/optimizer/prep/prepunion.c
***************
*** 851,857 **** choose_hashed_setop(PlannerInfo *root, List *groupClauses,
  	cost_agg(&hashed_p, root, AGG_HASHED, NULL,
  			 numGroupCols, dNumGroups,
  			 input_plan->startup_cost, input_plan->total_cost,
! 			 input_plan->plan_rows);
  
  	/*
  	 * Now for the sorted case.  Note that the input is *always* unsorted,
--- 851,857 ----
  	cost_agg(&hashed_p, root, AGG_HASHED, NULL,
  			 numGroupCols, dNumGroups,
  			 input_plan->startup_cost, input_plan->total_cost,
! 			 input_plan->plan_width, input_plan->plan_rows);
  
  	/*
  	 * Now for the sorted case.  Note that the input is *always* unsorted,
*** a/src/backend/optimizer/util/pathnode.c
--- b/src/backend/optimizer/util/pathnode.c
***************
*** 1379,1385 **** create_unique_path(PlannerInfo *root, RelOptInfo *rel, Path *subpath,
  					 numCols, pathnode->path.rows,
  					 subpath->startup_cost,
  					 subpath->total_cost,
! 					 rel->rows);
  	}
  
  	if (all_btree && all_hash)
--- 1379,1385 ----
  					 numCols, pathnode->path.rows,
  					 subpath->startup_cost,
  					 subpath->total_cost,
! 					 rel->width, rel->rows);
  	}
  
  	if (all_btree && all_hash)
*** a/src/backend/utils/hash/dynahash.c
--- b/src/backend/utils/hash/dynahash.c
***************
*** 291,301 **** hash_create(const char *tabname, long nelem, HASHCTL *info, int flags)
  			CurrentDynaHashCxt = info->hcxt;
  		else
  			CurrentDynaHashCxt = TopMemoryContext;
! 		CurrentDynaHashCxt = AllocSetContextCreate(CurrentDynaHashCxt,
! 												   tabname,
! 												   ALLOCSET_DEFAULT_MINSIZE,
! 												   ALLOCSET_DEFAULT_INITSIZE,
! 												   ALLOCSET_DEFAULT_MAXSIZE);
  	}
  
  	/* Initialize the hash header, plus a copy of the table name */
--- 291,303 ----
  			CurrentDynaHashCxt = info->hcxt;
  		else
  			CurrentDynaHashCxt = TopMemoryContext;
! 
! 		if ((flags & HASH_NOCHILDCXT) == 0)
! 			CurrentDynaHashCxt = AllocSetContextCreate(CurrentDynaHashCxt,
! 													   tabname,
! 													   ALLOCSET_DEFAULT_MINSIZE,
! 													   ALLOCSET_DEFAULT_INITSIZE,
! 													   ALLOCSET_DEFAULT_MAXSIZE);
  	}
  
  	/* Initialize the hash header, plus a copy of the table name */
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 771,776 **** static struct config_bool ConfigureNamesBool[] =
--- 771,785 ----
  		NULL, NULL, NULL
  	},
  	{
+ 		{"enable_hashagg_disk", PGC_USERSET, QUERY_TUNING_METHOD,
+ 			gettext_noop("Enables the planner's use of disk-based hashed aggregation plans."),
+ 			NULL
+ 		},
+ 		&enable_hashagg_disk,
+ 		true,
+ 		NULL, NULL, NULL
+ 	},
+ 	{
  		{"enable_material", PGC_USERSET, QUERY_TUNING_METHOD,
  			gettext_noop("Enables the planner's use of materialization."),
  			NULL
*** a/src/backend/utils/misc/postgresql.conf.sample
--- b/src/backend/utils/misc/postgresql.conf.sample
***************
*** 270,275 ****
--- 270,276 ----
  
  #enable_bitmapscan = on
  #enable_hashagg = on
+ #enable_hashagg_disk = on
  #enable_hashjoin = on
  #enable_indexscan = on
  #enable_indexonlyscan = on
*** a/src/include/executor/executor.h
--- b/src/include/executor/executor.h
***************
*** 147,152 **** extern TupleHashTable BuildTupleHashTable(int numCols, AttrNumber *keyColIdx,
--- 147,158 ----
  extern TupleHashEntry LookupTupleHashEntry(TupleHashTable hashtable,
  					 TupleTableSlot *slot,
  					 bool *isnew);
+ extern uint32 TupleHashEntryHash(TupleHashTable hashtable,
+ 					 TupleTableSlot *slot);
+ extern TupleHashEntry LookupTupleHashEntryHash(TupleHashTable hashtable,
+ 					 TupleTableSlot *slot,
+ 					 uint32 hashvalue,
+ 					 bool *isnew);
  extern TupleHashEntry FindTupleHashEntry(TupleHashTable hashtable,
  				   TupleTableSlot *slot,
  				   FmgrInfo *eqfunctions,
*** a/src/include/executor/nodeAgg.h
--- b/src/include/executor/nodeAgg.h
***************
*** 22,27 **** extern void ExecEndAgg(AggState *node);
--- 22,28 ----
  extern void ExecReScanAgg(AggState *node);
  
  extern Size hash_agg_entry_size(int numAggs);
+ extern Size hash_group_size(int numAggs, int inputWidth, Size transitionSpace);
  
  extern Datum aggregate_dummy(PG_FUNCTION_ARGS);
  
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 1759,1769 **** typedef struct AggState
--- 1759,1776 ----
  	AggStatePerGroup pergroup;	/* per-Aggref-per-group working state */
  	HeapTuple	grp_firstTuple; /* copy of first tuple of current group */
  	/* these fields are used in AGG_HASHED mode: */
+ 	MemoryContext hashcontext;	/* subcontext to use for hash table */
  	TupleHashTable hashtable;	/* hash table with one entry per group */
  	TupleTableSlot *hashslot;	/* slot for loading hash table */
  	List	   *hash_needed;	/* list of columns needed in hash table */
+ 	bool		hash_init_state; /* in initial state before execution? */
  	bool		table_filled;	/* hash table filled yet? */
+ 	int64		hash_disk;		/* bytes of disk space used */
+ 	uint64		hash_mem_min;	/* memory used by empty hash table */
+ 	uint64		hash_mem_peak;	/* memory used at peak of execution */
+ 	int			hash_num_batches; /* total number of batches created */
  	TupleHashIterator hashiter; /* for iterating through hash table */
+ 	List	   *hash_work;		/* remaining work to be done */
  } AggState;
  
  /* ----------------
*** a/src/include/nodes/plannodes.h
--- b/src/include/nodes/plannodes.h
***************
*** 666,671 **** typedef struct Agg
--- 666,673 ----
  	AttrNumber *grpColIdx;		/* their indexes in the target list */
  	Oid		   *grpOperators;	/* equality operators to compare with */
  	long		numGroups;		/* estimated number of groups in input */
+ 	Size		transitionSpace; /* estimated size of by-ref transition val */
+ 	int			plan_width;		/* input plan width */
  } Agg;
  
  /* ----------------
*** a/src/include/optimizer/cost.h
--- b/src/include/optimizer/cost.h
***************
*** 57,62 **** extern bool enable_bitmapscan;
--- 57,63 ----
  extern bool enable_tidscan;
  extern bool enable_sort;
  extern bool enable_hashagg;
+ extern bool enable_hashagg_disk;
  extern bool enable_nestloop;
  extern bool enable_material;
  extern bool enable_mergejoin;
***************
*** 102,108 **** extern void cost_agg(Path *path, PlannerInfo *root,
  		 AggStrategy aggstrategy, const AggClauseCosts *aggcosts,
  		 int numGroupCols, double numGroups,
  		 Cost input_startup_cost, Cost input_total_cost,
! 		 double input_tuples);
  extern void cost_windowagg(Path *path, PlannerInfo *root,
  			   List *windowFuncs, int numPartCols, int numOrderCols,
  			   Cost input_startup_cost, Cost input_total_cost,
--- 103,109 ----
  		 AggStrategy aggstrategy, const AggClauseCosts *aggcosts,
  		 int numGroupCols, double numGroups,
  		 Cost input_startup_cost, Cost input_total_cost,
! 		 int input_width, double input_tuples);
  extern void cost_windowagg(Path *path, PlannerInfo *root,
  			   List *windowFuncs, int numPartCols, int numOrderCols,
  			   Cost input_startup_cost, Cost input_total_cost,
*** a/src/include/utils/hsearch.h
--- b/src/include/utils/hsearch.h
***************
*** 93,98 **** typedef struct HASHCTL
--- 93,101 ----
  #define HASH_COMPARE	0x400	/* Set user defined comparison function */
  #define HASH_KEYCOPY	0x800	/* Set user defined key-copying function */
  #define HASH_FIXED_SIZE 0x1000	/* Initial size is a hard limit */
+ #define HASH_NOCHILDCXT 0x2000	/* Don't create a child context. Warning:
+ 								 * hash_destroy will delete the memory context
+ 								 * specified by the caller. */
  
  
  /* max_dsize value to indicate expansible directory */
*** a/src/test/regress/expected/rangefuncs.out
--- b/src/test/regress/expected/rangefuncs.out
***************
*** 3,8 **** SELECT name, setting FROM pg_settings WHERE name LIKE 'enable%';
--- 3,9 ----
  ----------------------+---------
   enable_bitmapscan    | on
   enable_hashagg       | on
+  enable_hashagg_disk  | on
   enable_hashjoin      | on
   enable_indexonlyscan | on
   enable_indexscan     | on
***************
*** 12,18 **** SELECT name, setting FROM pg_settings WHERE name LIKE 'enable%';
   enable_seqscan       | on
   enable_sort          | on
   enable_tidscan       | on
! (11 rows)
  
  CREATE TABLE foo2(fooid int, f2 int);
  INSERT INTO foo2 VALUES(1, 11);
--- 13,19 ----
   enable_seqscan       | on
   enable_sort          | on
   enable_tidscan       | on
! (12 rows)
  
  CREATE TABLE foo2(fooid int, f2 int);
  INSERT INTO foo2 VALUES(1, 11);