Proposed Patch to Improve Performance of Multi-Batch Hash Join for Skewed Data Sets

Index: src/backend/executor/nodeHash.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHash.c,v
retrieving revision 1.116
diff -c -r1.116 nodeHash.c
*** src/backend/executor/nodeHash.c	1 Jan 2008 19:45:49 -0000	1.116
--- src/backend/executor/nodeHash.c	5 Nov 2008 22:26:53 -0000
***************
*** 54,59 ****
--- 54,86 ----
  }
  
  /* ----------------------------------------------------------------
+ *		isAMostCommonValue
+ *
+ *		is the value one of the most common key values?
+ *  ----------------------------------------------------------------
+ */
+ bool isAMostCommonValue(HashJoinTable hashtable, uint32 hashvalue, int *partitionNumber)
+ {
+ 	int bucket = hashvalue & (hashtable->nMostCommonTuplePartitionHashBuckets - 1);
+ 
+ 	while (hashtable->mostCommonTuplePartition[bucket].hashvalue != 0
+ 		&& hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 	{
+ 		bucket = (bucket + 1) & (hashtable->nMostCommonTuplePartitionHashBuckets - 1);
+ 	}
+ 
+ 	if (hashtable->mostCommonTuplePartition[bucket].hashvalue == hashvalue)
+ 	{
+ 		*partitionNumber = bucket;
+ 		return true;
+ 	}
+ 
+ 	//must have run into an empty slot which means this is not an MCV
+ 	*partitionNumber = MCV_INVALID_PARTITION;
+ 	return false;
+ }
+ 
+ /* ----------------------------------------------------------------
   *		MultiExecHash
   *
   *		build hash table for hashjoin, doing partitioning if more
***************
*** 69,74 ****
--- 96,103 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int partitionNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,106 ****
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
  	}
  
--- 128,163 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			partitionNumber = MCV_INVALID_PARTITION;
! 
! 			if (hashtable->usingMostCommonValues && isAMostCommonValue(hashtable, hashvalue, &partitionNumber))
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple = (HashJoinTuple) palloc(hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple, mintuple->t_len);
! 
! 				hashTuple->next = hashtable->mostCommonTuplePartition[partitionNumber].tuples;
! 				hashtable->mostCommonTuplePartition[partitionNumber].tuples = hashTuple;
! 				
! 				hashtable->spaceUsed += hashTupleSize;
! 				
! 				if (hashtable->spaceUsed > hashtable->spaceAllowed) {
! 					ExecHashIncreaseNumBatches(hashtable);
! 				}
! 				
! 				hashtable->mostCommonTuplesStored++;
! 			}
! 
! 			if (partitionNumber == MCV_INVALID_PARTITION)
! 			{
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 				hashtable->totalTuples += 1;
! 			}
  		}
  	}
  
***************
*** 798,803 ****
--- 855,922 ----
  }
  
  /*
+  * ExecScanHashMostCommonTuples
+  *		scan a hash bucket for matches to the current outer tuple
+  *
+  * The current outer tuple must be stored in econtext->ecxt_outertuple.
+  */
+ HashJoinTuple
+ ExecScanHashMostCommonTuples(HashJoinState *hjstate,
+ 				   ExprContext *econtext)
+ {
+ 	List	   *hjclauses = hjstate->hashclauses;
+ 	HashJoinTable hashtable = hjstate->hj_HashTable;
+ 	HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+ 	uint32		hashvalue = hjstate->hj_CurHashValue;
+ 
+ 	/*
+ 	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
+ 	 * the last tuple returned from the current bucket.
+ 	 */
+ 	if (hashTuple == NULL)
+ 	{
+ 		//painstakingly make sure this is a valid partition index
+ 		Assert(hjstate->hj_OuterTupleMostCommonValuePartition > MCV_INVALID_PARTITION);
+ 		Assert(hjstate->hj_OuterTupleMostCommonValuePartition < hashtable->nMostCommonTuplePartitionHashBuckets);
+ 		Assert(hashtable->mostCommonTuplePartition[hjstate->hj_OuterTupleMostCommonValuePartition].hashvalue != 0);
+ 
+ 		hashTuple = hashtable->mostCommonTuplePartition[hjstate->hj_OuterTupleMostCommonValuePartition].tuples;
+ 	}
+ 	else
+ 		hashTuple = hashTuple->next;
+ 
+ 	while (hashTuple != NULL)
+ 	{
+ 		if (hashTuple->hashvalue == hashvalue)
+ 		{
+ 			TupleTableSlot *inntuple;
+ 
+ 			/* insert hashtable's tuple into exec slot so ExecQual sees it */
+ 			inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ 											 hjstate->hj_HashTupleSlot,
+ 											 false);	/* do not pfree */
+ 			econtext->ecxt_innertuple = inntuple;
+ 
+ 			/* reset temp memory each time to avoid leaks from qual expr */
+ 			ResetExprContext(econtext);
+ 
+ 			if (ExecQual(hjclauses, econtext, false))
+ 			{
+ 				hjstate->hj_CurTuple = hashTuple;
+ 				return hashTuple;
+ 			}
+ 		}
+ 
+ 		hashTuple = hashTuple->next;
+ 	}
+ 
+ 	/*
+ 	 * no match
+ 	 */
+ 	return NULL;
+ }
+ 
+ /*
   * ExecScanHashBucket
   *		scan a hash bucket for matches to the current outer tuple
   *
Index: src/backend/executor/nodeHashjoin.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHashjoin.c,v
retrieving revision 1.96
diff -c -r1.96 nodeHashjoin.c
*** src/backend/executor/nodeHashjoin.c	23 Oct 2008 14:34:34 -0000	1.96
--- src/backend/executor/nodeHashjoin.c	5 Nov 2008 22:56:59 -0000
***************
*** 20,25 ****
--- 20,30 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "optimizer/cost.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 ****
--- 39,146 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+ *          getMostCommonValues
+ *
+ *          
+ */
+ void getMostCommonValues(EState *estate, HashJoinState *hjstate)
+ {
+ 	HeapTupleData *statsTuple;
+ 	FuncExprState *clause;
+ 	ExprState *argstate;
+ 	Var *variable;
+ 
+ 	Datum	   *values;
+ 	int			nvalues;
+ 	float4	   *numbers;
+ 	int			nnumbers;
+ 
+ 	Oid relid;
+ 	AttrNumber relattnum;
+ 	Oid atttype;
+ 	int32 atttypmod;
+ 
+ 	int i;
+ 
+ 	//is it a join on more than one key?
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; //histojoin is not defined for more than one join key so run away
+ 
+ 	//make sure the outer node is a seq scan on a base relation otherwise we cant get MCVs at the moment and should not bother trying
+ 	if (outerPlanState(hjstate)->type != T_SeqScanState)
+ 		return;
+ 	
+ 	//grab the relation object id of the outer relation
+ 	relid = getrelid(((SeqScan *) ((SeqScanState *) outerPlanState(hjstate))->ps.plan)->scanrelid, estate->es_range_table);
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 	variable = (Var *) argstate->expr;
+ 
+ 	//grab the necessary properties of the join variable
+ 	relattnum = variable->varattno;
+ 	atttype = variable->vartype;
+ 	atttypmod = variable->vartypmod;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT,
+ 		ObjectIdGetDatum(relid),
+ 		Int16GetDatum(relattnum),
+ 		0, 0);
+ 
+ 	if (HeapTupleIsValid(statsTuple))
+ 	{
+ 		if (get_attstatsslot(statsTuple,
+ 			atttype, atttypmod,
+ 			STATISTIC_KIND_MCV, InvalidOid,
+ 			&values, &nvalues,
+ 			&numbers, &nnumbers))
+ 		{
+ 			HashJoinTable hashtable;
+ 			FmgrInfo   *hashfunctions;
+ 			//MCV Partitions is an open addressing hashtable with a power of 2 size greater than the number of MCV values
+ 			int nbuckets = 2;
+ 			uint32 collisionsWhileHashing = 0;
+ 			while (nbuckets <= nvalues)
+ 			{
+ 				nbuckets <<= 1;
+ 			}
+ 			//use two more bit just to help avoid collisions
+ 			nbuckets <<= 2;
+ 
+ 			hashtable = hjstate->hj_HashTable;
+ 			hashtable->usingMostCommonValues = true;
+ 			hashtable->nMostCommonTuplePartitionHashBuckets = nbuckets;
+ 			hashtable->mostCommonTuplePartition = palloc0(nbuckets * sizeof(HashJoinMostCommonValueTuplePartition));
+ 			hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 			//create the partitions
+ 			for (i = 0; i < nvalues; i++)
+ 			{
+ 				uint32 hashvalue = DatumGetUInt32(FunctionCall1(&hashfunctions[0], values[i]));
+ 				int bucket = hashvalue & (nbuckets - 1);
+ 
+ 				while (hashtable->mostCommonTuplePartition[bucket].hashvalue != 0
+ 					&& hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 				{
+ 					bucket = (bucket + 1) & (nbuckets - 1);
+ 					collisionsWhileHashing++;
+ 				}
+ 
+ 				//leave partition alone if it has the same hashvalue as current MCV.  we only want one partition per hashvalue
+ 				if (hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 				{
+ 					hashtable->mostCommonTuplePartition[bucket].tuples = NULL;
+ 					hashtable->mostCommonTuplePartition[bucket].hashvalue = hashvalue;
+ 					hashtable->nMostCommonTuplePartitions++;
+ 				}
+ 			}
+ 
+ 			free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 		}
+ 
+ 		ReleaseSysCache(statsTuple);
+ 	}
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 146,151 ****
--- 253,267 ----
  		hashtable = ExecHashTableCreate((Hash *) hashNode->ps.plan,
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
+ 		
+ 		hashtable->usingMostCommonValues = false;
+ 		hashtable->nMostCommonTuplePartitions = 0;
+ 		hashtable->nMostCommonTuplePartitionHashBuckets = 0;
+ 		hashtable->mostCommonTuplesStored = 0;
+ 		hashtable->mostCommonTuplePartition = NULL;
+ 
+ 		if (enable_hashjoin_usestatmcvs)
+ 			getMostCommonValues(estate, node);
  
  		/*
  		 * execute the Hash node, to build the hash table
***************
*** 157,163 ****
  		 * If the inner relation is completely empty, and we're not doing an
  		 * outer join, we can quit without scanning the outer relation.
  		 */
! 		if (hashtable->totalTuples == 0 && !HASHJOIN_IS_OUTER(node))
  			return NULL;
  
  		/*
--- 273,279 ----
  		 * If the inner relation is completely empty, and we're not doing an
  		 * outer join, we can quit without scanning the outer relation.
  		 */
! 		if (hashtable->totalTuples == 0 && hashtable->mostCommonTuplesStored == 0 && !HASHJOIN_IS_OUTER(node))
  			return NULL;
  
  		/*
***************
*** 205,227 ****
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
! 
! 			/*
! 			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
! 			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
! 				 * Need to postpone this outer tuple to a later batch. Save it
! 				 * in the corresponding outer-batch file.
  				 */
! 				Assert(batchno > hashtable->curbatch);
! 				ExecHashJoinSaveTuple(ExecFetchSlotMinimalTuple(outerTupleSlot),
! 									  hashvalue,
! 									  &hashtable->outerBatchFile[batchno]);
! 				node->hj_NeedNewOuter = true;
! 				continue;		/* loop around for a new outer tuple */
  			}
  		}
  
--- 321,349 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
! 			
! 			node->hj_OuterTupleMostCommonValuePartition = MCV_INVALID_PARTITION;
! 			
! 			
! 			if (!(hashtable->usingMostCommonValues && isAMostCommonValue(hashtable, hashvalue, &node->hj_OuterTupleMostCommonValuePartition)))
  			{
  				/*
! 				 * Now we've got an outer tuple and the corresponding hash bucket,
! 				 * but this tuple may not belong to the current batch.
  				 */
! 				if (batchno != hashtable->curbatch)
! 				{
! 					/*
! 					 * Need to postpone this outer tuple to a later batch. Save it
! 					 * in the corresponding outer-batch file.
! 					 */
! 					Assert(batchno > hashtable->curbatch);
! 					ExecHashJoinSaveTuple(ExecFetchSlotMinimalTuple(outerTupleSlot),
! 										  hashvalue,
! 										  &hashtable->outerBatchFile[batchno]);
! 					node->hj_NeedNewOuter = true;
! 					continue;		/* loop around for a new outer tuple */
! 				}
  			}
  		}
  
***************
*** 230,236 ****
  		 */
  		for (;;)
  		{
! 			curtuple = ExecScanHashBucket(node, econtext);
  			if (curtuple == NULL)
  				break;			/* out of matches */
  
--- 352,365 ----
  		 */
  		for (;;)
  		{
! 			if (node->hj_OuterTupleMostCommonValuePartition != MCV_INVALID_PARTITION)
! 			{
! 				curtuple = ExecScanHashMostCommonTuples(node, econtext);
! 			}
! 			else
! 			{
! 				curtuple = ExecScanHashBucket(node, econtext);
! 			}
  			if (curtuple == NULL)
  				break;			/* out of matches */
  
Index: src/backend/optimizer/path/costsize.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/optimizer/path/costsize.c,v
retrieving revision 1.200
diff -c -r1.200 costsize.c
*** src/backend/optimizer/path/costsize.c	21 Oct 2008 20:42:52 -0000	1.200
--- src/backend/optimizer/path/costsize.c	5 Nov 2008 22:57:01 -0000
***************
*** 109,114 ****
--- 109,116 ----
  bool		enable_mergejoin = true;
  bool		enable_hashjoin = true;
  
+ bool		enable_hashjoin_usestatmcvs = true;
+ 
  typedef struct
  {
  	PlannerInfo *root;
Index: src/backend/utils/misc/guc.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/utils/misc/guc.c,v
retrieving revision 1.475
diff -c -r1.475 guc.c
*** src/backend/utils/misc/guc.c	6 Oct 2008 13:05:36 -0000	1.475
--- src/backend/utils/misc/guc.c	9 Oct 2008 19:56:17 -0000
***************
*** 625,630 ****
--- 625,638 ----
  		true, NULL, NULL
  	},
  	{
+ 		{"enable_hashjoin_usestatmcvs", PGC_USERSET, QUERY_TUNING_METHOD,
+ 			gettext_noop("Enables the hash join's use of the MCVs stored in pg_statistic."),
+ 			NULL
+ 		},
+ 		&enable_hashjoin_usestatmcvs,
+ 		true, NULL, NULL
+ 	},
+ 	{
  		{"constraint_exclusion", PGC_USERSET, QUERY_TUNING_OTHER,
  			gettext_noop("Enables the planner to use constraints to optimize queries."),
  			gettext_noop("Child table scans will be skipped if their "
Index: src/include/executor/hashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/hashjoin.h,v
retrieving revision 1.48
diff -c -r1.48 hashjoin.h
*** src/include/executor/hashjoin.h	1 Jan 2008 19:45:57 -0000	1.48
--- src/include/executor/hashjoin.h	17 Oct 2008 23:48:46 -0000
***************
*** 72,77 ****
--- 72,84 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ typedef struct HashJoinMostCommonValueTuplePartition
+ {
+ 	uint32 hashvalue;
+ 	HashJoinTuple tuples;
+ } HashJoinMostCommonValueTuplePartition;
+ 
+ #define MCV_INVALID_PARTITION -1
  
  typedef struct HashJoinTableData
  {
***************
*** 116,121 ****
--- 123,134 ----
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	bool usingMostCommonValues;
+ 	HashJoinMostCommonValueTuplePartition *mostCommonTuplePartition;
+ 	int nMostCommonTuplePartitionHashBuckets;
+ 	int nMostCommonTuplePartitions;
+ 	uint32 mostCommonTuplesStored;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
Index: src/include/executor/nodeHash.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHash.h,v
retrieving revision 1.45
diff -c -r1.45 nodeHash.h
*** src/include/executor/nodeHash.h	1 Jan 2008 19:45:57 -0000	1.45
--- src/include/executor/nodeHash.h	30 Sep 2008 20:31:35 -0000
***************
*** 45,48 ****
--- 45,51 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern HashJoinTuple ExecScanHashMostCommonTuples(HashJoinState *hjstate, ExprContext *econtext);
+ extern bool isAMostCommonValue(HashJoinTable hashtable, uint32 hashvalue, int *partitionNumber);
+ 
  #endif   /* NODEHASH_H */
Index: src/include/executor/nodeHashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHashjoin.h,v
retrieving revision 1.37
diff -c -r1.37 nodeHashjoin.h
*** src/include/executor/nodeHashjoin.h	1 Jan 2008 19:45:57 -0000	1.37
--- src/include/executor/nodeHashjoin.h	30 Sep 2008 20:32:05 -0000
***************
*** 26,29 ****
--- 26,31 ----
  extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
  					  BufFile **fileptr);
  
+ extern void getMostCommonValues(EState *estate, HashJoinState *hjstate);
+ 
  #endif   /* NODEHASHJOIN_H */
Index: src/include/nodes/execnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/execnodes.h,v
retrieving revision 1.194
diff -c -r1.194 execnodes.h
*** src/include/nodes/execnodes.h	31 Oct 2008 19:37:56 -0000	1.194
--- src/include/nodes/execnodes.h	5 Nov 2008 22:57:08 -0000
***************
*** 1386,1391 ****
--- 1386,1392 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int		hj_OuterTupleMostCommonValuePartition;
  } HashJoinState;
  
  
Index: src/include/optimizer/cost.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/optimizer/cost.h,v
retrieving revision 1.93
diff -c -r1.93 cost.h
*** src/include/optimizer/cost.h	4 Oct 2008 21:56:55 -0000	1.93
--- src/include/optimizer/cost.h	7 Oct 2008 18:31:42 -0000
***************
*** 52,57 ****
--- 52,58 ----
  extern bool enable_nestloop;
  extern bool enable_mergejoin;
  extern bool enable_hashjoin;
+ extern bool enable_hashjoin_usestatmcvs;
  extern bool constraint_exclusion;
  
  extern double clamp_row_est(double nrows);

Index: src/backend/executor/nodeHash.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHash.c,v
retrieving revision 1.116
diff -c -r1.116 nodeHash.c
*** src/backend/executor/nodeHash.c	1 Jan 2008 19:45:49 -0000	1.116
--- src/backend/executor/nodeHash.c	24 Nov 2008 12:32:13 -0000
***************
*** 54,59 ****
--- 54,165 ----
  }
  
  /* ----------------------------------------------------------------
+ *		isAMostCommonValue
+ *
+ *		is the value one of the most common key values?
+ *  ----------------------------------------------------------------
+ */
+ bool isAMostCommonValue(HashJoinTable hashtable, uint32 hashvalue, int *partitionNumber)
+ {
+ 	int bucket = hashvalue & (hashtable->nMostCommonTuplePartitionHashBuckets - 1);
+ 
+ 	while (hashtable->mostCommonTuplePartition[bucket].hashvalue != 0
+ 		&& hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 	{
+ 		bucket = (bucket + 1) & (hashtable->nMostCommonTuplePartitionHashBuckets - 1);
+ 	}
+ 
+ 	if (!hashtable->mostCommonTuplePartition[bucket].frozen && hashtable->mostCommonTuplePartition[bucket].hashvalue == hashvalue)
+ 	{
+ 		*partitionNumber = bucket;
+ 		return true;
+ 	}
+ 
+ 	/* must have run into an empty slot which means this is not an MCV*/
+ 	*partitionNumber = MCV_INVALID_PARTITION;
+ 	return false;
+ }
+ 
+ /*
+ *	freezeNextMCVPartiton
+ *
+ *	flush the tuples of the next MCV partition by pushing them into the main hashtable
+ */
+ bool freezeNextMCVPartiton(HashJoinTable hashtable) {
+ 	int partitionToFlush = hashtable->nMostCommonTuplePartitions - 1 - hashtable->nMostCommonTuplePartitionsFlushed;
+ 	if (partitionToFlush < 0)
+ 		return false;
+ 	else
+ 	{
+ 
+ 		int		bucketno;
+ 		int		batchno;
+ 		uint32		hashvalue;
+ 		HashJoinTuple hashTuple;
+ 		HashJoinTuple nextHashTuple;
+ 		HashJoinMostCommonValueTuplePartition *partition;
+ 		MinimalTuple mintuple;
+ 
+ 		partition = hashtable->flushOrderedMostCommonTuplePartition[partitionToFlush];
+ 		hashvalue = partition->hashvalue;
+ 
+ 		Assert(hashvalue != 0);
+ 
+ 		hashTuple = partition->tuples;
+ 
+ 		ExecHashGetBucketAndBatch(hashtable, hashvalue,
+ 								  &bucketno, &batchno);
+ 
+ 		while (hashTuple != NULL)
+ 		{
+ 			/* decide whether to put the tuples in the hash table or a temp file */
+ 			if (batchno == hashtable->curbatch)
+ 			{
+ 				/* put the tuples in hash table */
+ 				nextHashTuple = hashTuple->next;
+ 
+ 				hashTuple->next = hashtable->buckets[bucketno];
+ 				hashtable->buckets[bucketno] = hashTuple;
+ 				
+ 				hashTuple = nextHashTuple;
+ 				hashtable->totalTuples++;
+ 				hashtable->mostCommonTuplesStored--;
+ 				
+ 				if (hashtable->spaceUsed > hashtable->spaceAllowed)
+ 				{
+ 					ExecHashIncreaseNumBatches(hashtable);
+ 					/* likely changed due to increase in batches */
+ 					ExecHashGetBucketAndBatch(hashtable, hashvalue,
+ 						&bucketno, &batchno);
+ 				}
+ 			}
+ 			else
+ 			{
+ 				/* put the tuples into a temp file for later batches */
+ 				Assert(batchno > hashtable->curbatch);
+ 				mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 				ExecHashJoinSaveTuple(mintuple,
+ 									  hashvalue,
+ 									  &hashtable->innerBatchFile[batchno]);
+ 				hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 				nextHashTuple = hashTuple->next;
+ 				pfree(hashTuple);
+ 				hashTuple = nextHashTuple;
+ 				hashtable->inTupIOs++;
+ 				hashtable->totalTuples++;
+ 				hashtable->mostCommonTuplesStored--;
+ 			}
+ 		}
+ 
+ 		partition->frozen = true;
+ 		partition->tuples = NULL;
+ 		hashtable->nMostCommonTuplePartitionsFlushed++;
+ 
+ 		return true;
+ 	}
+ }
+ 
+ /* ----------------------------------------------------------------
   *		MultiExecHash
   *
   *		build hash table for hashjoin, doing partitioning if more
***************
*** 69,74 ****
--- 175,182 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int partitionNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,106 ****
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
  	}
  
--- 207,240 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			partitionNumber = MCV_INVALID_PARTITION;
! 
! 			if (hashtable->usingMostCommonValues && isAMostCommonValue(hashtable, hashvalue, &partitionNumber))
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple = (HashJoinTuple) palloc(hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple, mintuple->t_len);
! 
! 				hashTuple->next = hashtable->mostCommonTuplePartition[partitionNumber].tuples;
! 				hashtable->mostCommonTuplePartition[partitionNumber].tuples = hashTuple;
! 				
! 				hashtable->spaceUsed += hashTupleSize;
! 				
! 				hashtable->mostCommonTuplesStored++;
! 				
! 				while (hashtable->spaceUsed > hashtable->spaceAllowed && freezeNextMCVPartiton(hashtable)) {}
! 			}
! 
! 			if (partitionNumber == MCV_INVALID_PARTITION)
! 			{
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 				hashtable->totalTuples += 1;
! 			}
  		}
  	}
  
***************
*** 461,466 ****
--- 595,606 ----
  			BufFileClose(hashtable->outerBatchFile[i]);
  	}
  
+ 	if (hashtable->usingMostCommonValues)
+ 	{
+ 		pfree(hashtable->mostCommonTuplePartition);
+ 		pfree(hashtable->flushOrderedMostCommonTuplePartition);
+ 	}
+ 
  	/* Release working memory (batchCxt is a child, so it goes away too) */
  	MemoryContextDelete(hashtable->hashCxt);
  
***************
*** 798,803 ****
--- 938,1005 ----
  }
  
  /*
+  * ExecScanHashMostCommonTuples
+  *		scan a hash bucket for matches to the current outer tuple
+  *
+  * The current outer tuple must be stored in econtext->ecxt_outertuple.
+  */
+ HashJoinTuple
+ ExecScanHashMostCommonTuples(HashJoinState *hjstate,
+ 				   ExprContext *econtext)
+ {
+ 	List	   *hjclauses = hjstate->hashclauses;
+ 	HashJoinTable hashtable = hjstate->hj_HashTable;
+ 	HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+ 	uint32		hashvalue = hjstate->hj_CurHashValue;
+ 
+ 	/*
+ 	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
+ 	 * the last tuple returned from the current bucket.
+ 	 */
+ 	if (hashTuple == NULL)
+ 	{
+ 		//painstakingly make sure this is a valid partition index
+ 		Assert(hjstate->hj_OuterTupleMostCommonValuePartition > MCV_INVALID_PARTITION);
+ 		Assert(hjstate->hj_OuterTupleMostCommonValuePartition < hashtable->nMostCommonTuplePartitionHashBuckets);
+ 		Assert(hashtable->mostCommonTuplePartition[hjstate->hj_OuterTupleMostCommonValuePartition].hashvalue != 0);
+ 
+ 		hashTuple = hashtable->mostCommonTuplePartition[hjstate->hj_OuterTupleMostCommonValuePartition].tuples;
+ 	}
+ 	else
+ 		hashTuple = hashTuple->next;
+ 
+ 	while (hashTuple != NULL)
+ 	{
+ 		if (hashTuple->hashvalue == hashvalue)
+ 		{
+ 			TupleTableSlot *inntuple;
+ 
+ 			/* insert hashtable's tuple into exec slot so ExecQual sees it */
+ 			inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ 											 hjstate->hj_HashTupleSlot,
+ 											 false);	/* do not pfree */
+ 			econtext->ecxt_innertuple = inntuple;
+ 
+ 			/* reset temp memory each time to avoid leaks from qual expr */
+ 			ResetExprContext(econtext);
+ 
+ 			if (ExecQual(hjclauses, econtext, false))
+ 			{
+ 				hjstate->hj_CurTuple = hashTuple;
+ 				return hashTuple;
+ 			}
+ 		}
+ 
+ 		hashTuple = hashTuple->next;
+ 	}
+ 
+ 	/*
+ 	 * no match
+ 	 */
+ 	return NULL;
+ }
+ 
+ /*
   * ExecScanHashBucket
   *		scan a hash bucket for matches to the current outer tuple
   *
Index: src/backend/executor/nodeHashjoin.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHashjoin.c,v
retrieving revision 1.96
diff -c -r1.96 nodeHashjoin.c
*** src/backend/executor/nodeHashjoin.c	23 Oct 2008 14:34:34 -0000	1.96
--- src/backend/executor/nodeHashjoin.c	24 Nov 2008 12:35:29 -0000
***************
*** 20,25 ****
--- 20,30 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "optimizer/cost.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 ****
--- 39,149 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+ *          getMostCommonValues
+ *
+ *          
+ */
+ void getMostCommonValues(EState *estate, HashJoinState *hjstate)
+ {
+ 	HeapTupleData *statsTuple;
+ 	FuncExprState *clause;
+ 	ExprState *argstate;
+ 	Var *variable;
+ 
+ 	Datum	   *values;
+ 	int			nvalues;
+ 	float4	   *numbers;
+ 	int			nnumbers;
+ 
+ 	Oid relid;
+ 	AttrNumber relattnum;
+ 	Oid atttype;
+ 	int32 atttypmod;
+ 
+ 	int i;
+ 
+ 	//is it a join on more than one key?
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; //histojoin is not defined for more than one join key so run away
+ 
+ 	//make sure the outer node is a seq scan on a base relation otherwise we cant get MCVs at the moment and should not bother trying
+ 	if (outerPlanState(hjstate)->type != T_SeqScanState)
+ 		return;
+ 	
+ 	//grab the relation object id of the outer relation
+ 	relid = getrelid(((SeqScan *) ((SeqScanState *) outerPlanState(hjstate))->ps.plan)->scanrelid, estate->es_range_table);
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 	variable = (Var *) argstate->expr;
+ 
+ 	//grab the necessary properties of the join variable
+ 	relattnum = variable->varattno;
+ 	atttype = variable->vartype;
+ 	atttypmod = variable->vartypmod;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT,
+ 		ObjectIdGetDatum(relid),
+ 		Int16GetDatum(relattnum),
+ 		0, 0);
+ 
+ 	if (HeapTupleIsValid(statsTuple))
+ 	{
+ 		if (get_attstatsslot(statsTuple,
+ 			atttype, atttypmod,
+ 			STATISTIC_KIND_MCV, InvalidOid,
+ 			&values, &nvalues,
+ 			&numbers, &nnumbers))
+ 		{
+ 			HashJoinTable hashtable;
+ 			FmgrInfo   *hashfunctions;
+ 			//MCV Partitions is an open addressing hashtable with a power of 2 size greater than the number of MCV values
+ 			int nbuckets = 2;
+ 			uint32 collisionsWhileHashing = 0;
+ 			while (nbuckets <= nvalues)
+ 			{
+ 				nbuckets <<= 1;
+ 			}
+ 			//use two more bit just to help avoid collisions
+ 			nbuckets <<= 2;
+ 
+ 			hashtable = hjstate->hj_HashTable;
+ 			hashtable->usingMostCommonValues = true;
+ 			hashtable->nMostCommonTuplePartitionHashBuckets = nbuckets;
+ 			hashtable->mostCommonTuplePartition = palloc0(nbuckets * sizeof(HashJoinMostCommonValueTuplePartition));
+ 			hashtable->flushOrderedMostCommonTuplePartition = palloc0(nvalues * sizeof(HashJoinMostCommonValueTuplePartition*));
+ 			hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 			//create the partitions
+ 			for (i = 0; i < nvalues; i++)
+ 			{
+ 				uint32 hashvalue = DatumGetUInt32(FunctionCall1(&hashfunctions[0], values[i]));
+ 				int bucket = hashvalue & (nbuckets - 1);
+ 
+ 				while (hashtable->mostCommonTuplePartition[bucket].hashvalue != 0
+ 					&& hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 				{
+ 					bucket = (bucket + 1) & (nbuckets - 1);
+ 					collisionsWhileHashing++;
+ 				}
+ 
+ 				//leave partition alone if it has the same hashvalue as current MCV.  we only want one partition per hashvalue
+ 				if (hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 				{
+ 					hashtable->mostCommonTuplePartition[bucket].tuples = NULL;
+ 					hashtable->mostCommonTuplePartition[bucket].hashvalue = hashvalue;
+ 					hashtable->mostCommonTuplePartition[bucket].frozen = false;
+ +					hashtable->flushOrderedMostCommonTuplePartition[hashtable->nMostCommonTuplePartitions] = &hashtable->mostCommonTuplePartition[bucket];
+ 					hashtable->nMostCommonTuplePartitions++;
+ 				}
+ 			}
+ 
+ 			free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 		}
+ 
+ 		ReleaseSysCache(statsTuple);
+ 	}
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 146,151 ****
--- 256,271 ----
  		hashtable = ExecHashTableCreate((Hash *) hashNode->ps.plan,
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
+ 		
+ 		hashtable->usingMostCommonValues = false;
+ 		hashtable->nMostCommonTuplePartitions = 0;
+ 		hashtable->nMostCommonTuplePartitionHashBuckets = 0;
+ 		hashtable->mostCommonTuplesStored = 0;
+ 		hashtable->mostCommonTuplePartition = NULL;
+ 		hashtable->nMostCommonTuplePartitionsFlushed = 0;
+ 
+ 		if (hashtable->nbatch > 1 && enable_hashjoin_usestatmcvs)
+ 			getMostCommonValues(estate, node);
  
  		/*
  		 * execute the Hash node, to build the hash table
***************
*** 157,163 ****
  		 * If the inner relation is completely empty, and we're not doing an
  		 * outer join, we can quit without scanning the outer relation.
  		 */
! 		if (hashtable->totalTuples == 0 && !HASHJOIN_IS_OUTER(node))
  			return NULL;
  
  		/*
--- 277,283 ----
  		 * If the inner relation is completely empty, and we're not doing an
  		 * outer join, we can quit without scanning the outer relation.
  		 */
! 		if (hashtable->totalTuples == 0 && hashtable->mostCommonTuplesStored == 0 && !HASHJOIN_IS_OUTER(node))
  			return NULL;
  
  		/*
***************
*** 205,227 ****
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
! 
! 			/*
! 			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
! 			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
! 				 * Need to postpone this outer tuple to a later batch. Save it
! 				 * in the corresponding outer-batch file.
  				 */
! 				Assert(batchno > hashtable->curbatch);
! 				ExecHashJoinSaveTuple(ExecFetchSlotMinimalTuple(outerTupleSlot),
! 									  hashvalue,
! 									  &hashtable->outerBatchFile[batchno]);
! 				node->hj_NeedNewOuter = true;
! 				continue;		/* loop around for a new outer tuple */
  			}
  		}
  
--- 325,353 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
! 			
! 			node->hj_OuterTupleMostCommonValuePartition = MCV_INVALID_PARTITION;
! 			
! 			
! 			if (!(hashtable->usingMostCommonValues && isAMostCommonValue(hashtable, hashvalue, &node->hj_OuterTupleMostCommonValuePartition)))
  			{
  				/*
! 				 * Now we've got an outer tuple and the corresponding hash bucket,
! 				 * but this tuple may not belong to the current batch.
  				 */
! 				if (batchno != hashtable->curbatch)
! 				{
! 					/*
! 					 * Need to postpone this outer tuple to a later batch. Save it
! 					 * in the corresponding outer-batch file.
! 					 */
! 					Assert(batchno > hashtable->curbatch);
! 					ExecHashJoinSaveTuple(ExecFetchSlotMinimalTuple(outerTupleSlot),
! 										  hashvalue,
! 										  &hashtable->outerBatchFile[batchno]);
! 					node->hj_NeedNewOuter = true;
! 					continue;		/* loop around for a new outer tuple */
! 				}
  			}
  		}
  
***************
*** 230,236 ****
  		 */
  		for (;;)
  		{
! 			curtuple = ExecScanHashBucket(node, econtext);
  			if (curtuple == NULL)
  				break;			/* out of matches */
  
--- 356,369 ----
  		 */
  		for (;;)
  		{
! 			if (node->hj_OuterTupleMostCommonValuePartition != MCV_INVALID_PARTITION)
! 			{
! 				curtuple = ExecScanHashMostCommonTuples(node, econtext);
! 			}
! 			else
! 			{
! 				curtuple = ExecScanHashBucket(node, econtext);
! 			}
  			if (curtuple == NULL)
  				break;			/* out of matches */
  
Index: src/backend/optimizer/path/costsize.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/optimizer/path/costsize.c,v
retrieving revision 1.201
diff -c -r1.201 costsize.c
*** src/backend/optimizer/path/costsize.c	22 Nov 2008 22:47:05 -0000	1.201
--- src/backend/optimizer/path/costsize.c	24 Nov 2008 12:15:00 -0000
***************
*** 109,114 ****
--- 109,116 ----
  bool		enable_mergejoin = true;
  bool		enable_hashjoin = true;
  
+ bool		enable_hashjoin_usestatmcvs = true;
+ 
  typedef struct
  {
  	PlannerInfo *root;
Index: src/backend/utils/misc/guc.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/utils/misc/guc.c,v
retrieving revision 1.481
diff -c -r1.481 guc.c
*** src/backend/utils/misc/guc.c	21 Nov 2008 20:14:27 -0000	1.481
--- src/backend/utils/misc/guc.c	24 Nov 2008 12:15:05 -0000
***************
*** 636,641 ****
--- 636,649 ----
  		true, NULL, NULL
  	},
  	{
+ 		{"enable_hashjoin_usestatmcvs", PGC_USERSET, QUERY_TUNING_METHOD,
+ 			gettext_noop("Enables the hash join's use of the MCVs stored in pg_statistic."),
+ 			NULL
+ 		},
+ 		&enable_hashjoin_usestatmcvs,
+ 		true, NULL, NULL
+ 	},
+ 	{
  		{"constraint_exclusion", PGC_USERSET, QUERY_TUNING_OTHER,
  			gettext_noop("Enables the planner to use constraints to optimize queries."),
  			gettext_noop("Child table scans will be skipped if their "
Index: src/include/executor/hashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/hashjoin.h,v
retrieving revision 1.48
diff -c -r1.48 hashjoin.h
*** src/include/executor/hashjoin.h	1 Jan 2008 19:45:57 -0000	1.48
--- src/include/executor/hashjoin.h	24 Nov 2008 12:40:18 -0000
***************
*** 72,77 ****
--- 72,85 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ typedef struct HashJoinMostCommonValueTuplePartition
+ {
+ 	uint32 hashvalue;
+ 	bool frozen;
+ 	HashJoinTuple tuples;
+ } HashJoinMostCommonValueTuplePartition;
+ 
+ #define MCV_INVALID_PARTITION -1
  
  typedef struct HashJoinTableData
  {
***************
*** 116,121 ****
--- 124,137 ----
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	bool usingMostCommonValues;
+ 	HashJoinMostCommonValueTuplePartition *mostCommonTuplePartition;
+ 	HashJoinMostCommonValueTuplePartition **flushOrderedMostCommonTuplePartition;
+ 	int nMostCommonTuplePartitionHashBuckets;
+ 	int nMostCommonTuplePartitions;
+ 	int nMostCommonTuplePartitionsFlushed;
+ 	uint32 mostCommonTuplesStored;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
Index: src/include/executor/nodeHash.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHash.h,v
retrieving revision 1.45
diff -c -r1.45 nodeHash.h
*** src/include/executor/nodeHash.h	1 Jan 2008 19:45:57 -0000	1.45
--- src/include/executor/nodeHash.h	30 Sep 2008 20:31:35 -0000
***************
*** 45,48 ****
--- 45,51 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern HashJoinTuple ExecScanHashMostCommonTuples(HashJoinState *hjstate, ExprContext *econtext);
+ extern bool isAMostCommonValue(HashJoinTable hashtable, uint32 hashvalue, int *partitionNumber);
+ 
  #endif   /* NODEHASH_H */
Index: src/include/executor/nodeHashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHashjoin.h,v
retrieving revision 1.37
diff -c -r1.37 nodeHashjoin.h
*** src/include/executor/nodeHashjoin.h	1 Jan 2008 19:45:57 -0000	1.37
--- src/include/executor/nodeHashjoin.h	30 Sep 2008 20:32:05 -0000
***************
*** 26,29 ****
--- 26,31 ----
  extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
  					  BufFile **fileptr);
  
+ extern void getMostCommonValues(EState *estate, HashJoinState *hjstate);
+ 
  #endif   /* NODEHASHJOIN_H */
Index: src/include/nodes/execnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/execnodes.h,v
retrieving revision 1.196
diff -c -r1.196 execnodes.h
*** src/include/nodes/execnodes.h	16 Nov 2008 17:34:28 -0000	1.196
--- src/include/nodes/execnodes.h	17 Nov 2008 20:05:27 -0000
***************
*** 1392,1397 ****
--- 1392,1398 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int		hj_OuterTupleMostCommonValuePartition;
  } HashJoinState;
  
  
Index: src/include/optimizer/cost.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/optimizer/cost.h,v
retrieving revision 1.93
diff -c -r1.93 cost.h
*** src/include/optimizer/cost.h	4 Oct 2008 21:56:55 -0000	1.93
--- src/include/optimizer/cost.h	7 Oct 2008 18:31:42 -0000
***************
*** 52,57 ****
--- 52,58 ----
  extern bool enable_nestloop;
  extern bool enable_mergejoin;
  extern bool enable_hashjoin;
+ extern bool enable_hashjoin_usestatmcvs;
  extern bool constraint_exclusion;
  
  extern double clamp_row_est(double nrows);

�PNG


IHDR
�
�J
N��PLTE


			





   !!!"""###$$$%%%&&&'''((()))***+++,,,///000111222333444555666777888999:::;;;<<<>>>???@@@AAABBBCCCDDDEEEFFFGGGHHHJJJKKKLLLMMMNNNOOOPPPQQQRRRSSSTTTUUUVVVWWWXXXZZZ[[[\\\]]]^^^___```bbbdddeeefffggghhhiiikkklllmmmnnnooopppqqqrrrsssuuuvvvwwwxxxyyyzzz{{{|||}}}~~~������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������"t�IDATx����c������b�V���*/���K,�5�7Z��R������5���RK�-��b�R�WE,U@����!!��%3��9�����9�y���a�|K�}���sJ
`M������`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3��� x�2`
����t���x��lLp��\����f3�����`�@0s ������[?���)�s)��Q�-���K-.�u���~-�Z\������;����"��r�q������"����	;s)���`�@0s �9�f��
3�����;b�l�����u��	�a�
�������0[����lhm�4\l��@�s��!��
���`+@�f3ms�&1ms�&1ms�����A���I�A���I�A��)�����p��#Z1 x��n���="�sm�5S����6	���\��V`�M�`+0�&A��{�
�����A������4�16�V�Ap�y��
Ms/�
C�h��� h��
C������A����0[��c81X3lal��@G����z�fy|�'��I���&d����
� x���|u�����
Cp�V�TH���;w��oy���0���������A���0���7s*�A����{�=-���
C�h���#�^:�.~"���bx���
Ap1��@pT�;Cv|�Si������A�1�)����	#"w��0�k��WL�u���~dV�#������5�E?*�����_�,)��J���
��^����/W�PDi��<X>�%x��Bb����V��^>{��[
���Z�+��[�� d1�)�9H��A�fR4���v�h� E3)�9H�@Bs �
����R��b������.�h,F�3y�,,F�3y�b� E3��-��p)E�M����6)g�Yh�rm�����&I���R�N���6)g�Yh�rm�����j���2>r�&�y\J�^�I�l9)r��R��U��R���T�FG�H�u�����
_��J��vqkt��_[��h�,����$.�|��S��m����4��[F}�$�&��%d���>�IQ�\t�����,)��`|�,'��h��m��`Bp�)��6���m��K�F����m���_R�����!E{��/)���}I����K��vp_R��;�#E���g����P�sb������_��D��;�A�<B�������xR\�YR���h������,��u��x/�
@�<�S���������k���v
���N�K�is7_�0�f�`�1�Z4���*!�` �9�S4R4�`�A�fBJ�O�+/}��`�.�R���`��D �vN���/#���_�C��!+��Q��#�D�^�����N$�_��?����. X��I����A�������R�2|\�2J��.�����Q�����F�q������L�]p��(�����I��O��#Q{���2XFI%E�,�u_���J�����e�D~;zbx��=�j����<���H��Nx�����PA����6IE0Rt/6�I�f�U�J��>;�A�v�M�bs��Fp�m�m�`m�����:E�"X�j�H���(X@��M����>�<�Z�%I�����z�nL?������{��M�~�f�U���^lh�LF����6��`��3���&�lu�>����T'���{8���6��`�S���k*c��8kz�];�6��lY�S��y�+c��-+��ta�_��[�k���W���)Z�K�	m�/��(��\?K�~����7�H�s����u��Fc���_>�|�-�	m��#����gx -��E��v&���z�-��j�hk��]��U�.�'����o���-��j��O�]�t��\�������,�$�3w����R��i@pl�����l�������aV�����W74��86�i���K��?����0���=0M���o�����K�\[�
����d�'>��]
Yc81X3lal��`�c�����4{E�NG���:���YR�/8i��4���2����k'�_���,�������E0�6I����b�`�m��1*X#��6�����1h}��v�M:���}����)��-�]h�;�~��U.�K�`�`�h
��/)Z���,
�
�c�k�T 8��'�]�����a%!����� z�p��'d(��%XD����@�.�6����v��)���^F���E��?x+�L���w���&����]���=�'	��QgV>�o�(
BV9�=+�,.��`�)�w�5�rL����N��`�S���6�v���	C�<�S�<lu������wc2�	���9d%��[�,x�`i����I;���	6��E��T@���E�Y6>�.dE�)Z��v?J��6�u�}I�:����vX��.,���(�H�N�ck��2l
m�`_R�������RtH�]X"8K��|X��d#G	���~���%E�h�T��T��):�)�j�Dl�X~.Rt�S�h���h Eg8��6I��O�Y����C"O��_���hmR�-��
�]�k�����%Ep"E+�K�.�MRI�J���u�IE�h%|I�:�$�)Z��hmR4������&!E��@�n��i��Ks��	}]������"�A�&�w���&9����\�}x�6�{�J�!KG��B�)zcd��#����\l���	���u�IN�h'��Mr2EE�)���E���.�L��������&�P�k�E�b��M ���MB�6�%mR4����(�MB�N)�=\���
H�i����hy�h��"��M���"��	���);��@�&�w���.)�@�)���	�����MB�&�O�K#Ep�����u������X):���.������#Q��b�R
):
����� ���E�������4��BRt#E��A�o!��R4s����[�B�N��`��48�-d��`�@�fR4s|YH�i0�����,�Q�N��{����%�(
.n��Rt�,��RtH��qT���!�n���LWqq�B*o/
�{K��&�Em���

7�|k������)M�^�����������rh�������h�����l�zf����2P�u�m;�Y|��\�
K�.\ZOx�-����q���&�k� X'[n
�uE���
��Il��$�sm[��7 �
 �9��.�`+0�w![���!�
��]�V`n�B�s{B��j�V��@��\~ [����z�
C�h���?�'�Zt��n�<�
�-E�sp��3 �R4��U��
A���$�@0s����/�_z�B�����U�c���/������K
16~���_v(�bl�j��/���/�� ���'���A�c�����?���|%����9 �9�f3�����`�@0s�	�M�&;�,��A�s�_��~
s���[5�����)�$j6�R�Vg���s��
s��$3�u�*�)@pv �g���*&���`}�J��f�������2U~n0Qa��(���xi�B�y	��0wO{A�U��*\���
c\��f3�����`�@0s �9�f3�������_K�����������T��/����a�?<�iZ�����$��q}��3�o-��g����7T��y��r�%���m��K�������e&'b\���
C�o�z�Z,�A�^y]� WXbn_���{VI�����q��Z�����J	�$
���D�0un�ns���nXXfr��L��-h-FpX��'��B�0yn���d8�n�}r��1dS��(�b
��
w�9�}��s����1Z�I��,>�9f,\����!7���s�M���+�zK����8��KN&�����d��L������[GR������'-W�����������&����JM�������`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9�f3�����`�@0s �9��l9i�hIEND�B`�

�PNG


IHDR
�
�J
N��PLTE


			





!!!"""###$$$&&&'''((()))***+++---...///000111222333555666777888999:::;;;<<<===???@@@AAABBBCCCDDDEEEFFFGGGHHHJJJKKKLLLMMMNNNPPPQQQRRRTTTUUUVVVWWWXXXZZZ[[[\\\]]]^^^___aaabbbccceeefffggghhhiiikkklllmmmooopppqqqssstttuuuvvvwwwxxxyyy{{{|||}}}~~~�����������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������������OIDATx����c���M��h���"����H��*�� *)-j��@�����x�
�B���Z�V,(�P"��+k��_�M����������>������}��M>$�}f���4��7����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&�����`���lm���^���f�yDOi�(tom�j}3�<"�k{�����`�Mm��m�M�7�#tn���Z;�o�)�8����L��`F�����A����-��&�Z)0�f��`F��N`@0�M�`'0 ��&A������c�O4�M�`0 m���M"�$��M"���u�,wK�������!�����bQ��8�=���e����+"w[,j�M�y���2����T�^��<n�J���I�c��*��&A��k� ��{�T���
������y�$m�>�h�I
����&���t�)E�MR�����y@0q|J�;nx���_��G�a��S�������~uQl�����a=�����a�a��S��}��������c�����u��1'�N�_�y�$�
��)E3�`>�h��>�h�!�8L�h� E3@p: E!�8L�h� E3@p: E!�8L�h� E3@p: E!�8L�h� E3@p: E!�8L�h� E3x*���!�gG��m��I���Y��J�L��%�  ��uU*d�h� E3���U*d�hq�x@p:@�8H�<��h`���y�����
��A��A@�� E3����`�,�,R4����%���1~�v��`�\��:)f�>��t������a<P�t��`} �
�N�h&��A��A@0B����������|���b�YE(���Q���>56�E(����2*6�]������������0BV
�;[�������0� �
Rt&�h�8P�v����@����#�6iIsH�����_��<i��:(��H�<I�:�����Z���6��&������6��i�I�?����,������CA��6��c���3Y�� 8�|H�te(6�lR:)����y���+CAp�m�:�����N]
���<����%������
����f	���n���Yj;�F#E?q��7�1��r(N�M��J��I���*@A��6	��K�(8�6�i(6�&�\t	3��\�u X�,��=}����_D�.BAp�k�������5����\s(
u�a��"$�:oK�>-6�]����#k�������P���l|�/6�]���d���Pl�M�f�>xY�"��c"��P}L(N�M�a��h�o��J�,���6I���y����i�IY<i���K8)8gSt���>W)�K���C�i��{tD0Rt	���!�M���\aAG�����b�	�\��Q0Rt	�o��CbAG#E�@�6���5���R�L�7x��L5ExRN���'�������)��d�Z���Y�t��n�;R,~���>
F�.�]R4�u��Te�hq���1X�S4 s!�G�H�%���A��A@�^�,�hq���1X�hg.d�()�Rt9UM��g����ne<)z�K�:^���-��D^}qq�v��H���������vY��h��K�c�86k������l@$ECp	�)��`�s�+�=@Z��YH����1X��<��!�_�.Od~S���?$,��`��Q��f�K��g���2aAG#E��?�+��&��v:dQ�]��`2��!���!��`��T#E��]�}?I�qa��M���YpAp#����S��Bp\,�3�2��`�h�Q�l��dY��I3�qa���E���IK�ks�&��1>V�BV9�	�y����F!�R����!%X��@�"E�.��`���}�q���9O�s!���?����w��<o
�|�H�:@0c�'6�1�Py��������qr�]�?��� ��1���6�������������` ���dl�)Ew�6�kF/��`>��)m����Sc���'����_F��!��O)��������)6���"���>�oh��
C�.�f�<|J� ��O!��6i�$�����ij�}h�����on�����h��I�~��+}d=dy�&�&���D�MS�6)R�8d���|x�j9�Ry}���&��U:�_Wy}�R�m���5C����_Jc��`�6����1��p2�c���h��K���F8�uE����Xr{8�9�>�h�6��a��F�18����5a,pN�O!���c��Se��=8�K��NO�dw�'��/bA����qzs����^�zrG����N���
��>~��������d�M��G�!n���t�^c��B���rW�<�����H��b�
e=E��lYO�><��E�S��&�8���BY���B'��Wt0�`]\�C��)Z`]��$�h�u�n�D�z���MB�X��6	)Z�D�%k�/{S"3f%�):0"�����C��=6.xG�����������C�X7�5m^��E�b������YY���CAP��
WM��x��@(E��Iy.c<w���^	�u�� �}Zl�j�u�����E�����'���ll\������Z�m����o��`��}��D�����>��`�,��Y�\�Vg'�(x
�����<~�@�
Y:����(>�h�/��z�f�#����@J�{�u��hWh�����p:E�?B��-�'�������M����������#��:;��Yl�[���t�f!�F����v�f��`�h	>yQ���N�S��mG#EK�-
R�2Vy���F���[G�hqdS����-<�h�N��H���@���]���W�Z��
)Z}^F�V���u@�G�\�H���@���]���Wv�������;+�v��J.�-N5��-�2�?En��\)Z�h	|�V��)�8H��������`/��H��x))Z��	F�v�;g�|-r{��:H���7�1M�#�%_*�
��`/��i��
H��A�&R4q�����c�H��M�h� E)����;���
H��A�&R4q�����c�H���������@x��^��� �z���x[�Nw��)
����8��=������\������YK��LN�-����J) 8 �8H��L�;[�5���
C0�R����B�����a��S��8w<*6�<|����]����`>�����|��5~Av�h�@p:����&)�S�B���O��&)�S�F��2h����>�h�����6I
��MR���6�e�&mR:���?]���3�v�
|
Yq�Z��57��(�$��&�Z��)�<y���������k� 8Na?�E��I���$v{m;��' �	 �8��=}���b�������<�hVl������C�m��b��&�:oK�>-6�N`@��������Kc��&Rt��M����b������'G�5������>�����h�=�N}n]��Ov�^3�5�":'�r���s���h���5* �mhn�:,A�0�i��������������w��,�_��`�zu X�4�r]���GV���t���`���;��>��v��1�%����/zR*�3�;~^H��J���Z�AJ@0q �8L&�����`�@0q �8��m:������se'�3�����������s���o��F�nXXer��.�]����f�b���a�����=t�Za���u#�^��������D���8�h���~�l�p������[����j���=�R���3����U&'b]p�;(�(�%�R(������e�F�v�r�J�
�L��g�|!�NGpX���Q-,=7�
�<y�d���,����,Q�-�6Z������}m��S�������8�����&���G�]�]7��b�3���0�`W��+�J���=0���''H��_a����l���QS>+�Q�6w��=*��/�T�~���9l��u�Y����] �8L&�����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&�����`�@0q �8L&���"����J�IEND�B`�

Index: src/backend/executor/nodeHash.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHash.c,v
retrieving revision 1.116
diff -c -r1.116 nodeHash.c
*** src/backend/executor/nodeHash.c	1 Jan 2008 19:45:49 -0000	1.116
--- src/backend/executor/nodeHash.c	29 Dec 2008 07:38:52 -0000
***************
*** 53,58 ****
--- 53,201 ----
  	return NULL;
  }
  
+ /*
+ *	ExecHashGetMCVPartition
+ *
+ *	returns MCV_INVALID_PARTITION if the hashvalue does not correspond
+ *   to any MCV partition or it corresponds to a partition that has been frozen
+ *	or MCVs are not being used
+ *
+ *   otherwise it returns the index of the MCV partition for this hashvalue
+ *
+ *	it is possible for a non-MCV tuple to hash to an MCV partition due to
+ *	the limited number of hash values but it is unlikely and everything
+ *	continues to work even if it does happen. we would accidentally prioritize
+ *	some less optimal tuples in memory but the result would be accurate
+ *
+ *	hashtable->mostCommonTuplePartition is an open addressing hashtable of
+ *	MCV partitions (HashJoinMostCommonValueTuplePartition)
+ */
+ int 
+ ExecHashGetMCVPartition(HashJoinTable hashtable, uint32 hashvalue)
+ {
+ 	int bucket;
+ 
+ 	if (!hashtable->usingMostCommonValues)
+ 		return MCV_INVALID_PARTITION;
+ 	
+ 	/* modulo the hashvalue (using bitmask) to find the MCV partition */
+ 	bucket = hashvalue & (hashtable->nMostCommonTuplePartitionHashBuckets - 1);
+ 
+ 	/*
+ 	 * while we have not hit a hole in the hashtable and have not hit the actual partition
+ 	 * we have collided in the hashtable so try the next partition slot
+ 	 */
+ 	while (hashtable->mostCommonTuplePartition[bucket].hashvalue != 0
+ 		&& hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 		bucket = (bucket + 1) & (hashtable->nMostCommonTuplePartitionHashBuckets - 1);
+ 
+ 	/* if the partition is not frozen and has been correctly determined return the partition index */
+ 	if (!hashtable->mostCommonTuplePartition[bucket].frozen && hashtable->mostCommonTuplePartition[bucket].hashvalue == hashvalue)
+ 		return bucket;
+ 
+ 	/* must have run into an empty slot which means this is not an MCV*/
+ 	return MCV_INVALID_PARTITION;
+ }
+ 
+ /*
+ *	ExecHashFreezeNextMCVPartition
+ *
+ *	flush the tuples of the next MCV partition by pushing them into the main hashtable
+ */
+ static bool 
+ ExecHashFreezeNextMCVPartition(HashJoinTable hashtable) {
+ 	/*
+ 	 * calculate the flushOrderedMostCommonTuplePartition index of
+ 	 * the partition to flush. not to be confused with the index of
+ 	 * the partition in the MCV partitions hashtable
+ 	 */
+ 	int partitionToFlush = hashtable->nMostCommonTuplePartitions
+ 		- 1 - hashtable->nMostCommonTuplePartitionsFlushed;
+ 	int			bucketno;
+ 	int			batchno;
+ 	uint32		hashvalue;
+ 	HashJoinTuple hashTuple;
+ 	HashJoinTuple nextHashTuple;
+ 	HashJoinMostCommonValueTuplePartition *partition;
+ 	MinimalTuple mintuple;
+ 
+ 	/* if all MCV partitions have already been flushed */
+ 	if (partitionToFlush < 0)
+ 		return false;
+ 
+ 	/* grab a pointer to the actual MCV partition */
+ 	partition = hashtable->flushOrderedMostCommonTuplePartition[partitionToFlush];
+ 	hashvalue = partition->hashvalue;
+ 
+ 	Assert(hashvalue != 0);
+ 
+ 	/* grab a pointer to the first tuple in the soon to be frozen MCV partition */
+ 	hashTuple = partition->tuples;
+ 
+ 	/*
+ 	 * calculate which bucket and batch the tuples belong to in the main
+ 	 * non-MCV hashtable
+ 	 */
+ 	ExecHashGetBucketAndBatch(hashtable, hashvalue,
+ 							  &bucketno, &batchno);
+ 
+ 	/* until we have read all tuples from this partition */
+ 	while (hashTuple != NULL)
+ 	{
+ 		/* decide whether to put the tuple in the hash table or a temp file */
+ 		if (batchno == hashtable->curbatch)
+ 		{
+ 			/* put the tuple in hash table */
+ 			nextHashTuple = hashTuple->next;
+ 			hashTuple->next = hashtable->buckets[bucketno];
+ 			hashtable->buckets[bucketno] = hashTuple;
+ 			
+ 			hashTuple = nextHashTuple;
+ 			hashtable->totalTuples++;
+ 			hashtable->nMostCommonTuplesStored--;
+ 			
+ 			/*
+ 			 * since the tuple is not being removed from memory we may still
+ 			 * be using too much memory.  if this is the first tuple flushed
+ 			 * then we definately must try to free some memory but we must test
+ 			 * again for subsequent tuples in case more memory must be freed.
+ 			 */
+ 			if (hashtable->spaceUsed > hashtable->spaceAllowed)
+ 			{
+ 				ExecHashIncreaseNumBatches(hashtable);
+ 				/* batchno may have changed due to increase in batches */
+ 				ExecHashGetBucketAndBatch(hashtable, hashvalue,
+ 					&bucketno, &batchno);
+ 			}
+ 		}
+ 		else
+ 		{
+ 			/* put the tuples into a temp file for later batches */
+ 			Assert(batchno > hashtable->curbatch);
+ 			mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 			ExecHashJoinSaveTuple(mintuple,
+ 								  hashvalue,
+ 								  &hashtable->innerBatchFile[batchno]);
+ 			/*
+ 			 * some memory has been freed up. this must be done before we
+ 			 * pfree the hashTuple of we lose access to the tuple size
+ 			 */
+ 			hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			nextHashTuple = hashTuple->next;
+ 			pfree(hashTuple);
+ 			hashTuple = nextHashTuple;
+ 			hashtable->totalTuples++;
+ 			hashtable->nMostCommonTuplesStored--;
+ 		}
+ 	}
+ 
+ 	partition->frozen = true;
+ 	partition->tuples = NULL;
+ 	hashtable->nMostCommonTuplePartitionsFlushed++;
+ 
+ 	return true;
+ }
+ 
  /* ----------------------------------------------------------------
   *		MultiExecHash
   *
***************
*** 69,74 ****
--- 212,219 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int partitionNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,106 ****
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
  	}
  
--- 244,279 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			partitionNumber = ExecHashGetMCVPartition(hashtable, hashvalue);
! 
! 			/* if this tuple belongs in an MCV partition */
! 			if (partitionNumber != MCV_INVALID_PARTITION)
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				/* get the HashJoinTuple */
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple = (HashJoinTuple) palloc(hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple, mintuple->t_len);
! 
! 				/* push the HashJoinTuple onto the front of the MCV partition tuple list */
! 				hashTuple->next = hashtable->mostCommonTuplePartition[partitionNumber].tuples;
! 				hashtable->mostCommonTuplePartition[partitionNumber].tuples = hashTuple;
! 				
! 				/* move memory is now in use so make sure we are not over memory */
! 				hashtable->spaceUsed += hashTupleSize;
! 				hashtable->nMostCommonTuplesStored++;
! 				while (hashtable->spaceUsed > hashtable->spaceAllowed && ExecHashFreezeNextMCVPartition(hashtable))
! 					;
! 			}
! 			else
! 			{
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 				hashtable->totalTuples += 1;
! 			}
  		}
  	}
  
***************
*** 269,274 ****
--- 442,454 ----
  	hashtable->outerBatchFile = NULL;
  	hashtable->spaceUsed = 0;
  	hashtable->spaceAllowed = work_mem * 1024L;
+ 	/* initialize MCV related hashtable variables */
+ 	hashtable->usingMostCommonValues = false;
+ 	hashtable->nMostCommonTuplePartitions = 0;
+ 	hashtable->nMostCommonTuplePartitionHashBuckets = 0;
+ 	hashtable->nMostCommonTuplesStored = 0;
+ 	hashtable->mostCommonTuplePartition = NULL;
+ 	hashtable->nMostCommonTuplePartitionsFlushed = 0;
  
  	/*
  	 * Get info about the hash functions to be used for each hash key. Also
***************
*** 566,571 ****
--- 746,752 ----
  				ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(tuple),
  									  tuple->hashvalue,
  									  &hashtable->innerBatchFile[batchno]);
+ 
  				/* and remove from hash table */
  				if (prevtuple)
  					prevtuple->next = nexttuple;
***************
*** 798,803 ****
--- 979,1046 ----
  }
  
  /*
+  * ExecScanHashMostCommonTuples
+  *		scan a MCV partition for matches to the current outer tuple
+  *
+  * The current outer tuple must be stored in econtext->ecxt_outertuple.
+  */
+ HashJoinTuple
+ ExecScanHashMostCommonTuples(HashJoinState *hjstate,
+ 				   ExprContext *econtext)
+ {
+ 	List	   *hjclauses = hjstate->hashclauses;
+ 	HashJoinTable hashtable = hjstate->hj_HashTable;
+ 	HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+ 	uint32		hashvalue = hjstate->hj_CurHashValue;
+ 
+ 	/*
+ 	 * hj_CurTuple is NULL to start scanning a new partition, or the address of
+ 	 * the last tuple returned from the current partition.
+ 	 */
+ 	if (hashTuple == NULL)
+ 	{
+ 		/* painstakingly make sure this is a valid partition index */
+ 		Assert(hjstate->hj_OuterTupleMostCommonValuePartition > MCV_INVALID_PARTITION);
+ 		Assert(hjstate->hj_OuterTupleMostCommonValuePartition < hashtable->nMostCommonTuplePartitionHashBuckets);
+ 		Assert(hashtable->mostCommonTuplePartition[hjstate->hj_OuterTupleMostCommonValuePartition].hashvalue != 0);
+ 
+ 		hashTuple = hashtable->mostCommonTuplePartition[hjstate->hj_OuterTupleMostCommonValuePartition].tuples;
+ 	}
+ 	else
+ 		hashTuple = hashTuple->next;
+ 
+ 	while (hashTuple != NULL)
+ 	{
+ 		if (hashTuple->hashvalue == hashvalue)
+ 		{
+ 			TupleTableSlot *inntuple;
+ 
+ 			/* insert hashtable's tuple into exec slot so ExecQual sees it */
+ 			inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ 											 hjstate->hj_HashTupleSlot,
+ 											 false);	/* do not pfree */
+ 			econtext->ecxt_innertuple = inntuple;
+ 
+ 			/* reset temp memory each time to avoid leaks from qual expr */
+ 			ResetExprContext(econtext);
+ 
+ 			if (ExecQual(hjclauses, econtext, false))
+ 			{
+ 				hjstate->hj_CurTuple = hashTuple;
+ 				return hashTuple;
+ 			}
+ 		}
+ 
+ 		hashTuple = hashTuple->next;
+ 	}
+ 
+ 	/*
+ 	 * no match
+ 	 */
+ 	return NULL;
+ }
+ 
+ /*
   * ExecScanHashBucket
   *		scan a hash bucket for matches to the current outer tuple
   *
Index: src/backend/executor/nodeHashjoin.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHashjoin.c,v
retrieving revision 1.96
diff -c -r1.96 nodeHashjoin.c
*** src/backend/executor/nodeHashjoin.c	23 Oct 2008 14:34:34 -0000	1.96
--- src/backend/executor/nodeHashjoin.c	29 Dec 2008 04:34:46 -0000
***************
*** 20,25 ****
--- 20,30 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "optimizer/cost.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 ****
--- 39,159 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+  *	ExecHashJoinGetMostCommonValues
+  *
+  *	if the MCV statistics can be found for the join attribute of this
+  *	hashjoin 
+  *
+  */
+ static void 
+ ExecHashJoinGetMostCommonValues(EState *estate, HashJoinState *hjstate)
+ {
+ 	HeapTupleData *statsTuple;
+ 	FuncExprState *clause;
+ 	ExprState *argstate;
+ 	Var *variable;
+ 
+ 	Datum	   *values;
+ 	int			nvalues;
+ 	float4	   *numbers;
+ 	int			nnumbers;
+ 
+ 	Oid relid;
+ 	Oid atttype;
+ 
+ 	int i;
+ 
+ 	/* Only use statistics if there is a single join attribute. */
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; /* histojoin is not defined for more than one join key so run away */
+ 
+ 	/* Determine the relation id and attribute id of the single join attribute of the probe relation. */
+ 	
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 
+ 	/* Do not try to exploit stats if the join attribute is an expression instead of just a simple attribute. */		
+ 	if (argstate->expr->type != T_Var)
+ 		return;
+ 
+ 	variable = (Var *) argstate->expr;
+ 	relid = getrelid(variable->varnoold, estate->es_range_table);
+ 
+ 	/* grab the necessary properties of the join variable */
+ 	atttype = variable->vartype;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT, ObjectIdGetDatum(relid),
+ 		Int16GetDatum(variable->varoattno), 0, 0);
+ 
+ 	if (!HeapTupleIsValid(statsTuple))
+ 		return;
+ 
+ 	/* if there are MCV statistics for the attribute */
+ 	if (get_attstatsslot(statsTuple,
+ 		atttype, variable->vartypmod,
+ 		STATISTIC_KIND_MCV, InvalidOid,
+ 		&values, &nvalues,
+ 		&numbers, &nnumbers))
+ 	{
+ 		MemoryContext oldcxt;
+ 		HashJoinTable hashtable;
+ 		FmgrInfo   *hashfunctions;
+ 		/* MCV Partitions is an open addressing hashtable with a 
+ 		power of 2 size greater than the number of MCV values. */
+ 		int nbuckets = 2;
+ 
+ 		while (nbuckets <= nvalues)
+ 			nbuckets <<= 1;
+ 		/* use two more bits just to help avoid collisions */
+ 		nbuckets <<= 2;
+ 
+ 		hashtable = hjstate->hj_HashTable;
+ 		hashtable->usingMostCommonValues = true;
+ 		hashtable->nMostCommonTuplePartitionHashBuckets = nbuckets;
+ 
+ 		/* allocate the partition memory in the hashtable's memory context */
+ 		oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+ 		hashtable->mostCommonTuplePartition = palloc0(nbuckets * sizeof(HashJoinMostCommonValueTuplePartition));
+ 		hashtable->flushOrderedMostCommonTuplePartition = palloc0(nvalues * sizeof(HashJoinMostCommonValueTuplePartition*));
+ 		MemoryContextSwitchTo(oldcxt);
+ 
+ 		/* grab the hash functions as we will be generating the hashvalues here */
+ 		hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 		/* create the partitions */
+ 		for (i = 0; i < nvalues; i++)
+ 		{
+ 			uint32 hashvalue = DatumGetUInt32(FunctionCall1(&hashfunctions[0], values[i]));
+ 			int bucket = hashvalue & (nbuckets - 1);
+ 
+ 			/*
+ 			 * while we have not hit a hole in the hashtable and have not hit a partition
+ 			 * with the same hashvalue we have collided in the hashtable so try the next
+ 			 * partition slot (remember it is an open addressing hashtable)
+ 			 */
+ 			while (hashtable->mostCommonTuplePartition[bucket].hashvalue != 0
+ 				&& hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 				bucket = (bucket + 1) & (nbuckets - 1);
+ 
+ 			/*
+ 			 * leave partition alone if it has the same hashvalue as current MCV.  
+ 			 * we only want one partition per hashvalue. even if two MCV values
+ 			 * hash to the same partition we are fine
+ 			 */
+ 			if (hashtable->mostCommonTuplePartition[bucket].hashvalue != hashvalue)
+ 			{
+ 				hashtable->mostCommonTuplePartition[bucket].hashvalue = hashvalue;
+ 				hashtable->flushOrderedMostCommonTuplePartition[hashtable->nMostCommonTuplePartitions] = &hashtable->mostCommonTuplePartition[bucket];
+ 				hashtable->nMostCommonTuplePartitions++;
+ 			}
+ 		}
+ 
+ 		free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 	}
+ 
+ 	ReleaseSysCache(statsTuple);
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 147,152 ****
--- 267,276 ----
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
  
+ 		/* we dont bother exploiting MCVs if we can do the entire join in memory */
+ 		if (hashtable->nbatch > 1)
+ 			ExecHashJoinGetMostCommonValues(estate, node);
+ 
  		/*
  		 * execute the Hash node, to build the hash table
  		 */
***************
*** 157,163 ****
  		 * If the inner relation is completely empty, and we're not doing an
  		 * outer join, we can quit without scanning the outer relation.
  		 */
! 		if (hashtable->totalTuples == 0 && !HASHJOIN_IS_OUTER(node))
  			return NULL;
  
  		/*
--- 281,287 ----
  		 * If the inner relation is completely empty, and we're not doing an
  		 * outer join, we can quit without scanning the outer relation.
  		 */
! 		if (hashtable->totalTuples == 0 && hashtable->nMostCommonTuplesStored == 0 && !HASHJOIN_IS_OUTER(node))
  			return NULL;
  
  		/*
***************
*** 205,227 ****
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
  
! 			/*
! 			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
! 			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
! 				 * Need to postpone this outer tuple to a later batch. Save it
! 				 * in the corresponding outer-batch file.
  				 */
! 				Assert(batchno > hashtable->curbatch);
! 				ExecHashJoinSaveTuple(ExecFetchSlotMinimalTuple(outerTupleSlot),
! 									  hashvalue,
! 									  &hashtable->outerBatchFile[batchno]);
! 				node->hj_NeedNewOuter = true;
! 				continue;		/* loop around for a new outer tuple */
  			}
  		}
  
--- 329,356 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
+ 			
+ 			node->hj_OuterTupleMostCommonValuePartition = ExecHashGetMCVPartition(hashtable, hashvalue);
  
! 			if (node->hj_OuterTupleMostCommonValuePartition == MCV_INVALID_PARTITION)
  			{
  				/*
! 				 * Now we've got an outer tuple and the corresponding hash bucket,
! 				 * but this tuple may not belong to the current batch.
  				 */
! 				if (batchno != hashtable->curbatch)
! 				{
! 					/*
! 					 * Need to postpone this outer tuple to a later batch. Save it
! 					 * in the corresponding outer-batch file.
! 					 */
! 					Assert(batchno > hashtable->curbatch);
! 					ExecHashJoinSaveTuple(ExecFetchSlotMinimalTuple(outerTupleSlot),
! 										  hashvalue,
! 										  &hashtable->outerBatchFile[batchno]);
! 					node->hj_NeedNewOuter = true;
! 					continue;		/* loop around for a new outer tuple */
! 				}
  			}
  		}
  
***************
*** 230,236 ****
  		 */
  		for (;;)
  		{
! 			curtuple = ExecScanHashBucket(node, econtext);
  			if (curtuple == NULL)
  				break;			/* out of matches */
  
--- 359,373 ----
  		 */
  		for (;;)
  		{
! 			/* if the tuple hashed to an MCV partition then scan the MCV tuples */
! 			if (node->hj_OuterTupleMostCommonValuePartition != MCV_INVALID_PARTITION)
! 			{
! 				curtuple = ExecScanHashMostCommonTuples(node, econtext);
! 			}
! 			else /* otherwise scan the standard hashtable buckets */
! 			{
! 				curtuple = ExecScanHashBucket(node, econtext);
! 			}
  			if (curtuple == NULL)
  				break;			/* out of matches */
  
Index: src/include/executor/hashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/hashjoin.h,v
retrieving revision 1.48
diff -c -r1.48 hashjoin.h
*** src/include/executor/hashjoin.h	1 Jan 2008 19:45:57 -0000	1.48
--- src/include/executor/hashjoin.h	29 Dec 2008 01:39:50 -0000
***************
*** 72,77 ****
--- 72,85 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ typedef struct HashJoinMostCommonValueTuplePartition
+ {
+ 	uint32 hashvalue;
+ 	bool frozen;
+ 	HashJoinTuple tuples;
+ } HashJoinMostCommonValueTuplePartition;
+ 
+ #define MCV_INVALID_PARTITION -1
  
  typedef struct HashJoinTableData
  {
***************
*** 116,121 ****
--- 124,141 ----
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	bool usingMostCommonValues;	/* will the join use MCV partitions */
+ 	HashJoinMostCommonValueTuplePartition *mostCommonTuplePartition; /* hashtable of MCV partitions */
+ 	/*
+ 	 * array of pointers to the MCV partitions hashtable buckets in the opposite order that
+ 	 * they would be flushed to disk
+ 	 */
+ 	HashJoinMostCommonValueTuplePartition **flushOrderedMostCommonTuplePartition;
+ 	int nMostCommonTuplePartitionHashBuckets; /* # of buckets in the MCV partitions hashtable */
+ 	int nMostCommonTuplePartitions; /* # of actual partitions in the MCV partitions hashtable */
+ 	int nMostCommonTuplePartitionsFlushed; /* # of MCV partitions that have already been flushed to disk */
+ 	uint32 nMostCommonTuplesStored; /* total # of build tuples currently stored in the MCV partitions */
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
Index: src/include/executor/nodeHash.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHash.h,v
retrieving revision 1.45
diff -c -r1.45 nodeHash.h
*** src/include/executor/nodeHash.h	1 Jan 2008 19:45:57 -0000	1.45
--- src/include/executor/nodeHash.h	23 Dec 2008 07:54:03 -0000
***************
*** 44,48 ****
--- 44,51 ----
  extern void ExecChooseHashTableSize(double ntuples, int tupwidth,
  						int *numbuckets,
  						int *numbatches);
+ 						
+ extern HashJoinTuple ExecScanHashMostCommonTuples(HashJoinState *hjstate, ExprContext *econtext);
+ extern int ExecHashGetMCVPartition(HashJoinTable hashtable, uint32 hashvalue);
  
  #endif   /* NODEHASH_H */
Index: src/include/nodes/execnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/execnodes.h,v
retrieving revision 1.197
diff -c -r1.197 execnodes.h
*** src/include/nodes/execnodes.h	28 Dec 2008 18:54:00 -0000	1.197
--- src/include/nodes/execnodes.h	29 Dec 2008 04:40:02 -0000
***************
*** 1381,1386 ****
--- 1381,1387 ----
   *		hj_NeedNewOuter			true if need new outer tuple on next call
   *		hj_MatchedOuter			true if found a join match for current outer
   *		hj_OuterNotEmpty		true if outer relation known not empty
+  *		hj_OuterTupleMostCommonValuePartition	partition# for the current outer tuple
   * ----------------
   */
  
***************
*** 1406,1411 ****
--- 1407,1413 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int		hj_OuterTupleMostCommonValuePartition;
  } HashJoinState;
  
  

Index: src/backend/executor/nodeHash.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHash.c,v
retrieving revision 1.117
diff -c -r1.117 nodeHash.c
*** src/backend/executor/nodeHash.c	1 Jan 2009 17:23:41 -0000	1.117
--- src/backend/executor/nodeHash.c	6 Jan 2009 23:18:16 -0000
***************
*** 53,58 ****
--- 53,220 ----
  	return NULL;
  }
  
+ /*
+ *	ExecHashGetIMBucket
+ *
+ *	Returns IM_INVALID_BUCKET if the hashvalue does not correspond
+ *	to any IM bucket or it corresponds to a bucket that has been frozen
+ *	or skew optimization is not being used.
+ *
+ *	Otherwise it returns the index of the IM bucket for this hashvalue.
+ *
+ *	It is possible for a tuple whose join attribute value is not a MCV to
+ *	hash to an IM bucket due to the limited number of hash values but it is
+ *	unlikely and everything continues to work even if it does happen. We would
+ *	accidentally prioritize some less optimal tuples in memory but the join
+ *	result would still be accurate.
+ *
+ *	hashtable->imBucket is an open addressing hashtable of
+ *	IM buckets (HashJoinIMBucket).
+ */
+ int 
+ ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue)
+ {
+ 	int bucket;
+ 
+ 	if (!hashtable->enableSkewOptimization)
+ 		return IM_INVALID_BUCKET;
+ 	
+ 	/* Modulo the hashvalue (using bitmask) to find the IM bucket. */
+ 	bucket = hashvalue & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * While we have not hit a hole in the hashtable and have not hit the 
+ 	 * actual bucket we have collided in the hashtable so try the next
+ 	 * bucket location.
+ 	 */
+ 	while (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 		bucket = (bucket + 1) & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * If the bucket exists and has been correctly determined return
+ 	 * the bucket index.
+ 	 */
+ 	if (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue == hashvalue)
+ 		return bucket;
+ 
+ 	/*
+ 	 * Must have run into an empty location or a frozen bucket which means the
+ 	 * tuple with this hashvalue is not to be handled as if it matches with an
+ 	 * IM bucket.
+ 	 */
+ 	return IM_INVALID_BUCKET;
+ }
+ 
+ /*
+ *	ExecHashFreezeNextIMBucket
+ *
+ *	Freeze the tuples of the next IM bucket by pushing them into the main
+ *	hashtable.  Buckets are frozen in order so that the best tuples are kept
+ *	in memory the longest.
+ */
+ static bool 
+ ExecHashFreezeNextIMBucket(HashJoinTable hashtable) {
+ 	int						bucketToFreeze;
+ 	int						bucketno;
+ 	int						batchno;
+ 	uint32					hashvalue;
+ 	HashJoinTuple			hashTuple;
+ 	HashJoinTuple			nextHashTuple;
+ 	HashJoinIMBucket		*bucket;
+ 	MinimalTuple			mintuple;
+ 
+ 	/* Calculate the imBucket index of the bucket to freeze. */
+ 	bucketToFreeze = hashtable->imBucketFreezeOrder
+ 		[hashtable->nUsedIMBuckets - 1 - hashtable->nIMBucketsFrozen];
+ 
+ 	/* Grab a pointer to the actual IM bucket. */
+ 	bucket = hashtable->imBucket[bucketToFreeze];
+ 	hashvalue = bucket->hashvalue;
+ 
+ 	/*
+ 	 * Grab a pointer to the first tuple in the soon to be frozen IM bucket.
+ 	 */
+ 	hashTuple = bucket->tuples;
+ 
+ 	/*
+ 	 * Calculate which bucket and batch the tuples belong to in the main
+ 	 * non-IM hashtable.
+ 	 */
+ 	ExecHashGetBucketAndBatch(hashtable, hashvalue,
+ 							  &bucketno, &batchno);
+ 
+ 	/* until we have read all tuples from this bucket */
+ 	while (hashTuple != NULL)
+ 	{
+ 		/*
+ 		 * Some of this code is very similar to that of ExecHashTableInsert.
+ 		 * We do not call ExecHashTableInsert directly as
+ 		 * ExecHashTableInsert expects a TupleTableSlot and we already have
+ 		 * HashJoinTuples.
+ 		 */
+ 		
+ 		mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 
+ 		/* Decide whether to put the tuple in the hash table or a temp file. */
+ 		if (batchno == hashtable->curbatch)
+ 		{
+ 			/* Put the tuple in hash table. */
+ 			nextHashTuple = hashTuple->next;
+ 			hashTuple->next = hashtable->buckets[bucketno];
+ 			hashtable->buckets[bucketno] = hashTuple;
+ 			
+ 			hashTuple = nextHashTuple;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 		}
+ 		else
+ 		{
+ 			/* Put the tuples into a temp file for later batches. */
+ 			Assert(batchno > hashtable->curbatch);
+ 			ExecHashJoinSaveTuple(mintuple, hashvalue,
+ 								  &hashtable->innerBatchFile[batchno]);
+ 			/*
+ 			 * Some memory has been freed up. This must be done before we
+ 			 * pfree the hashTuple of we lose access to the tuple size.
+ 			 */
+ 			hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			nextHashTuple = hashTuple->next;
+ 			pfree(hashTuple);
+ 			hashTuple = nextHashTuple;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * Free the memory the bucket struct was using as it is not necessary
+ 	 * any more.  All code treats a frozen IM bucket the same as one that
+ 	 * did not exist so by setting the pointer to null the rest of the code
+ 	 * will function as if we had not created this IM bucket at all.
+ 	 */
+ 	pfree(bucket);
+ 	hashtable->imBucket[bucketToFreeze] = NULL;
+ 	hashtable->spaceUsed -= IM_BUCKET_OVERHEAD;
+ 	hashtable->spaceUsedIM -= IM_BUCKET_OVERHEAD;
+ 	hashtable->nIMBucketsFrozen++;
+ 
+ 	/*
+ 	 * All IM buckets have been frozen and deleted from memory so turn off
+ 	 * skew aware partitioning and remove the structs from memory as they are
+ 	 * just wasting space from now on.
+ 	 */
+ 	if (hashtable->nUsedIMBuckets == hashtable->nIMBucketsFrozen)
+ 	{
+ 		hashtable->enableSkewOptimization = false;
+ 		pfree(hashtable->imBucket);
+ 		pfree(hashtable->imBucketFreezeOrder);
+ 		hashtable->spaceUsed -= hashtable->spaceUsedIM;
+ 		hashtable->spaceUsedIM = 0;
+ 	}
+ 
+ 	return true;
+ }
+ 
  /* ----------------------------------------------------------------
   *		MultiExecHash
   *
***************
*** 69,74 ****
--- 231,238 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int bucketNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,105 ****
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
  			hashtable->totalTuples += 1;
  		}
  	}
--- 263,307 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			bucketNumber = ExecHashGetIMBucket(hashtable, hashvalue);
! 
! 			/* if this tuple does not belong in an IM bucket */
! 			if (bucketNumber == IM_INVALID_BUCKET)
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 			else
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				/* get the HashJoinTuple */
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple
! 					= (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
! 														hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple
! 					, mintuple->t_len);
! 
! 				/* Push the HashJoinTuple onto the front of the IM bucket. */
! 				hashTuple->next 
! 					= hashtable->imBucket[bucketNumber]->tuples;
! 				hashtable->imBucket[bucketNumber]->tuples
! 					= hashTuple;
! 				
! 				/*
! 				 * More memory is now in use so make sure we are not over
! 				 * spaceAllowedIM.
! 				 */
! 				hashtable->spaceUsed += hashTupleSize;
! 				hashtable->spaceUsedIM += hashTupleSize;
! 				while (hashtable->spaceUsedIM > hashtable->spaceAllowedIM
! 					&& ExecHashFreezeNextIMBucket(hashtable))
! 					;
! 				/* Guarantee we are not over the spaceAllowed. */
! 				if (hashtable->spaceUsed > hashtable->spaceAllowed)
! 					ExecHashIncreaseNumBatches(hashtable);
! 			}
  			hashtable->totalTuples += 1;
  		}
  	}
***************
*** 269,274 ****
--- 471,485 ----
  	hashtable->outerBatchFile = NULL;
  	hashtable->spaceUsed = 0;
  	hashtable->spaceAllowed = work_mem * 1024L;
+ 	/* Initialize skew optimization related hashtable variables. */
+ 	hashtable->spaceUsedIM = 0;
+ 	hashtable->spaceAllowedIM
+ 		= hashtable->spaceAllowed * IM_WORK_MEM_PERCENT / 100;
+ 	hashtable->enableSkewOptimization = false;
+ 	hashtable->nUsedIMBuckets = 0;
+ 	hashtable->nIMBuckets = 0;
+ 	hashtable->imBucket = NULL;
+ 	hashtable->nIMBucketsFrozen = 0;
  
  	/*
  	 * Get info about the hash functions to be used for each hash key. Also
***************
*** 815,823 ****
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
  	 */
  	if (hashTuple == NULL)
! 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
  	else
  		hashTuple = hashTuple->next;
  
--- 1026,1041 ----
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
+ 	 *
+ 	 * If the tuple hashed to an IM bucket then scan the IM bucket
+ 	 * otherwise scan the standard hashtable bucket.
  	 */
  	if (hashTuple == NULL)
! 		if (hjstate->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
! 			hashTuple = hashtable->imBucket[hjstate->hj_OuterTupleIMBucketNo]
! 											->tuples;
! 		else
! 			hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
  	else
  		hashTuple = hashTuple->next;
  
Index: src/backend/executor/nodeHashjoin.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHashjoin.c,v
retrieving revision 1.97
diff -c -r1.97 nodeHashjoin.c
*** src/backend/executor/nodeHashjoin.c	1 Jan 2009 17:23:41 -0000	1.97
--- src/backend/executor/nodeHashjoin.c	6 Jan 2009 23:51:01 -0000
***************
*** 20,25 ****
--- 20,29 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 ****
--- 38,234 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+  *	ExecHashJoinDetectSkew
+  *
+  *	If MCV statistics can be found for the join attribute of this hashjoin
+  *	then create a hash table of buckets. Each bucket will correspond to
+  *	a MCV hashvalue and will be filled with inner relation tuples whose join
+  *	attribute hashes to the same value as that MCV.  If a join attribute
+  *	value is a MCV for the join attribute in the outer (probe) relation,
+  *	tuples with this value in the inner (build) relation are more likely to
+  *	join with outer relation tuples and a benefit can be gained by keeping
+  *	them in memory while joining the first batch of tuples.
+  */
+ static void 
+ ExecHashJoinDetectSkew(EState *estate, HashJoinState *hjstate, int tupwidth)
+ {
+ 	HeapTupleData	*statsTuple;
+ 	FuncExprState	*clause;
+ 	ExprState		*argstate;
+ 	Var				*variable;
+ 	HashJoinTable	hashtable;
+ 	Datum			*values;
+ 	int				nvalues;
+ 	float4			*numbers;
+ 	int				nnumbers;
+ 	Oid				relid;
+ 	Oid				atttype;
+ 	int				i;
+ 	int				mcvsToUse;
+ 
+ 	/* Only use statistics if there is a single join attribute. */
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; /* Histojoin is not defined for more than one join key */
+ 	
+ 	hashtable = hjstate->hj_HashTable;
+ 	
+ 	/*
+ 	 * Estimate the number of IM buckets that will fit in
+ 	 * the memory allowed for IM buckets.
+ 	 *
+ 	 * hashtable->imBucket will have up to 8 times as many HashJoinIMBucket
+ 	 * pointers as the number of MCV hashvalues. A uint16 index in
+ 	 * hashtable->imBucketFreezeOrder will be created for each IM bucket. One
+ 	 * actual HashJoinIMBucket struct will be created for each
+ 	 * unique MCV hashvalue so up to one struct per MCV.
+ 	 *
+ 	 * It is also estimated that each IM bucket will have a single build
+ 	 * tuple stored in it after partitioning the build relation input.  This
+ 	 * estimate could be high if tuples are filtered out before this join but
+ 	 * in that case the extra memory is used by the regular hashjoin batch.
+ 	 * This estimate could be low if it is a many to many join but in that
+ 	 * case IM buckets will be frozen to free up memory as needed
+ 	 * during the inner relation partitioning phase.
+ 	 */
+ 	mcvsToUse = hashtable->spaceAllowedIM / (
+ 		/* size of a hash tuple */
+ 		HJTUPLE_OVERHEAD + MAXALIGN(sizeof(MinimalTupleData))
+ 			+ MAXALIGN(tupwidth)
+ 		/* max size of hashtable pointers per MCV */
+ 		+ (8 * sizeof(HashJoinIMBucket*))
+ 		+ sizeof(uint16) /* size of imBucketFreezeOrder entry */
+ 		+ IM_BUCKET_OVERHEAD /* size of IM bucket struct */
+ 		);
+ 
+ 	/*
+ 	 * If we cannot fit any MCV tuples in memory then it is not necessary to
+ 	 * even look at the MCVs.
+ 	 */
+ 	if (mcvsToUse == 0)
+ 		return;
+ 
+ 	/*
+ 	 * Determine the relation id and attribute id of the single join
+ 	 * attribute of the probe relation.
+ 	 */
+ 
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 
+ 	/*
+ 	 * Do not try to exploit stats if the join attribute is an expression
+ 	 * instead of just a simple attribute.
+ 	 */		
+ 	if (argstate->expr->type != T_Var)
+ 		return;
+ 
+ 	variable = (Var *) argstate->expr;
+ 	relid = getrelid(variable->varnoold, estate->es_range_table);
+ 	atttype = variable->vartype;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT, ObjectIdGetDatum(relid),
+ 		Int16GetDatum(variable->varoattno), 0, 0);
+ 
+ 	if (!HeapTupleIsValid(statsTuple))
+ 		return;
+ 
+ 	/* if there are MCV statistics for the attribute */
+ 	if (get_attstatsslot(statsTuple,
+ 		atttype, variable->vartypmod,
+ 		STATISTIC_KIND_MCV, InvalidOid,
+ 		&values, &nvalues,
+ 		&numbers, &nnumbers))
+ 	{
+ 		FmgrInfo   *hashfunctions;
+ 		/*
+ 		 * IM buckets (imBucket) is an open addressing hashtable with a 
+ 		 * power of 2 size that is greater than the number of MCV values.
+ 		 */
+ 		int nbuckets = 2;
+ 
+ 		if (mcvsToUse > nvalues)
+ 			mcvsToUse = nvalues;
+ 
+ 		while (nbuckets <= mcvsToUse)
+ 			nbuckets <<= 1;
+ 		/* use two more bits just to help avoid collisions */
+ 		nbuckets <<= 2;
+ 
+ 		hashtable->enableSkewOptimization = true;
+ 		hashtable->nIMBuckets = nbuckets;
+ 
+ 		/*
+ 		 * Allocate the bucket memory in the hashtable's batch context
+ 		 * because it is only relevant and necessary during the first batch
+ 		 * and will be nicely removed once the first batch is done.
+ 		 */
+ 		hashtable->imBucket = 
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				nbuckets * sizeof(HashJoinIMBucket*));
+ 		hashtable->imBucketFreezeOrder = 
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				mcvsToUse * sizeof(uint16));
+ 		/* Count the overhead of the IM pointers immediately. */
+ 		hashtable->spaceUsed += nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 		hashtable->spaceUsedIM +=  nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 
+ 		/*
+ 		 * Grab the hash functions as we will be generating the hashvalues
+ 		 * in this section.
+ 		 */
+ 		hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 		/* Create the buckets */
+ 		for (i = 0; i < mcvsToUse; i++)
+ 		{
+ 			uint32 hashvalue = DatumGetUInt32(
+ 				FunctionCall1(&hashfunctions[0], values[i]));
+ 			int bucket = hashvalue & (nbuckets - 1);
+ 
+ 			/*
+ 			 * While we have not hit a hole in the hashtable and have not hit
+ 			 * a bucket with the same hashvalue we have collided in the
+ 			 * hashtable so try the next bucket location (remember it is an
+ 			 * open addressing hashtable).
+ 			 */
+ 			while (hashtable->imBucket[bucket] != NULL
+ 				&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 				bucket = (bucket + 1) & (nbuckets - 1);
+ 
+ 			/*
+ 			 * Leave bucket alone if it has the same hashvalue as current
+ 			 * MCV. We only want one bucket per hashvalue. Even if two MCV
+ 			 * values hash to the same bucket we are fine.
+ 			 */
+ 			if (hashtable->imBucket[bucket] == NULL)
+ 			{
+ 				/*
+ 				 * Allocate the actual bucket structure in the hashtable's batch
+ 				 * context because it is only relevant and necessary during
+ 				 * the first batch and will be nicely removed once the first
+ 				 * batch is done.
+ 				 */
+ 				hashtable->imBucket[bucket]
+ 					= MemoryContextAllocZero(hashtable->batchCxt,
+ 						sizeof(HashJoinIMBucket));
+ 				hashtable->imBucket[bucket]->hashvalue = hashvalue;
+ 				hashtable->imBucketFreezeOrder[hashtable->nUsedIMBuckets]
+ 					= bucket;
+ 				hashtable->nUsedIMBuckets++;
+ 				/* Count the overhead of the IM bucket struct */
+ 				hashtable->spaceUsed += IM_BUCKET_OVERHEAD;
+ 				hashtable->spaceUsedIM += IM_BUCKET_OVERHEAD;
+ 			}
+ 		}
+ 
+ 		free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 	}
+ 
+ 	ReleaseSysCache(statsTuple);
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 147,152 ****
--- 342,352 ----
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
  
+ 		/* Use skew optimization only when there is more than one batch. */
+ 		if (hashtable->nbatch > 1)
+ 			ExecHashJoinDetectSkew(estate, node,
+ 				(outerPlan((Hash *) hashNode->ps.plan))->plan_width );
+ 
  		/*
  		 * execute the Hash node, to build the hash table
  		 */
***************
*** 205,216 ****
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
  
  			/*
! 			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
  			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
--- 405,423 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
+ 			
+ 			/* Does the outer tuple match an IM bucket? */
+ 			node->hj_OuterTupleIMBucketNo = 
+ 				ExecHashGetIMBucket(hashtable, hashvalue);
  
  			/*
! 			 * Now we've got an outer tuple and the corresponding hash bucket.
! 			 *
! 			 * If the outer tuple does not match an IM bucket and it does not
! 			 * belong to the current batch.
  			 */
! 			if (node->hj_OuterTupleIMBucketNo == IM_INVALID_BUCKET
! 				&& batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
***************
*** 641,647 ****
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
--- 848,874 ----
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	/* if we just finished the first batch */
! 	if (curbatch == 0)
! 	{
! 		/*
! 		 * Reset some of the skew optimization state variables. IM buckets are
! 		 * no longer being used as of this point because they are only
! 		 * necessary while joining the first batch (before the cleanup phase).
! 		 *
! 		 * Especially need to make sure ExecHashGetIMBucket returns
! 		 * IM_INVALID_BUCKET quickly for all subsequent calls.
! 		 *
! 		 * IM buckets are only taking up memory if this is a multi-batch join
! 		 * and in that case ExecHashTableReset is about to be called which
! 		 * will free all memory currently used by IM buckets and tuples when
! 		 * it deletes hashtable->batchCxt.  If this is a single batch join
! 		 * then imBucket and imBucketFreezeOrder are already NULL and empty.
! 		 */
! 		hashtable->enableSkewOptimization = false;
! 		hashtable->spaceUsedIM = 0;
! 	}
! 	else if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
Index: src/include/executor/hashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/hashjoin.h,v
retrieving revision 1.49
diff -c -r1.49 hashjoin.h
*** src/include/executor/hashjoin.h	1 Jan 2009 17:23:59 -0000	1.49
--- src/include/executor/hashjoin.h	6 Jan 2009 23:13:19 -0000
***************
*** 72,77 ****
--- 72,96 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ /*
+  * Stores a hashvalue and linked list of tuples that share that hashvalue.
+  *
+  * When processing MCVs to detect skew in the probe relation of a hash join
+  * the hashvalue is generated and stored before any tuples have been read 
+  * (see ExecHashJoinDetectSkew).
+  *
+  * Build tuples that hash to the same hashvalue are placed in the bucket while
+  * reading the build relation.
+  */
+ typedef struct HashJoinIMBucket
+ {
+ 	uint32 hashvalue;
+ 	HashJoinTuple tuples;
+ } HashJoinIMBucket;
+ 
+ #define IM_INVALID_BUCKET -1
+ #define IM_WORK_MEM_PERCENT 2
+ #define IM_BUCKET_OVERHEAD MAXALIGN(sizeof(HashJoinIMBucket))
  
  typedef struct HashJoinTableData
  {
***************
*** 113,121 ****
--- 132,161 ----
  
  	Size		spaceUsed;		/* memory space currently used by tuples */
  	Size		spaceAllowed;	/* upper limit for space used */
+ 	/* memory space currently used by IM buckets and tuples */
+ 	Size		spaceUsedIM;
+ 	/* upper limit for space used by IM buckets and tuples */
+ 	Size		spaceAllowedIM;
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	/* will the join optimize memory usage when probe relation is skewed */
+ 	bool enableSkewOptimization;
+ 	HashJoinIMBucket **imBucket; /* hashtable of IM buckets */
+ 	/*
+ 	 * array of imBucket indexes to the created IM buckets sorted
+ 	 * in the opposite order that they would be frozen to disk
+ 	 */
+ 	uint16 *imBucketFreezeOrder;
+ 	int nIMBuckets; /* # of buckets in the IM buckets hashtable */
+ 	/*
+ 	 * # of used buckets in the IM buckets hashtable and length of
+ 	 * imBucketFreezeOrder array
+ 	 */
+ 	int nUsedIMBuckets;
+ 	/* # of IM buckets that have already been frozen to disk */
+ 	int nIMBucketsFrozen;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
Index: src/include/executor/nodeHash.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHash.h,v
retrieving revision 1.46
diff -c -r1.46 nodeHash.h
*** src/include/executor/nodeHash.h	1 Jan 2009 17:23:59 -0000	1.46
--- src/include/executor/nodeHash.h	6 Jan 2009 23:29:18 -0000
***************
*** 45,48 ****
--- 45,50 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern int ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue);
+ 
  #endif   /* NODEHASH_H */
Index: src/include/nodes/execnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/execnodes.h,v
retrieving revision 1.199
diff -c -r1.199 execnodes.h
*** src/include/nodes/execnodes.h	1 Jan 2009 17:23:59 -0000	1.199
--- src/include/nodes/execnodes.h	6 Jan 2009 23:11:33 -0000
***************
*** 1381,1386 ****
--- 1381,1387 ----
   *		hj_NeedNewOuter			true if need new outer tuple on next call
   *		hj_MatchedOuter			true if found a join match for current outer
   *		hj_OuterNotEmpty		true if outer relation known not empty
+  *		hj_OuterTupleIMBucketNo	IM bucket# for the current outer tuple
   * ----------------
   */
  
***************
*** 1406,1411 ****
--- 1407,1413 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int			hj_OuterTupleIMBucketNo;
  } HashJoinState;
  
  
Index: src/include/nodes/primnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/primnodes.h,v
retrieving revision 1.145
diff -c -r1.145 primnodes.h
*** src/include/nodes/primnodes.h	1 Jan 2009 17:24:00 -0000	1.145
--- src/include/nodes/primnodes.h	5 Jan 2009 12:57:25 -0000
***************
*** 142,148 ****
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno, for debugging */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;
--- 142,148 ----
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;

*** a/src/backend/executor/nodeHash.c
--- b/src/backend/executor/nodeHash.c
***************
*** 53,58 **** ExecHash(HashState *node)
--- 53,222 ----
  	return NULL;
  }
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashGetIMBucket
+  *
+  *  	Returns the index of the in-memory bucket for this
+  *		hashvalue, or IM_INVALID_BUCKET if the hashvalue is not
+  *		associated with any unfrozen bucket (or if skew
+  *		optimization is not being used).
+  *  
+  *		It is possible for a tuple whose join attribute value is
+  *		not a MCV to hash to an in-memory bucket due to the limited
+  * 		number of hash values but it is unlikely and everything
+  *		continues to work even if it does happen. We would
+  *		accidentally cache some less optimal tuples in memory
+  *		but the join result would still be accurate.
+  *
+  *		hashtable->imBucket is an open addressing hashtable of
+  *		in-memory buckets (HashJoinIMBucket).
+  * ----------------------------------------------------------------
+  */
+ int 
+ ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue)
+ {
+ 	int bucket;
+ 
+ 	if (!hashtable->enableSkewOptimization)
+ 		return IM_INVALID_BUCKET;
+ 	
+ 	/* Modulo the hashvalue (using bitmask) to find the IM bucket. */
+ 	bucket = hashvalue & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * While we have not hit a hole in the hashtable and have not hit the 
+ 	 * actual bucket we have collided in the hashtable so try the next
+ 	 * bucket location.
+ 	 */
+ 	while (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 		bucket = (bucket + 1) & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * If the bucket exists and has been correctly determined return
+ 	 * the bucket index.
+ 	 */
+ 	if (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue == hashvalue)
+ 		return bucket;
+ 
+ 	/*
+ 	 * Must have run into an empty location or a frozen bucket which means the
+ 	 * tuple with this hashvalue is not to be handled as if it matches with an
+ 	 * in-memory bucket.
+ 	 */
+ 	return IM_INVALID_BUCKET;
+ }
+ 
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashFreezeNextIMBucket
+  *
+  *		Freeze the tuples of the next in-memory bucket by pushing
+  *		them into the main hashtable.  Buckets are frozen in order
+  *		so that the best tuples are kept in memory the longest.
+  * ----------------------------------------------------------------
+  */
+ static bool 
+ ExecHashFreezeNextIMBucket(HashJoinTable hashtable)
+ {
+ 	int bucketToFreeze;
+ 	int bucketno;
+ 	int batchno;
+ 	uint32 hashvalue;
+ 	HashJoinTuple hashTuple;
+ 	HashJoinTuple nextHashTuple;
+ 	HashJoinIMBucket *bucket;
+ 	MinimalTuple mintuple;
+ 
+ 	/* Calculate the imBucket index of the bucket to freeze. */
+ 	bucketToFreeze = hashtable->imBucketFreezeOrder
+ 		[hashtable->nUsedIMBuckets - 1 - hashtable->nIMBucketsFrozen];
+ 
+ 	/* Grab a pointer to the actual IM bucket. */
+ 	bucket = hashtable->imBucket[bucketToFreeze];
+ 	hashvalue = bucket->hashvalue;
+ 
+ 	/*
+ 	 * Grab a pointer to the first tuple in the soon to be frozen IM bucket.
+ 	 */
+ 	hashTuple = bucket->tuples;
+ 
+ 	/*
+ 	 * Calculate which bucket and batch the tuples belong to in the main
+ 	 * non-IM hashtable.
+ 	 */
+ 	ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+ 
+ 	/* until we have read all tuples from this bucket */
+ 	while (hashTuple != NULL)
+ 	{
+ 		/*
+ 		 * Some of this code is very similar to that of ExecHashTableInsert.
+ 		 * We do not call ExecHashTableInsert directly as
+ 		 * ExecHashTableInsert expects a TupleTableSlot and we already have
+ 		 * HashJoinTuples.
+ 		 */
+ 		mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 
+ 		/* Decide whether to put the tuple in the hash table or a temp file. */
+ 		if (batchno == hashtable->curbatch)
+ 		{
+ 			/* Put the tuple in hash table. */
+ 			nextHashTuple = hashTuple->next;
+ 			hashTuple->next = hashtable->buckets[bucketno];
+ 			hashtable->buckets[bucketno] = hashTuple;
+ 			hashTuple = nextHashTuple;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 		}
+ 		else
+ 		{
+ 			/* Put the tuples into a temp file for later batches. */
+ 			Assert(batchno > hashtable->curbatch);
+ 			ExecHashJoinSaveTuple(mintuple, hashvalue,
+ 								  &hashtable->innerBatchFile[batchno]);
+ 			/*
+ 			 * Some memory has been freed up. This must be done before we
+ 			 * pfree the hashTuple of we lose access to the tuple size.
+ 			 */
+ 			hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			nextHashTuple = hashTuple->next;
+ 			pfree(hashTuple);
+ 			hashTuple = nextHashTuple;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * Free the memory the bucket struct was using as it is not necessary
+ 	 * any more.  All code treats a frozen in-memory bucket the same as one
+ 	 * that did not exist; by setting the pointer to null the rest of the code
+ 	 * will function as if we had not created this bucket at all.
+ 	 */
+ 	pfree(bucket);
+ 	hashtable->imBucket[bucketToFreeze] = NULL;
+ 	hashtable->spaceUsed -= IM_BUCKET_OVERHEAD;
+ 	hashtable->spaceUsedIM -= IM_BUCKET_OVERHEAD;
+ 	hashtable->nIMBucketsFrozen++;
+ 
+ 	/*
+ 	 * All buckets have been frozen and deleted from memory so turn off
+ 	 * skew aware partitioning and remove the structs from memory as they are
+ 	 * just wasting space from now on.
+ 	 */
+ 	if (hashtable->nUsedIMBuckets == hashtable->nIMBucketsFrozen)
+ 	{
+ 		hashtable->enableSkewOptimization = false;
+ 		pfree(hashtable->imBucket);
+ 		pfree(hashtable->imBucketFreezeOrder);
+ 		hashtable->spaceUsed -= hashtable->spaceUsedIM;
+ 		hashtable->spaceUsedIM = 0;
+ 	}
+ 
+ 	return true;
+ }
+ 
  /* ----------------------------------------------------------------
   *		MultiExecHash
   *
***************
*** 69,74 **** MultiExecHash(HashState *node)
--- 233,240 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int bucketNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,106 **** MultiExecHash(HashState *node)
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
  	}
  
--- 265,306 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			bucketNumber = ExecHashGetIMBucket(hashtable, hashvalue);
! 
! 			/* handle tuples not destined for an in-memory bucket normally */
! 			if (bucketNumber == IM_INVALID_BUCKET)
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 			else
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				/* get the HashJoinTuple */
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple = (HashJoinTuple)
! 					MemoryContextAlloc(hashtable->batchCxt, hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple, mintuple->t_len);
! 
! 				/* Push the HashJoinTuple onto the front of the IM bucket. */
! 				hashTuple->next = hashtable->imBucket[bucketNumber]->tuples;
! 				hashtable->imBucket[bucketNumber]->tuples = hashTuple;
! 
! 				/*
! 				 * More memory is now in use so make sure we are not over
! 				 * spaceAllowedIM.
! 				 */
! 				hashtable->spaceUsed += hashTupleSize;
! 				hashtable->spaceUsedIM += hashTupleSize;
! 				while (hashtable->spaceUsedIM > hashtable->spaceAllowedIM
! 					&& ExecHashFreezeNextIMBucket(hashtable))
! 					;
! 				/* Guarantee we are not over the spaceAllowed. */
! 				if (hashtable->spaceUsed > hashtable->spaceAllowed)
! 					ExecHashIncreaseNumBatches(hashtable);
! 			}
! 			hashtable->totalTuples++;
  		}
  	}
  
***************
*** 269,274 **** ExecHashTableCreate(Hash *node, List *hashOperators)
--- 469,483 ----
  	hashtable->outerBatchFile = NULL;
  	hashtable->spaceUsed = 0;
  	hashtable->spaceAllowed = work_mem * 1024L;
+ 	/* Initialize skew optimization related hashtable variables. */
+ 	hashtable->spaceUsedIM = 0;
+ 	hashtable->spaceAllowedIM
+ 		= hashtable->spaceAllowed * IM_WORK_MEM_PERCENT / 100;
+ 	hashtable->enableSkewOptimization = false;
+ 	hashtable->nUsedIMBuckets = 0;
+ 	hashtable->nIMBuckets = 0;
+ 	hashtable->imBucket = NULL;
+ 	hashtable->nIMBucketsFrozen = 0;
  
  	/*
  	 * Get info about the hash functions to be used for each hash key. Also
***************
*** 815,825 **** ExecScanHashBucket(HashJoinState *hjstate,
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
  	 */
! 	if (hashTuple == NULL)
! 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
! 	else
  		hashTuple = hashTuple->next;
  
  	while (hashTuple != NULL)
  	{
--- 1024,1040 ----
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
+ 	 *
+ 	 * If the tuple hashed to an IM bucket then scan the IM bucket
+ 	 * otherwise scan the standard hashtable bucket.
  	 */
! 	if (hashTuple != NULL)
  		hashTuple = hashTuple->next;
+ 	else if (hjstate->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
+ 		hashTuple = hashtable->imBucket[hjstate->hj_OuterTupleIMBucketNo]
+ 			->tuples;
+ 	else
+ 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
  
  	while (hashTuple != NULL)
  	{
*** a/src/backend/executor/nodeHashjoin.c
--- b/src/backend/executor/nodeHashjoin.c
***************
*** 20,25 ****
--- 20,29 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 **** static TupleTableSlot *ExecHashJoinGetSavedTuple(HashJoinState *hjstate,
--- 38,227 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashJoinDetectSkew
+  *
+  *		If MCV statistics can be found for the join attribute of
+  *		this hashjoin then create a hash table of buckets. Each
+  *		bucket will correspond to an MCV hashvalue and will be
+  *		filled with inner relation tuples whose join attribute
+  *		hashes to the same value as that MCV.  If a join
+  *		attribute value is a MCV for the join attribute in the
+  *		outer (probe) relation, tuples with this value in the
+  *		inner (build) relation are more likely to join with outer
+  *		relation tuples and a benefit can be gained by keeping
+  *		them in memory while joining the first batch of tuples.
+  * ----------------------------------------------------------------
+  */
+ static void 
+ ExecHashJoinDetectSkew(EState *estate, HashJoinState *hjstate, int tupwidth)
+ {
+ 	HeapTupleData	*statsTuple;
+ 	FuncExprState	*clause;
+ 	ExprState		*argstate;
+ 	Var				*variable;
+ 	HashJoinTable	hashtable;
+ 	Datum			*values;
+ 	int				nvalues;
+ 	float4			*numbers;
+ 	int				nnumbers;
+ 	Oid				relid;
+ 	Oid				atttype;
+ 	int				i;
+ 	int				mcvsToUse;
+ 
+ 	/* Only use statistics if there is a single join attribute. */
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; /* Histojoin is not defined for more than one join key */
+ 	
+ 	hashtable = hjstate->hj_HashTable;
+ 	
+ 	/*
+ 	 * Estimate the number of IM buckets that will fit in
+ 	 * the memory allowed for IM buckets.
+ 	 *
+ 	 * hashtable->imBucket will have up to 8 times as many HashJoinIMBucket
+ 	 * pointers as the number of MCV hashvalues. A uint16 index in
+ 	 * hashtable->imBucketFreezeOrder will be created for each IM bucket. One
+ 	 * actual HashJoinIMBucket struct will be created for each
+ 	 * unique MCV hashvalue so up to one struct per MCV.
+ 	 *
+ 	 * It is also estimated that each IM bucket will have a single build
+ 	 * tuple stored in it after partitioning the build relation input.  This
+ 	 * estimate could be high if tuples are filtered out before this join but
+ 	 * in that case the extra memory is used by the regular hashjoin batch.
+ 	 * This estimate could be low if it is a many to many join but in that
+ 	 * case IM buckets will be frozen to free up memory as needed
+ 	 * during the inner relation partitioning phase.
+ 	 */
+ 	mcvsToUse = hashtable->spaceAllowedIM / (
+ 		/* size of a hash tuple */
+ 		HJTUPLE_OVERHEAD + MAXALIGN(sizeof(MinimalTupleData))
+ 			+ MAXALIGN(tupwidth)
+ 		/* max size of hashtable pointers per MCV */
+ 		+ (8 * sizeof(HashJoinIMBucket*))
+ 		+ sizeof(uint16) /* size of imBucketFreezeOrder entry */
+ 		+ IM_BUCKET_OVERHEAD /* size of IM bucket struct */
+ 		);
+ 	if (mcvsToUse == 0)
+ 		return;	/* No point in considering this any further. */
+ 
+ 	/*
+ 	 * Determine the relation id and attribute id of the single join
+ 	 * attribute of the probe relation.
+ 	 */
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 
+ 	/*
+ 	 * Do not try to exploit stats if the join attribute is an expression
+ 	 * instead of just a simple attribute.
+ 	 */		
+ 	if (argstate->expr->type != T_Var)
+ 		return;
+ 
+ 	variable = (Var *) argstate->expr;
+ 	relid = getrelid(variable->varnoold, estate->es_range_table);
+ 	atttype = variable->vartype;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT, ObjectIdGetDatum(relid),
+ 		Int16GetDatum(variable->varoattno), 0, 0);
+ 	if (!HeapTupleIsValid(statsTuple))
+ 		return;
+ 
+ 	/* Look for MCV statistics for the attribute. */
+ 	if (get_attstatsslot(statsTuple, atttype, variable->vartypmod,
+ 		STATISTIC_KIND_MCV, InvalidOid, &values, &nvalues,
+ 		&numbers, &nnumbers))
+ 	{
+ 		FmgrInfo   *hashfunctions;
+ 		int nbuckets = 2;
+ 
+ 		/*
+ 		 * IM buckets (imBucket) is an open addressing hashtable with a 
+ 		 * power of 2 size that is greater than the number of MCV values.
+ 		 */
+ 		if (mcvsToUse > nvalues)
+ 			mcvsToUse = nvalues;
+ 		while (nbuckets <= mcvsToUse)
+ 			nbuckets <<= 1;
+ 		/* use two more bits just to help avoid collisions */
+ 		nbuckets <<= 2;
+ 		hashtable->nIMBuckets = nbuckets;
+ 		hashtable->enableSkewOptimization = true;
+ 
+ 		/*
+ 		 * Allocate the bucket memory in the hashtable's batch context
+ 		 * because it is only relevant and necessary during the first batch
+ 		 * and will be nicely removed once the first batch is done.
+ 		 */
+ 		hashtable->imBucket = 
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				nbuckets * sizeof(HashJoinIMBucket*));
+ 		hashtable->imBucketFreezeOrder = 
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				mcvsToUse * sizeof(uint16));
+ 		/* Count the overhead of the IM pointers immediately. */
+ 		hashtable->spaceUsed += nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 		hashtable->spaceUsedIM +=  nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 
+ 		/*
+ 		 * Grab the hash functions as we will be generating the hashvalues
+ 		 * in this section.
+ 		 */
+ 		hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 		/* Create the buckets */
+ 		for (i = 0; i < mcvsToUse; i++)
+ 		{
+ 			uint32 hashvalue = DatumGetUInt32(
+ 				FunctionCall1(&hashfunctions[0], values[i]));
+ 			int bucket = hashvalue & (nbuckets - 1);
+ 
+ 			/*
+ 			 * While we have not hit a hole in the hashtable and have not hit
+ 			 * a bucket with the same hashvalue we have collided in the
+ 			 * hashtable so try the next bucket location (remember it is an
+ 			 * open addressing hashtable).
+ 			 */
+ 			while (hashtable->imBucket[bucket] != NULL
+ 				&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 				bucket = (bucket + 1) & (nbuckets - 1);
+ 
+ 			/*
+ 			 * Leave bucket alone if it has the same hashvalue as current
+ 			 * MCV. We only want one bucket per hashvalue. Even if two MCV
+ 			 * values hash to the same bucket we are fine.
+ 			 */
+ 			if (hashtable->imBucket[bucket] == NULL)
+ 			{
+ 				/*
+ 				 * Allocate the actual bucket structure in the hashtable's batch
+ 				 * context because it is only relevant and necessary during
+ 				 * the first batch and will be nicely removed once the first
+ 				 * batch is done.
+ 				 */
+ 				hashtable->imBucket[bucket]
+ 					= MemoryContextAllocZero(hashtable->batchCxt,
+ 						sizeof(HashJoinIMBucket));
+ 				hashtable->imBucket[bucket]->hashvalue = hashvalue;
+ 				hashtable->imBucketFreezeOrder[hashtable->nUsedIMBuckets]
+ 					= bucket;
+ 				hashtable->nUsedIMBuckets++;
+ 				/* Count the overhead of the IM bucket struct */
+ 				hashtable->spaceUsed += IM_BUCKET_OVERHEAD;
+ 				hashtable->spaceUsedIM += IM_BUCKET_OVERHEAD;
+ 			}
+ 		}
+ 
+ 		free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 	}
+ 
+ 	ReleaseSysCache(statsTuple);
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 147,152 **** ExecHashJoin(HashJoinState *node)
--- 335,345 ----
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
  
+ 		/* Use skew optimization only when there is more than one batch. */
+ 		if (hashtable->nbatch > 1)
+ 			ExecHashJoinDetectSkew(estate, node,
+ 				(outerPlan((Hash *) hashNode->ps.plan))->plan_width);
+ 
  		/*
  		 * execute the Hash node, to build the hash table
  		 */
***************
*** 205,216 **** ExecHashJoin(HashJoinState *node)
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
  			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
--- 398,415 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
+ 			
+ 			/* Does the outer tuple match an IM bucket? */
+ 			node->hj_OuterTupleIMBucketNo = 
+ 				ExecHashGetIMBucket(hashtable, hashvalue);
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but in might not belong to the current batch, or it might need
! 			 * to go into an in-memory bucket.
  			 */
! 			if (node->hj_OuterTupleIMBucketNo == IM_INVALID_BUCKET
! 				&& batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
***************
*** 641,647 **** start_over:
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
--- 840,866 ----
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	/* if we just finished the first batch */
! 	if (curbatch == 0)
! 	{
! 		/*
! 		 * Reset some of the skew optimization state variables. IM buckets are
! 		 * no longer being used as of this point because they are only
! 		 * necessary while joining the first batch (before the cleanup phase).
! 		 *
! 		 * Especially need to make sure ExecHashGetIMBucket returns
! 		 * IM_INVALID_BUCKET quickly for all subsequent calls.
! 		 *
! 		 * IM buckets are only taking up memory if this is a multi-batch join
! 		 * and in that case ExecHashTableReset is about to be called which
! 		 * will free all memory currently used by IM buckets and tuples when
! 		 * it deletes hashtable->batchCxt.  If this is a single batch join
! 		 * then imBucket and imBucketFreezeOrder are already NULL and empty.
! 		 */
! 		hashtable->enableSkewOptimization = false;
! 		hashtable->spaceUsedIM = 0;
! 	}
! 	else if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
*** a/src/include/executor/hashjoin.h
--- b/src/include/executor/hashjoin.h
***************
*** 72,77 **** typedef struct HashJoinTupleData
--- 72,96 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ /*
+  * Stores a hashvalue and linked list of tuples that share that hashvalue.
+  *
+  * When processing MCVs to detect skew in the probe relation of a hash join
+  * the hashvalue is generated and stored before any tuples have been read 
+  * (see ExecHashJoinDetectSkew).
+  *
+  * Build tuples that hash to the same hashvalue are placed in the bucket while
+  * reading the build relation.
+  */
+ typedef struct HashJoinIMBucket
+ {
+ 	uint32 hashvalue;
+ 	HashJoinTuple tuples;
+ } HashJoinIMBucket;
+ 
+ #define IM_INVALID_BUCKET -1
+ #define IM_WORK_MEM_PERCENT 2
+ #define IM_BUCKET_OVERHEAD MAXALIGN(sizeof(HashJoinIMBucket))
  
  typedef struct HashJoinTableData
  {
***************
*** 113,121 **** typedef struct HashJoinTableData
--- 132,161 ----
  
  	Size		spaceUsed;		/* memory space currently used by tuples */
  	Size		spaceAllowed;	/* upper limit for space used */
+ 	/* memory space currently used by IM buckets and tuples */
+ 	Size		spaceUsedIM;
+ 	/* upper limit for space used by IM buckets and tuples */
+ 	Size		spaceAllowedIM;
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	/* will the join optimize memory usage when probe relation is skewed */
+ 	bool enableSkewOptimization;
+ 	HashJoinIMBucket **imBucket; /* hashtable of IM buckets */
+ 	/*
+ 	 * array of imBucket indexes to the created IM buckets sorted
+ 	 * in the opposite order that they would be frozen to disk
+ 	 */
+ 	uint16 *imBucketFreezeOrder;
+ 	int nIMBuckets; /* # of buckets in the IM buckets hashtable */
+ 	/*
+ 	 * # of used buckets in the IM buckets hashtable and length of
+ 	 * imBucketFreezeOrder array
+ 	 */
+ 	int nUsedIMBuckets;
+ 	/* # of IM buckets that have already been frozen to disk */
+ 	int nIMBucketsFrozen;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
*** a/src/include/executor/nodeHash.h
--- b/src/include/executor/nodeHash.h
***************
*** 45,48 **** extern void ExecChooseHashTableSize(double ntuples, int tupwidth,
--- 45,50 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern int ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue);
+ 
  #endif   /* NODEHASH_H */
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 1381,1386 **** typedef struct MergeJoinState
--- 1381,1387 ----
   *		hj_NeedNewOuter			true if need new outer tuple on next call
   *		hj_MatchedOuter			true if found a join match for current outer
   *		hj_OuterNotEmpty		true if outer relation known not empty
+  *		hj_OuterTupleIMBucketNo	IM bucket# for the current outer tuple
   * ----------------
   */
  
***************
*** 1406,1411 **** typedef struct HashJoinState
--- 1407,1413 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int			hj_OuterTupleIMBucketNo;
  } HashJoinState;
  
  
*** a/src/include/nodes/primnodes.h
--- b/src/include/nodes/primnodes.h
***************
*** 121,128 **** typedef struct Expr
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  * The code doesn't really need varnoold/varoattno, but they are very useful
!  * for debugging and interpreting completed plans, so we keep them around.
   */
  #define    INNER		65000
  #define    OUTER		65001
--- 121,132 ----
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  *
!  * For the most part, the code doesn't really need varnoold/varoattno, but
!  * they are very useful for debugging and interpreting completed plans, so we
!  * keep them around.  As of PostgreSQL 8.4, these values are also used by
!  * ExecHashJoinDetectSkew to fetch MCV statistics when performing multi-batch
!  * hash joins.
   */
  #define    INNER		65000
  #define    OUTER		65001
***************
*** 142,148 **** typedef struct Var
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno, for debugging */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;
--- 146,152 ----
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;

Index: src/backend/executor/nodeHash.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHash.c,v
retrieving revision 1.117
diff -c -r1.117 nodeHash.c
*** src/backend/executor/nodeHash.c	1 Jan 2009 17:23:41 -0000	1.117
--- src/backend/executor/nodeHash.c	14 Jan 2009 06:36:51 -0000
***************
*** 53,58 ****
--- 53,224 ----
  	return NULL;
  }
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashGetIMBucket
+  *
+  *  	Returns the index of the in-memory bucket for this
+  *		hashvalue, or IM_INVALID_BUCKET if the hashvalue is not
+  *		associated with any unfrozen bucket (or if skew
+  *		optimization is not being used).
+  *  
+  *		It is possible for a tuple whose join attribute value is
+  *		not a MCV to hash to an in-memory bucket due to the limited
+  * 		number of hash values but it is unlikely and everything
+  *		continues to work even if it does happen. We would
+  *		accidentally cache some less optimal tuples in memory
+  *		but the join result would still be accurate.
+  *
+  *		hashtable->imBucket is an open addressing hashtable of
+  *		in-memory buckets (HashJoinIMBucket).
+  * ----------------------------------------------------------------
+  */
+ int 
+ ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue)
+ {
+ 	int bucket;
+ 
+ 	if (!hashtable->enableSkewOptimization)
+ 		return IM_INVALID_BUCKET;
+ 	
+ 	/* Modulo the hashvalue (using bitmask) to find the IM bucket. */
+ 	bucket = hashvalue & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * While we have not hit a hole in the hashtable and have not hit the 
+ 	 * actual bucket we have collided in the hashtable so try the next
+ 	 * bucket location.
+ 	 */
+ 	while (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 		bucket = (bucket + 1) & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * If the bucket exists and has been correctly determined return
+ 	 * the bucket index.
+ 	 */
+ 	if (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue == hashvalue)
+ 		return bucket;
+ 
+ 	/*
+ 	 * Must have run into an empty location or a frozen bucket which means the
+ 	 * tuple with this hashvalue is not to be handled as if it matches with an
+ 	 * in-memory bucket.
+ 	 */
+ 	return IM_INVALID_BUCKET;
+ }
+ 
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashFreezeNextIMBucket
+  *
+  *		Freeze the tuples of the next in-memory bucket by pushing
+  *		them into the main hashtable.  Buckets are frozen in order
+  *		so that the best tuples are kept in memory the longest.
+  * ----------------------------------------------------------------
+  */
+ static bool 
+ ExecHashFreezeNextIMBucket(HashJoinTable hashtable)
+ {
+ 	int bucketToFreeze;
+ 	int bucketno;
+ 	int batchno;
+ 	uint32 hashvalue;
+ 	HashJoinTuple hashTuple;
+ 	HashJoinTuple nextHashTuple;
+ 	HashJoinIMBucket *bucket;
+ 	MinimalTuple mintuple;
+ 
+ 	/* Calculate the imBucket index of the bucket to freeze. */
+ 	bucketToFreeze = hashtable->imBucketFreezeOrder
+ 		[hashtable->nUsedIMBuckets - 1 - hashtable->nIMBucketsFrozen];
+ 
+ 	/* Grab a pointer to the actual IM bucket. */
+ 	bucket = hashtable->imBucket[bucketToFreeze];
+ 	hashvalue = bucket->hashvalue;
+ 
+ 	/*
+ 	 * Grab a pointer to the first tuple in the soon to be frozen IM bucket.
+ 	 */
+ 	hashTuple = bucket->tuples;
+ 
+ 	/*
+ 	 * Calculate which bucket and batch the tuples belong to in the main
+ 	 * non-IM hashtable.
+ 	 */
+ 	ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+ 
+ 	/* until we have read all tuples from this bucket */
+ 	while (hashTuple != NULL)
+ 	{
+ 		/*
+ 		 * Some of this code is very similar to that of ExecHashTableInsert.
+ 		 * We do not call ExecHashTableInsert directly as
+ 		 * ExecHashTableInsert expects a TupleTableSlot and we already have
+ 		 * HashJoinTuples.
+ 		 */
+ 		mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 
+ 		hashtable->nInnerIMTupFrozen++;
+ 
+ 		/* Decide whether to put the tuple in the hash table or a temp file. */
+ 		if (batchno == hashtable->curbatch)
+ 		{
+ 			/* Put the tuple in hash table. */
+ 			nextHashTuple = hashTuple->next;
+ 			hashTuple->next = hashtable->buckets[bucketno];
+ 			hashtable->buckets[bucketno] = hashTuple;
+ 			hashTuple = nextHashTuple;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 		}
+ 		else
+ 		{
+ 			/* Put the tuples into a temp file for later batches. */
+ 			Assert(batchno > hashtable->curbatch);
+ 			ExecHashJoinSaveTuple(mintuple, hashvalue,
+ 								  &hashtable->innerBatchFile[batchno]);
+ 			/*
+ 			 * Some memory has been freed up. This must be done before we
+ 			 * pfree the hashTuple of we lose access to the tuple size.
+ 			 */
+ 			hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			nextHashTuple = hashTuple->next;
+ 			pfree(hashTuple);
+ 			hashTuple = nextHashTuple;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * Free the memory the bucket struct was using as it is not necessary
+ 	 * any more.  All code treats a frozen in-memory bucket the same as one
+ 	 * that did not exist; by setting the pointer to null the rest of the code
+ 	 * will function as if we had not created this bucket at all.
+ 	 */
+ 	pfree(bucket);
+ 	hashtable->imBucket[bucketToFreeze] = NULL;
+ 	hashtable->spaceUsed -= IM_BUCKET_OVERHEAD;
+ 	hashtable->spaceUsedIM -= IM_BUCKET_OVERHEAD;
+ 	hashtable->nIMBucketsFrozen++;
+ 
+ 	/*
+ 	 * All buckets have been frozen and deleted from memory so turn off
+ 	 * skew aware partitioning and remove the structs from memory as they are
+ 	 * just wasting space from now on.
+ 	 */
+ 	if (hashtable->nUsedIMBuckets == hashtable->nIMBucketsFrozen)
+ 	{
+ 		hashtable->enableSkewOptimization = false;
+ 		pfree(hashtable->imBucket);
+ 		pfree(hashtable->imBucketFreezeOrder);
+ 		hashtable->spaceUsed -= hashtable->spaceUsedIM;
+ 		hashtable->spaceUsedIM = 0;
+ 	}
+ 
+ 	return true;
+ }
+ 
  /* ----------------------------------------------------------------
   *		MultiExecHash
   *
***************
*** 69,74 ****
--- 235,242 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int bucketNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,106 ****
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
  	}
  
--- 267,322 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			int bucketno;
! 			int batchno;
! 
! 			bucketNumber = ExecHashGetIMBucket(hashtable, hashvalue);
! 
! 			ExecHashGetBucketAndBatch(hashtable, hashvalue,
! 									  &bucketno, &batchno);
! 			if (batchno == 0)
! 			{
! 				hashtable->nInnerPotentialBatchZeroTup++;
! 				if (bucketNumber == IM_INVALID_BUCKET)
! 					hashtable->nInnerBatchZeroTup++;
! 			}
! 
! 			/* handle tuples not destined for an in-memory bucket normally */
! 			if (bucketNumber == IM_INVALID_BUCKET)
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 			else
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				/* get the HashJoinTuple */
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple = (HashJoinTuple)
! 					MemoryContextAlloc(hashtable->batchCxt, hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple, mintuple->t_len);
! 
! 				/* Push the HashJoinTuple onto the front of the IM bucket. */
! 				hashTuple->next = hashtable->imBucket[bucketNumber]->tuples;
! 				hashtable->imBucket[bucketNumber]->tuples = hashTuple;
! 				
! 				/*
! 				 * More memory is now in use so make sure we are not over
! 				 * spaceAllowedIM.
! 				 */
! 				hashtable->spaceUsed += hashTupleSize;
! 				hashtable->spaceUsedIM += hashTupleSize;
! 				hashtable->nInnerIMTup++;
! 				while (hashtable->spaceUsedIM > hashtable->spaceAllowedIM
! 					&& ExecHashFreezeNextIMBucket(hashtable))
! 					;
! 				/* Guarantee we are not over the spaceAllowed. */
! 				if (hashtable->spaceUsed > hashtable->spaceAllowed)
! 					ExecHashIncreaseNumBatches(hashtable);
! 			}
! 			hashtable->totalTuples++;
! 			hashtable->nInnerTup++;
  		}
  	}
  
***************
*** 269,274 ****
--- 485,512 ----
  	hashtable->outerBatchFile = NULL;
  	hashtable->spaceUsed = 0;
  	hashtable->spaceAllowed = work_mem * 1024L;
+ 	/* Initialize skew optimization related hashtable variables. */
+ 	hashtable->spaceUsedIM = 0;
+ 	hashtable->spaceAllowedIM
+ 		= hashtable->spaceAllowed * IM_WORK_MEM_PERCENT / 100;
+ 	hashtable->enableSkewOptimization = false;
+ 	hashtable->nUsedIMBuckets = 0;
+ 	hashtable->nIMBuckets = 0;
+ 	hashtable->imBucket = NULL;
+ 	hashtable->nIMBucketsFrozen = 0;
+ 	hashtable->nInnerTup = 0;
+ 	hashtable->nOuterTup = 0;
+ 	hashtable->nInnerIMTup = 0;
+ 	hashtable->nOuterIMTup = 0;
+ 	hashtable->nOutputTup = 0;
+ 	hashtable->nOutputIMTup = 0;
+ 	hashtable->nInnerIMTupFrozen = 0;
+ 	hashtable->nOuterBatchZeroTup = 0;
+ 	hashtable->nOuterPotentialBatchZeroTup = 0;
+ 	hashtable->nOutputBatchZeroTup = 0;
+ 	hashtable->nOutputPotentialBatchZeroTup = 0;
+ 	hashtable->nInnerBatchZeroTup = 0;
+ 	hashtable->nInnerPotentialBatchZeroTup = 0;
  
  	/*
  	 * Get info about the hash functions to be used for each hash key. Also
***************
*** 448,453 ****
--- 686,754 ----
  {
  	int			i;
  
+ 	/* the total # of inner tuples received by join */
+ 	ereport(NOTICE, (errmsg("Total Inner Tuples: %d", hashtable->nInnerTup)));
+ 	/* # inner tuples in the IM buckets */
+ 	ereport(NOTICE, (errmsg("IM Inner Tuples: %d", hashtable->nInnerIMTup)));
+ 	/* # inner tuples that fell in batch 0 */
+ 	ereport(NOTICE, (errmsg("Batch Zero Inner Tuples: %d", 
+ 		hashtable->nInnerBatchZeroTup)));
+ 	/*
+ 	 * # inner tuples that would have fallen in batch 0 if IM buckets were
+ 	 * not in use at all
+ 	 */
+ 	ereport(NOTICE, (errmsg("Batch Zero Potential Inner Tuples: %d", 
+ 		hashtable->nInnerPotentialBatchZeroTup)));
+ 	/* total # of outer tuples received by join */
+ 	ereport(NOTICE, (errmsg("Total Outer Tuples: %d", hashtable->nOuterTup)));
+ 	/* # outer tuples that matched with the IM buckets */
+ 	ereport(NOTICE, (errmsg("IM Outer Tuples: %d", hashtable->nOuterIMTup)));
+ 	/* # outer tuples that matched with batch 0 */
+ 	ereport(NOTICE, (errmsg("Batch Zero Outer Tuples: %d", 
+ 		hashtable->nOuterBatchZeroTup)));
+ 	/*
+ 	 * # outer tuples that would have fallen in batch 0 if IM buckets were
+ 	 * not in use at all
+ 	 */
+ 	ereport(NOTICE, (errmsg("Batch Zero Potential Outer Tuples: %d", 
+ 		hashtable->nOuterPotentialBatchZeroTup)));
+ 	/* total # output tuples produced by join */
+ 	ereport(NOTICE, (errmsg("Total Output Tuples: %d", 
+ 		hashtable->nOutputTup)));
+ 	/* # output tuples that came from matches with IM bucket inner tuples */
+ 	ereport(NOTICE, (errmsg("IM Output Tuples: %d", 
+ 		hashtable->nOutputIMTup)));
+ 	/* # output tuples that came from matches with batch 0 inner tuples */
+ 	ereport(NOTICE, (errmsg("Batch Zero Output Tuples: %d", 
+ 		hashtable->nOutputBatchZeroTup)));
+ 	/*
+ 	 * # output tuples that would have came from matches with batch 0 if IM
+ 	 * buckets were not in use at all
+ 	 */
+ 	ereport(NOTICE, (errmsg("Batch Zero Potential Output Tuples: %d", 
+ 		hashtable->nOutputPotentialBatchZeroTup)));
+ 	/*
+ 	 * # of inner IM tuples that were frozen back to the main hashtable when
+ 	 * an IM bucket was frozen
+ 	 */
+ 	ereport(NOTICE, (errmsg("IM Tuples Frozen: %d", 
+ 		hashtable->nInnerIMTupFrozen)));
+ 	/* percentage less tuple IOs compared to HHJ due to skew optimization */
+ 	if (hashtable->nInnerTup - hashtable->nInnerPotentialBatchZeroTup
+ 			+ hashtable->nOuterTup - hashtable->nOuterPotentialBatchZeroTup
+ 			!= 0)
+ 		ereport(NOTICE, (errmsg("Percentage less tuple IOs than HHJ: %4.2f", 
+ 			(1 - (
+ 			(float)(hashtable->nInnerTup - hashtable->nInnerIMTup
+ 				- hashtable->nInnerBatchZeroTup 
+ 				+ hashtable->nOuterTup - hashtable->nOuterIMTup
+ 				- hashtable->nOuterBatchZeroTup)
+ 			/
+ 			(hashtable->nInnerTup - hashtable->nInnerPotentialBatchZeroTup
+ 				+ hashtable->nOuterTup
+ 				- hashtable->nOuterPotentialBatchZeroTup))
+ 			) * 100)));
+ 
  	/*
  	 * Make sure all the temp files are closed.  We skip batch 0, since it
  	 * can't have any temp files (and the arrays might not even exist if
***************
*** 815,825 ****
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
  	 */
! 	if (hashTuple == NULL)
! 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
! 	else
  		hashTuple = hashTuple->next;
  
  	while (hashTuple != NULL)
  	{
--- 1116,1132 ----
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
+ 	 *
+ 	 * If the tuple hashed to an IM bucket then scan the IM bucket
+ 	 * otherwise scan the standard hashtable bucket.
  	 */
! 	if (hashTuple != NULL)
  		hashTuple = hashTuple->next;
+ 	else if (hjstate->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
+ 		hashTuple = hashtable->imBucket[hjstate->hj_OuterTupleIMBucketNo]
+ 			->tuples;
+ 	else
+ 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
  
  	while (hashTuple != NULL)
  	{
Index: src/backend/executor/nodeHashjoin.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHashjoin.c,v
retrieving revision 1.97
diff -c -r1.97 nodeHashjoin.c
*** src/backend/executor/nodeHashjoin.c	1 Jan 2009 17:23:41 -0000	1.97
--- src/backend/executor/nodeHashjoin.c	13 Jan 2009 21:28:41 -0000
***************
*** 20,25 ****
--- 20,30 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
+ #include "optimizer/cost.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 ****
--- 39,246 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashJoinDetectSkew
+  *
+  *		If MCV statistics can be found for the join attribute of
+  *		this hashjoin then create a hash table of buckets. Each
+  *		bucket will correspond to an MCV hashvalue and will be
+  *		filled with inner relation tuples whose join attribute
+  *		hashes to the same value as that MCV.  If a join
+  *		attribute value is a MCV for the join attribute in the
+  *		outer (probe) relation, tuples with this value in the
+  *		inner (build) relation are more likely to join with outer
+  *		relation tuples and a benefit can be gained by keeping
+  *		them in memory while joining the first batch of tuples.
+  * ----------------------------------------------------------------
+  */
+ static void 
+ ExecHashJoinDetectSkew(EState *estate, HashJoinState *hjstate, int tupwidth)
+ {
+ 	HeapTupleData	*statsTuple;
+ 	FuncExprState	*clause;
+ 	ExprState		*argstate;
+ 	Var				*variable;
+ 	HashJoinTable	hashtable;
+ 	Datum			*values;
+ 	int				nvalues;
+ 	float4			*numbers;
+ 	int				nnumbers;
+ 	Oid				relid;
+ 	Oid				atttype;
+ 	int				i;
+ 	int				mcvsToUse;
+ 
+ 	PlanState  *outerNode = outerPlanState(hjstate);
+ 
+ 	/* Only use statistics if there is a single join attribute. */
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; /* Histojoin is not defined for more than one join key */
+ 	
+ 	hashtable = hjstate->hj_HashTable;
+ 	
+ 	/*
+ 	 * Estimate the number of IM buckets that will fit in
+ 	 * the memory allowed for IM buckets.
+ 	 *
+ 	 * hashtable->imBucket will have up to 8 times as many HashJoinIMBucket
+ 	 * pointers as the number of MCV hashvalues. A uint16 index in
+ 	 * hashtable->imBucketFreezeOrder will be created for each IM bucket. One
+ 	 * actual HashJoinIMBucket struct will be created for each
+ 	 * unique MCV hashvalue so up to one struct per MCV.
+ 	 *
+ 	 * It is also estimated that each IM bucket will have a single build
+ 	 * tuple stored in it after partitioning the build relation input.  This
+ 	 * estimate could be high if tuples are filtered out before this join but
+ 	 * in that case the extra memory is used by the regular hashjoin batch.
+ 	 * This estimate could be low if it is a many to many join but in that
+ 	 * case IM buckets will be frozen to free up memory as needed
+ 	 * during the inner relation partitioning phase.
+ 	 */
+ 	mcvsToUse = hashtable->spaceAllowedIM / (
+ 		/* size of a hash tuple */
+ 		HJTUPLE_OVERHEAD + MAXALIGN(sizeof(MinimalTupleData))
+ 			+ MAXALIGN(tupwidth)
+ 		/* max size of hashtable pointers per MCV */
+ 		+ (8 * sizeof(HashJoinIMBucket*))
+ 		+ sizeof(uint16) /* size of imBucketFreezeOrder entry */
+ 		+ IM_BUCKET_OVERHEAD /* size of IM bucket struct */
+ 		);
+ 	if (mcvsToUse == 0)
+ 		return;	/* No point in considering this any further. */
+ 
+ 	/*
+ 	 * Determine the relation id and attribute id of the single join
+ 	 * attribute of the probe relation.
+ 	 */
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 
+ 	/*
+ 	 * Do not try to exploit stats if the join attribute is an expression
+ 	 * instead of just a simple attribute.
+ 	 */		
+ 	if (argstate->expr->type != T_Var)
+ 		return;
+ 
+ 	variable = (Var *) argstate->expr;
+ 	relid = getrelid(variable->varnoold, estate->es_range_table);
+ 	atttype = variable->vartype;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT, ObjectIdGetDatum(relid),
+ 		Int16GetDatum(variable->varoattno), 0, 0);
+ 	if (!HeapTupleIsValid(statsTuple))
+ 		return;
+ 
+ 	/* Look for MCV statistics for the attribute. */
+ 	if (get_attstatsslot(statsTuple, atttype, variable->vartypmod,
+ 		STATISTIC_KIND_MCV, InvalidOid, &values, &nvalues,
+ 		&numbers, &nnumbers))
+ 	{
+ 		FmgrInfo   *hashfunctions;
+ 		int nbuckets = 2;
+ 		double frac = 0;
+ 		
+ 		/*
+ 		 * IM buckets (imBucket) is an open addressing hashtable with a 
+ 		 * power of 2 size that is greater than the number of MCV values.
+ 		 */
+ 		if (mcvsToUse > nvalues)
+ 			mcvsToUse = nvalues;
+ 			
+ 		for (i = 0; i < mcvsToUse; i++)
+ 			frac += numbers[i];
+ 
+ 		ereport(NOTICE, (errmsg("Values: %d Skew: %4.2f  Est. tuples: %4.2f Batches: %d  Est. Save: %4.2f",
+                         nvalues, frac, outerNode->plan->plan_rows, hashtable->nbatch, 
+ 			(frac*(1-1.0/hashtable->nbatch)*outerNode->plan->plan_rows))));
+ 		
+ 		if (frac < IM_MIN_BENEFIT_PERCENT)
+ 		{
+ 			free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 			ReleaseSysCache(statsTuple);
+ 			return;
+ 		}
+ 		
+ 		while (nbuckets <= mcvsToUse)
+ 			nbuckets <<= 1;
+ 		/* use two more bits just to help avoid collisions */
+ 		nbuckets <<= 2;
+ 		hashtable->nIMBuckets = nbuckets;
+ 		hashtable->enableSkewOptimization = true;
+ 
+ 		/*
+ 		 * Allocate the bucket memory in the hashtable's batch context
+ 		 * because it is only relevant and necessary during the first batch
+ 		 * and will be nicely removed once the first batch is done.
+ 		 */
+ 		hashtable->imBucket = 
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				nbuckets * sizeof(HashJoinIMBucket*));
+ 		hashtable->imBucketFreezeOrder = 
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				mcvsToUse * sizeof(uint16));
+ 		/* Count the overhead of the IM pointers immediately. */
+ 		hashtable->spaceUsed += nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 		hashtable->spaceUsedIM +=  nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 
+ 		/*
+ 		 * Grab the hash functions as we will be generating the hashvalues
+ 		 * in this section.
+ 		 */
+ 		hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 		/* Create the buckets */
+ 		for (i = 0; i < mcvsToUse; i++)
+ 		{
+ 			uint32 hashvalue = DatumGetUInt32(
+ 				FunctionCall1(&hashfunctions[0], values[i]));
+ 			int bucket = hashvalue & (nbuckets - 1);
+ 
+ 			/*
+ 			 * While we have not hit a hole in the hashtable and have not hit
+ 			 * a bucket with the same hashvalue we have collided in the
+ 			 * hashtable so try the next bucket location (remember it is an
+ 			 * open addressing hashtable).
+ 			 */
+ 			while (hashtable->imBucket[bucket] != NULL
+ 				&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 				bucket = (bucket + 1) & (nbuckets - 1);
+ 
+ 			/*
+ 			 * Leave bucket alone if it has the same hashvalue as current
+ 			 * MCV. We only want one bucket per hashvalue. Even if two MCV
+ 			 * values hash to the same bucket we are fine.
+ 			 */
+ 			if (hashtable->imBucket[bucket] == NULL)
+ 			{
+ 				/*
+ 				 * Allocate the actual bucket structure in the hashtable's batch
+ 				 * context because it is only relevant and necessary during
+ 				 * the first batch and will be nicely removed once the first
+ 				 * batch is done.
+ 				 */
+ 				hashtable->imBucket[bucket]
+ 					= MemoryContextAllocZero(hashtable->batchCxt,
+ 						sizeof(HashJoinIMBucket));
+ 				hashtable->imBucket[bucket]->hashvalue = hashvalue;
+ 				hashtable->imBucketFreezeOrder[hashtable->nUsedIMBuckets]
+ 					= bucket;
+ 				hashtable->nUsedIMBuckets++;
+ 				/* Count the overhead of the IM bucket struct */
+ 				hashtable->spaceUsed += IM_BUCKET_OVERHEAD;
+ 				hashtable->spaceUsedIM += IM_BUCKET_OVERHEAD;
+ 			}
+ 		}
+ 
+ 		free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 	}
+ 
+ 	ReleaseSysCache(statsTuple);
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 147,152 ****
--- 354,364 ----
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
  
+ 		/* Use skew optimization only when there is more than one batch. */
+ 		if (hashtable->nbatch > 1 && enable_hashjoin_usestatmcvs)
+ 			ExecHashJoinDetectSkew(estate, node,
+ 				(outerPlan((Hash *) hashNode->ps.plan))->plan_width);
+ 
  		/*
  		 * execute the Hash node, to build the hash table
  		 */
***************
*** 205,216 ****
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
  			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
--- 417,440 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
+ 			
+ 			/* Does the outer tuple match an IM bucket? */
+ 			node->hj_OuterTupleIMBucketNo = 
+ 				ExecHashGetIMBucket(hashtable, hashvalue);
+ 			if (node->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
+ 				hashtable->nOuterIMTup++;
+ 			else if (batchno == 0)
+ 				hashtable->nOuterBatchZeroTup++;
+ 			if (batchno == 0)
+ 				hashtable->nOuterPotentialBatchZeroTup++;
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but in might not belong to the current batch, or it might need
! 			 * to go into an in-memory bucket.
  			 */
! 			if (node->hj_OuterTupleIMBucketNo == IM_INVALID_BUCKET
! 				&& batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
***************
*** 281,286 ****
--- 505,517 ----
  					{
  						node->js.ps.ps_TupFromTlist =
  							(isDone == ExprMultipleResult);
+ 						hashtable->nOutputTup++;
+ 						if (node->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
+ 							hashtable->nOutputIMTup++;
+ 						else if (batchno == 0)
+ 							hashtable->nOutputBatchZeroTup++;
+ 						if (batchno == 0)
+ 							hashtable->nOutputPotentialBatchZeroTup++;
  						return result;
  					}
  				}
***************
*** 582,587 ****
--- 813,819 ----
  			{
  				/* remember outer relation is not empty for possible rescan */
  				hjstate->hj_OuterNotEmpty = true;
+ 				hashtable->nOuterTup++;
  
  				return slot;
  			}
***************
*** 641,647 ****
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
--- 873,899 ----
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	/* if we just finished the first batch */
! 	if (curbatch == 0)
! 	{
! 		/*
! 		 * Reset some of the skew optimization state variables. IM buckets are
! 		 * no longer being used as of this point because they are only
! 		 * necessary while joining the first batch (before the cleanup phase).
! 		 *
! 		 * Especially need to make sure ExecHashGetIMBucket returns
! 		 * IM_INVALID_BUCKET quickly for all subsequent calls.
! 		 *
! 		 * IM buckets are only taking up memory if this is a multi-batch join
! 		 * and in that case ExecHashTableReset is about to be called which
! 		 * will free all memory currently used by IM buckets and tuples when
! 		 * it deletes hashtable->batchCxt.  If this is a single batch join
! 		 * then imBucket and imBucketFreezeOrder are already NULL and empty.
! 		 */
! 		hashtable->enableSkewOptimization = false;
! 		hashtable->spaceUsedIM = 0;
! 	}
! 	else if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
Index: src/backend/optimizer/path/costsize.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/optimizer/path/costsize.c,v
retrieving revision 1.203
diff -c -r1.203 costsize.c
*** src/backend/optimizer/path/costsize.c	1 Jan 2009 17:23:43 -0000	1.203
--- src/backend/optimizer/path/costsize.c	13 Jan 2009 08:02:42 -0000
***************
*** 108,113 ****
--- 108,114 ----
  bool		enable_nestloop = true;
  bool		enable_mergejoin = true;
  bool		enable_hashjoin = true;
+ bool		enable_hashjoin_usestatmcvs = true;
  
  typedef struct
  {
Index: src/backend/utils/misc/guc.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/utils/misc/guc.c,v
retrieving revision 1.493
diff -c -r1.493 guc.c
*** src/backend/utils/misc/guc.c	12 Jan 2009 05:10:44 -0000	1.493
--- src/backend/utils/misc/guc.c	13 Jan 2009 08:03:33 -0000
***************
*** 656,661 ****
--- 656,669 ----
  		true, NULL, NULL
  	},
  	{
+ 		{"enable_hashjoin_usestatmcvs", PGC_USERSET, QUERY_TUNING_METHOD,
+ 			gettext_noop("Enables the hash join's use of the MCVs stored in pg_statistic."),
+ 			NULL
+ 		},
+ 		&enable_hashjoin_usestatmcvs,
+ 		true, NULL, NULL
+ 	},
+ 	{
  		{"geqo", PGC_USERSET, QUERY_TUNING_GEQO,
  			gettext_noop("Enables genetic query optimization."),
  			gettext_noop("This algorithm attempts to do planning without "
Index: src/include/executor/hashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/hashjoin.h,v
retrieving revision 1.49
diff -c -r1.49 hashjoin.h
*** src/include/executor/hashjoin.h	1 Jan 2009 17:23:59 -0000	1.49
--- src/include/executor/hashjoin.h	14 Jan 2009 06:42:53 -0000
***************
*** 72,77 ****
--- 72,97 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ /*
+  * Stores a hashvalue and linked list of tuples that share that hashvalue.
+  *
+  * When processing MCVs to detect skew in the probe relation of a hash join
+  * the hashvalue is generated and stored before any tuples have been read 
+  * (see ExecHashJoinDetectSkew).
+  *
+  * Build tuples that hash to the same hashvalue are placed in the bucket while
+  * reading the build relation.
+  */
+ typedef struct HashJoinIMBucket
+ {
+ 	uint32 hashvalue;
+ 	HashJoinTuple tuples;
+ } HashJoinIMBucket;
+ 
+ #define IM_INVALID_BUCKET -1
+ #define IM_WORK_MEM_PERCENT 2
+ #define IM_MIN_BENEFIT_PERCENT .01
+ #define IM_BUCKET_OVERHEAD MAXALIGN(sizeof(HashJoinIMBucket))
  
  typedef struct HashJoinTableData
  {
***************
*** 113,121 ****
--- 133,201 ----
  
  	Size		spaceUsed;		/* memory space currently used by tuples */
  	Size		spaceAllowed;	/* upper limit for space used */
+ 	/* memory space currently used by IM buckets and tuples */
+ 	Size		spaceUsedIM;
+ 	/* upper limit for space used by IM buckets and tuples */
+ 	Size		spaceAllowedIM;
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	/* will the join optimize memory usage when probe relation is skewed */
+ 	bool enableSkewOptimization;
+ 	HashJoinIMBucket **imBucket; /* hashtable of IM buckets */
+ 	/*
+ 	 * array of imBucket indexes to the created IM buckets sorted
+ 	 * in the opposite order that they would be frozen to disk
+ 	 */
+ 	uint16 *imBucketFreezeOrder;
+ 	int nIMBuckets; /* # of buckets in the IM buckets hashtable */
+ 	/*
+ 	 * # of used buckets in the IM buckets hashtable and length of
+ 	 * imBucketFreezeOrder array
+ 	 */
+ 	int nUsedIMBuckets;
+ 	/* # of IM buckets that have already been frozen to disk */
+ 	int nIMBucketsFrozen;
+ 
+ 	/* the total # of inner tuples received by join */
+ 	uint32 nInnerTup;
+ 	/* total # of outer tuples received by join */
+ 	uint32 nOuterTup;
+ 	/* # inner tuples in the IM buckets */
+ 	uint32 nInnerIMTup;
+ 	/* # outer tuples that matched with the IM buckets */
+ 	uint32 nOuterIMTup;
+ 	/* total # output tuples produced by join */
+ 	uint32 nOutputTup;
+ 	/* # output tuples that came from matches with IM bucket inner tuples */
+ 	uint32 nOutputIMTup;
+ 	/*
+ 	 * # of inner IM tuples that were frozen back to the main hashtable when
+ 	 * an IM bucket was frozen
+ 	 */
+ 	uint32 nInnerIMTupFrozen;
+ 	/* # outer tuples that matched with batch 0 */
+ 	uint32 nOuterBatchZeroTup;
+ 	/*
+ 	 * # outer tuples that would have fallen in batch 0 if IM buckets were
+ 	 * not in use at all
+ 	 */
+ 	uint32 nOuterPotentialBatchZeroTup;
+ 	/* # output tuples that came from matches with batch 0 inner tuples */
+ 	uint32 nOutputBatchZeroTup;
+ 	/*
+ 	 * # output tuples that would have came from matches with batch 0 if IM
+ 	 * buckets were not in use at all
+ 	 */
+ 	uint32 nOutputPotentialBatchZeroTup;
+ 	/* # inner tuples that fell in batch 0 */
+ 	uint32 nInnerBatchZeroTup;
+ 	/*
+ 	 * # inner tuples that would have fallen in batch 0 if IM buckets were
+ 	 * not in use at all
+ 	 */
+ 	uint32 nInnerPotentialBatchZeroTup;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
Index: src/include/executor/nodeHash.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHash.h,v
retrieving revision 1.46
diff -c -r1.46 nodeHash.h
*** src/include/executor/nodeHash.h	1 Jan 2009 17:23:59 -0000	1.46
--- src/include/executor/nodeHash.h	6 Jan 2009 23:29:18 -0000
***************
*** 45,48 ****
--- 45,50 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern int ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue);
+ 
  #endif   /* NODEHASH_H */
Index: src/include/nodes/execnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/execnodes.h,v
retrieving revision 1.201
diff -c -r1.201 execnodes.h
*** src/include/nodes/execnodes.h	12 Jan 2009 05:10:45 -0000	1.201
--- src/include/nodes/execnodes.h	12 Jan 2009 20:29:58 -0000
***************
*** 1389,1394 ****
--- 1389,1395 ----
   *		hj_NeedNewOuter			true if need new outer tuple on next call
   *		hj_MatchedOuter			true if found a join match for current outer
   *		hj_OuterNotEmpty		true if outer relation known not empty
+  *		hj_OuterTupleIMBucketNo	IM bucket# for the current outer tuple
   * ----------------
   */
  
***************
*** 1414,1419 ****
--- 1415,1421 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int			hj_OuterTupleIMBucketNo;
  } HashJoinState;
  
  
Index: src/include/nodes/primnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/primnodes.h,v
retrieving revision 1.145
diff -c -r1.145 primnodes.h
*** src/include/nodes/primnodes.h	1 Jan 2009 17:24:00 -0000	1.145
--- src/include/nodes/primnodes.h	7 Jan 2009 05:48:16 -0000
***************
*** 121,128 ****
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  * The code doesn't really need varnoold/varoattno, but they are very useful
!  * for debugging and interpreting completed plans, so we keep them around.
   */
  #define    INNER		65000
  #define    OUTER		65001
--- 121,132 ----
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  *
!  * For the most part, the code doesn't really need varnoold/varoattno, but
!  * they are very useful for debugging and interpreting completed plans, so we
!  * keep them around.  As of PostgreSQL 8.4, these values are also used by
!  * ExecHashJoinDetectSkew to fetch MCV statistics when performing multi-batch
!  * hash joins.
   */
  #define    INNER		65000
  #define    OUTER		65001
***************
*** 142,148 ****
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno, for debugging */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;
--- 146,152 ----
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;
Index: src/include/optimizer/cost.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/optimizer/cost.h,v
retrieving revision 1.96
diff -c -r1.96 cost.h
*** src/include/optimizer/cost.h	7 Jan 2009 22:40:49 -0000	1.96
--- src/include/optimizer/cost.h	13 Jan 2009 08:02:06 -0000
***************
*** 59,64 ****
--- 59,65 ----
  extern bool enable_nestloop;
  extern bool enable_mergejoin;
  extern bool enable_hashjoin;
+ extern bool enable_hashjoin_usestatmcvs;
  extern int	constraint_exclusion;
  
  extern double clamp_row_est(double nrows);

*** a/src/backend/executor/nodeHash.c
--- b/src/backend/executor/nodeHash.c
***************
*** 53,58 **** ExecHash(HashState *node)
--- 53,222 ----
  	return NULL;
  }
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashGetIMBucket
+  *
+  *  	Returns the index of the in-memory bucket for this
+  *		hashvalue, or IM_INVALID_BUCKET if the hashvalue is not
+  *		associated with any unfrozen bucket (or if skew
+  *		optimization is not being used).
+  *
+  *		It is possible for a tuple whose join attribute value is
+  *		not a MCV to hash to an in-memory bucket due to the limited
+  * 		number of hash values but it is unlikely and everything
+  *		continues to work even if it does happen. We would
+  *		accidentally cache some less optimal tuples in memory
+  *		but the join result would still be accurate.
+  *
+  *		hashtable->imBucket is an open addressing hashtable of
+  *		in-memory buckets (HashJoinIMBucket).
+  * ----------------------------------------------------------------
+  */
+ int
+ ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue)
+ {
+ 	int bucket;
+ 
+ 	if (!hashtable->enableSkewOptimization)
+ 		return IM_INVALID_BUCKET;
+ 	
+ 	/* Modulo the hashvalue (using bitmask) to find the IM bucket. */
+ 	bucket = hashvalue & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * While we have not hit a hole in the hashtable and have not hit the
+ 	 * actual bucket we have collided in the hashtable so try the next
+ 	 * bucket location.
+ 	 */
+ 	while (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 		bucket = (bucket + 1) & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * If the bucket exists and has been correctly determined return
+ 	 * the bucket index.
+ 	 */
+ 	if (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue == hashvalue)
+ 		return bucket;
+ 
+ 	/*
+ 	 * Must have run into an empty location or a frozen bucket which means the
+ 	 * tuple with this hashvalue is not to be handled as if it matches with an
+ 	 * in-memory bucket.
+ 	 */
+ 	return IM_INVALID_BUCKET;
+ }
+ 
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashFreezeNextIMBucket
+  *
+  *		Freeze the tuples of the next in-memory bucket by pushing
+  *		them into the main hashtable.  Buckets are frozen in order
+  *		so that the best tuples are kept in memory the longest.
+  * ----------------------------------------------------------------
+  */
+ static bool
+ ExecHashFreezeNextIMBucket(HashJoinTable hashtable)
+ {
+ 	int bucketToFreeze;
+ 	int bucketno;
+ 	int batchno;
+ 	uint32 hashvalue;
+ 	HashJoinTuple hashTuple;
+ 	HashJoinTuple nextHashTuple;
+ 	HashJoinIMBucket *bucket;
+ 	MinimalTuple mintuple;
+ 
+ 	/* Calculate the imBucket index of the bucket to freeze. */
+ 	bucketToFreeze = hashtable->imBucketFreezeOrder
+ 		[hashtable->nUsedIMBuckets - 1 - hashtable->nIMBucketsFrozen];
+ 
+ 	/* Grab a pointer to the actual IM bucket. */
+ 	bucket = hashtable->imBucket[bucketToFreeze];
+ 	hashvalue = bucket->hashvalue;
+ 
+ 	/*
+ 	 * Grab a pointer to the first tuple in the soon to be frozen IM bucket.
+ 	 */
+ 	hashTuple = bucket->tuples;
+ 
+ 	/*
+ 	 * Calculate which bucket and batch the tuples belong to in the main
+ 	 * non-IM hashtable.
+ 	 */
+ 	ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+ 
+ 	/* until we have read all tuples from this bucket */
+ 	while (hashTuple != NULL)
+ 	{
+ 		/*
+ 		 * Some of this code is very similar to that of ExecHashTableInsert.
+ 		 * We do not call ExecHashTableInsert directly as
+ 		 * ExecHashTableInsert expects a TupleTableSlot and we already have
+ 		 * HashJoinTuples.
+ 		 */
+ 		mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 
+ 		/* Decide whether to put the tuple in the hash table or a temp file. */
+ 		if (batchno == hashtable->curbatch)
+ 		{
+ 			/* Put the tuple in hash table. */
+ 			nextHashTuple = hashTuple->next;
+ 			hashTuple->next = hashtable->buckets[bucketno];
+ 			hashtable->buckets[bucketno] = hashTuple;
+ 			hashTuple = nextHashTuple;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 		}
+ 		else
+ 		{
+ 			/* Put the tuples into a temp file for later batches. */
+ 			Assert(batchno > hashtable->curbatch);
+ 			ExecHashJoinSaveTuple(mintuple, hashvalue,
+ 								  &hashtable->innerBatchFile[batchno]);
+ 			/*
+ 			 * Some memory has been freed up. This must be done before we
+ 			 * pfree the hashTuple of we lose access to the tuple size.
+ 			 */
+ 			hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			nextHashTuple = hashTuple->next;
+ 			pfree(hashTuple);
+ 			hashTuple = nextHashTuple;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * Free the memory the bucket struct was using as it is not necessary
+ 	 * any more.  All code treats a frozen in-memory bucket the same as one
+ 	 * that did not exist; by setting the pointer to null the rest of the code
+ 	 * will function as if we had not created this bucket at all.
+ 	 */
+ 	pfree(bucket);
+ 	hashtable->imBucket[bucketToFreeze] = NULL;
+ 	hashtable->spaceUsed -= IM_BUCKET_OVERHEAD;
+ 	hashtable->spaceUsedIM -= IM_BUCKET_OVERHEAD;
+ 	hashtable->nIMBucketsFrozen++;
+ 
+ 	/*
+ 	 * All buckets have been frozen and deleted from memory so turn off
+ 	 * skew aware partitioning and remove the structs from memory as they are
+ 	 * just wasting space from now on.
+ 	 */
+ 	if (hashtable->nUsedIMBuckets == hashtable->nIMBucketsFrozen)
+ 	{
+ 		hashtable->enableSkewOptimization = false;
+ 		pfree(hashtable->imBucket);
+ 		pfree(hashtable->imBucketFreezeOrder);
+ 		hashtable->spaceUsed -= hashtable->spaceUsedIM;
+ 		hashtable->spaceUsedIM = 0;
+ 	}
+ 
+ 	return true;
+ }
+ 
  /* ----------------------------------------------------------------
   *		MultiExecHash
   *
***************
*** 69,74 **** MultiExecHash(HashState *node)
--- 233,240 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int bucketNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 99,106 **** MultiExecHash(HashState *node)
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
  	}
  
--- 265,306 ----
  		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
  								 &hashvalue))
  		{
! 			bucketNumber = ExecHashGetIMBucket(hashtable, hashvalue);
! 
! 			/* handle tuples not destined for an in-memory bucket normally */
! 			if (bucketNumber == IM_INVALID_BUCKET)
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 			else
! 			{
! 				HashJoinTuple hashTuple;
! 				int			hashTupleSize;
! 				
! 				/* get the HashJoinTuple */
! 				mintuple = ExecFetchSlotMinimalTuple(slot);
! 				hashTupleSize = HJTUPLE_OVERHEAD + mintuple->t_len;
! 				hashTuple = (HashJoinTuple)
! 					MemoryContextAlloc(hashtable->batchCxt, hashTupleSize);
! 				hashTuple->hashvalue = hashvalue;
! 				memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple, mintuple->t_len);
! 
! 				/* Push the HashJoinTuple onto the front of the IM bucket. */
! 				hashTuple->next = hashtable->imBucket[bucketNumber]->tuples;
! 				hashtable->imBucket[bucketNumber]->tuples = hashTuple;
! 
! 				/*
! 				 * More memory is now in use so make sure we are not over
! 				 * spaceAllowedIM.
! 				 */
! 				hashtable->spaceUsed += hashTupleSize;
! 				hashtable->spaceUsedIM += hashTupleSize;
! 				while (hashtable->spaceUsedIM > hashtable->spaceAllowedIM
! 					&& ExecHashFreezeNextIMBucket(hashtable))
! 					;
! 				/* Guarantee we are not over the spaceAllowed. */
! 				if (hashtable->spaceUsed > hashtable->spaceAllowed)
! 					ExecHashIncreaseNumBatches(hashtable);
! 			}
! 			hashtable->totalTuples++;
  		}
  	}
  
***************
*** 269,274 **** ExecHashTableCreate(Hash *node, List *hashOperators)
--- 469,483 ----
  	hashtable->outerBatchFile = NULL;
  	hashtable->spaceUsed = 0;
  	hashtable->spaceAllowed = work_mem * 1024L;
+ 	/* Initialize skew optimization related hashtable variables. */
+ 	hashtable->spaceUsedIM = 0;
+ 	hashtable->spaceAllowedIM
+ 		= hashtable->spaceAllowed * IM_WORK_MEM_PERCENT / 100;
+ 	hashtable->enableSkewOptimization = false;
+ 	hashtable->nUsedIMBuckets = 0;
+ 	hashtable->nIMBuckets = 0;
+ 	hashtable->imBucket = NULL;
+ 	hashtable->nIMBucketsFrozen = 0;
  
  	/*
  	 * Get info about the hash functions to be used for each hash key. Also
***************
*** 815,825 **** ExecScanHashBucket(HashJoinState *hjstate,
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
  	 */
! 	if (hashTuple == NULL)
! 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
! 	else
  		hashTuple = hashTuple->next;
  
  	while (hashTuple != NULL)
  	{
--- 1024,1040 ----
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
+ 	 *
+ 	 * If the tuple hashed to an IM bucket then scan the IM bucket
+ 	 * otherwise scan the standard hashtable bucket.
  	 */
! 	if (hashTuple != NULL)
  		hashTuple = hashTuple->next;
+ 	else if (hjstate->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
+ 		hashTuple = hashtable->imBucket[hjstate->hj_OuterTupleIMBucketNo]
+ 			->tuples;
+ 	else
+ 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
  
  	while (hashTuple != NULL)
  	{
*** a/src/backend/executor/nodeHashjoin.c
--- b/src/backend/executor/nodeHashjoin.c
***************
*** 20,25 ****
--- 20,29 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 **** static TupleTableSlot *ExecHashJoinGetSavedTuple(HashJoinState *hjstate,
--- 38,227 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashJoinDetectSkew
+  *
+  *		If MCV statistics can be found for the join attribute of
+  *		this hashjoin then create a hash table of buckets. Each
+  *		bucket will correspond to an MCV hashvalue and will be
+  *		filled with inner relation tuples whose join attribute
+  *		hashes to the same value as that MCV.  If a join
+  *		attribute value is a MCV for the join attribute in the
+  *		outer (probe) relation, tuples with this value in the
+  *		inner (build) relation are more likely to join with outer
+  *		relation tuples and a benefit can be gained by keeping
+  *		them in memory while joining the first batch of tuples.
+  * ----------------------------------------------------------------
+  */
+ static void
+ ExecHashJoinDetectSkew(EState *estate, HashJoinState *hjstate, int tupwidth)
+ {
+ 	HeapTupleData	*statsTuple;
+ 	FuncExprState	*clause;
+ 	ExprState		*argstate;
+ 	Var				*variable;
+ 	HashJoinTable	hashtable;
+ 	Datum			*values;
+ 	int				nvalues;
+ 	float4			*numbers;
+ 	int				nnumbers;
+ 	Oid				relid;
+ 	Oid				atttype;
+ 	int				i;
+ 	int				mcvsToUse;
+ 
+ 	/* Only use statistics if there is a single join attribute. */
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; /* Histojoin is not defined for more than one join key */
+ 	
+ 	hashtable = hjstate->hj_HashTable;
+ 	
+ 	/*
+ 	 * Estimate the number of IM buckets that will fit in
+ 	 * the memory allowed for IM buckets.
+ 	 *
+ 	 * hashtable->imBucket will have up to 8 times as many HashJoinIMBucket
+ 	 * pointers as the number of MCV hashvalues. A uint16 index in
+ 	 * hashtable->imBucketFreezeOrder will be created for each IM bucket. One
+ 	 * actual HashJoinIMBucket struct will be created for each
+ 	 * unique MCV hashvalue so up to one struct per MCV.
+ 	 *
+ 	 * It is also estimated that each IM bucket will have a single build
+ 	 * tuple stored in it after partitioning the build relation input.  This
+ 	 * estimate could be high if tuples are filtered out before this join but
+ 	 * in that case the extra memory is used by the regular hashjoin batch.
+ 	 * This estimate could be low if it is a many to many join but in that
+ 	 * case IM buckets will be frozen to free up memory as needed
+ 	 * during the inner relation partitioning phase.
+ 	 */
+ 	mcvsToUse = hashtable->spaceAllowedIM / (
+ 		/* size of a hash tuple */
+ 		HJTUPLE_OVERHEAD + MAXALIGN(sizeof(MinimalTupleData))
+ 			+ MAXALIGN(tupwidth)
+ 		/* max size of hashtable pointers per MCV */
+ 		+ (8 * sizeof(HashJoinIMBucket*))
+ 		+ sizeof(uint16) /* size of imBucketFreezeOrder entry */
+ 		+ IM_BUCKET_OVERHEAD /* size of IM bucket struct */
+ 		);
+ 	if (mcvsToUse == 0)
+ 		return;	/* No point in considering this any further. */
+ 
+ 	/*
+ 	 * Determine the relation id and attribute id of the single join
+ 	 * attribute of the probe relation.
+ 	 */
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 
+ 	/*
+ 	 * Do not try to exploit stats if the join attribute is an expression
+ 	 * instead of just a simple attribute.
+ 	 */		
+ 	if (argstate->expr->type != T_Var)
+ 		return;
+ 
+ 	variable = (Var *) argstate->expr;
+ 	relid = getrelid(variable->varnoold, estate->es_range_table);
+ 	atttype = variable->vartype;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT, ObjectIdGetDatum(relid),
+ 		Int16GetDatum(variable->varoattno), 0, 0);
+ 	if (!HeapTupleIsValid(statsTuple))
+ 		return;
+ 
+ 	/* Look for MCV statistics for the attribute. */
+ 	if (get_attstatsslot(statsTuple, atttype, variable->vartypmod,
+ 		STATISTIC_KIND_MCV, InvalidOid, &values, &nvalues,
+ 		&numbers, &nnumbers))
+ 	{
+ 		FmgrInfo   *hashfunctions;
+ 		int nbuckets = 2;
+ 
+ 		/*
+ 		 * IM buckets (imBucket) is an open addressing hashtable with a
+ 		 * power of 2 size that is greater than the number of MCV values.
+ 		 */
+ 		if (mcvsToUse > nvalues)
+ 			mcvsToUse = nvalues;
+ 		while (nbuckets <= mcvsToUse)
+ 			nbuckets <<= 1;
+ 		/* use two more bits just to help avoid collisions */
+ 		nbuckets <<= 2;
+ 		hashtable->nIMBuckets = nbuckets;
+ 		hashtable->enableSkewOptimization = true;
+ 
+ 		/*
+ 		 * Allocate the bucket memory in the hashtable's batch context
+ 		 * because it is only relevant and necessary during the first batch
+ 		 * and will be nicely removed once the first batch is done.
+ 		 */
+ 		hashtable->imBucket =
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				nbuckets * sizeof(HashJoinIMBucket*));
+ 		hashtable->imBucketFreezeOrder =
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				mcvsToUse * sizeof(uint16));
+ 		/* Count the overhead of the IM pointers immediately. */
+ 		hashtable->spaceUsed += nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 		hashtable->spaceUsedIM +=  nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 
+ 		/*
+ 		 * Grab the hash functions as we will be generating the hashvalues
+ 		 * in this section.
+ 		 */
+ 		hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 		/* Create the buckets */
+ 		for (i = 0; i < mcvsToUse; i++)
+ 		{
+ 			uint32 hashvalue = DatumGetUInt32(
+ 				FunctionCall1(&hashfunctions[0], values[i]));
+ 			int bucket = hashvalue & (nbuckets - 1);
+ 
+ 			/*
+ 			 * While we have not hit a hole in the hashtable and have not hit
+ 			 * a bucket with the same hashvalue we have collided in the
+ 			 * hashtable so try the next bucket location (remember it is an
+ 			 * open addressing hashtable).
+ 			 */
+ 			while (hashtable->imBucket[bucket] != NULL
+ 				&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 				bucket = (bucket + 1) & (nbuckets - 1);
+ 
+ 			/*
+ 			 * Leave bucket alone if it has the same hashvalue as current
+ 			 * MCV. We only want one bucket per hashvalue. Even if two MCV
+ 			 * values hash to the same bucket we are fine.
+ 			 */
+ 			if (hashtable->imBucket[bucket] == NULL)
+ 			{
+ 				/*
+ 				 * Allocate the actual bucket structure in the hashtable's batch
+ 				 * context because it is only relevant and necessary during
+ 				 * the first batch and will be nicely removed once the first
+ 				 * batch is done.
+ 				 */
+ 				hashtable->imBucket[bucket]
+ 					= MemoryContextAllocZero(hashtable->batchCxt,
+ 						sizeof(HashJoinIMBucket));
+ 				hashtable->imBucket[bucket]->hashvalue = hashvalue;
+ 				hashtable->imBucketFreezeOrder[hashtable->nUsedIMBuckets]
+ 					= bucket;
+ 				hashtable->nUsedIMBuckets++;
+ 				/* Count the overhead of the IM bucket struct */
+ 				hashtable->spaceUsed += IM_BUCKET_OVERHEAD;
+ 				hashtable->spaceUsedIM += IM_BUCKET_OVERHEAD;
+ 			}
+ 		}
+ 
+ 		free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 	}
+ 
+ 	ReleaseSysCache(statsTuple);
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 147,152 **** ExecHashJoin(HashJoinState *node)
--- 335,345 ----
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
  
+ 		/* Use skew optimization only when there is more than one batch. */
+ 		if (hashtable->nbatch > 1)
+ 			ExecHashJoinDetectSkew(estate, node,
+ 				(outerPlan((Hash *) hashNode->ps.plan))->plan_width);
+ 
  		/*
  		 * execute the Hash node, to build the hash table
  		 */
***************
*** 205,216 **** ExecHashJoin(HashJoinState *node)
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
  			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
--- 398,415 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
+ 			
+ 			/* Does the outer tuple match an IM bucket? */
+ 			node->hj_OuterTupleIMBucketNo =
+ 				ExecHashGetIMBucket(hashtable, hashvalue);
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but in might not belong to the current batch, or it might need
! 			 * to go into an in-memory bucket.
  			 */
! 			if (node->hj_OuterTupleIMBucketNo == IM_INVALID_BUCKET
! 				&& batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
***************
*** 641,647 **** start_over:
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
--- 840,866 ----
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	/* if we just finished the first batch */
! 	if (curbatch == 0)
! 	{
! 		/*
! 		 * Reset some of the skew optimization state variables. IM buckets are
! 		 * no longer being used as of this point because they are only
! 		 * necessary while joining the first batch (before the cleanup phase).
! 		 *
! 		 * Especially need to make sure ExecHashGetIMBucket returns
! 		 * IM_INVALID_BUCKET quickly for all subsequent calls.
! 		 *
! 		 * IM buckets are only taking up memory if this is a multi-batch join
! 		 * and in that case ExecHashTableReset is about to be called which
! 		 * will free all memory currently used by IM buckets and tuples when
! 		 * it deletes hashtable->batchCxt.  If this is a single batch join
! 		 * then imBucket and imBucketFreezeOrder are already NULL and empty.
! 		 */
! 		hashtable->enableSkewOptimization = false;
! 		hashtable->spaceUsedIM = 0;
! 	}
! 	else if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
*** a/src/include/executor/hashjoin.h
--- b/src/include/executor/hashjoin.h
***************
*** 72,77 **** typedef struct HashJoinTupleData
--- 72,96 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ /*
+  * Stores a hashvalue and linked list of tuples that share that hashvalue.
+  *
+  * When processing MCVs to detect skew in the probe relation of a hash join
+  * the hashvalue is generated and stored before any tuples have been read
+  * (see ExecHashJoinDetectSkew).
+  *
+  * Build tuples that hash to the same hashvalue are placed in the bucket while
+  * reading the build relation.
+  */
+ typedef struct HashJoinIMBucket
+ {
+ 	uint32 hashvalue;
+ 	HashJoinTuple tuples;
+ } HashJoinIMBucket;
+ 
+ #define IM_INVALID_BUCKET -1
+ #define IM_WORK_MEM_PERCENT 2
+ #define IM_BUCKET_OVERHEAD MAXALIGN(sizeof(HashJoinIMBucket))
  
  typedef struct HashJoinTableData
  {
***************
*** 113,121 **** typedef struct HashJoinTableData
--- 132,161 ----
  
  	Size		spaceUsed;		/* memory space currently used by tuples */
  	Size		spaceAllowed;	/* upper limit for space used */
+ 	/* memory space currently used by IM buckets and tuples */
+ 	Size		spaceUsedIM;
+ 	/* upper limit for space used by IM buckets and tuples */
+ 	Size		spaceAllowedIM;
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	/* will the join optimize memory usage when probe relation is skewed */
+ 	bool enableSkewOptimization;
+ 	HashJoinIMBucket **imBucket; /* hashtable of IM buckets */
+ 	/*
+ 	 * array of imBucket indexes to the created IM buckets sorted
+ 	 * in the opposite order that they would be frozen to disk
+ 	 */
+ 	uint16 *imBucketFreezeOrder;
+ 	int nIMBuckets; /* # of buckets in the IM buckets hashtable */
+ 	/*
+ 	 * # of used buckets in the IM buckets hashtable and length of
+ 	 * imBucketFreezeOrder array
+ 	 */
+ 	int nUsedIMBuckets;
+ 	/* # of IM buckets that have already been frozen to disk */
+ 	int nIMBucketsFrozen;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
*** a/src/include/executor/nodeHash.h
--- b/src/include/executor/nodeHash.h
***************
*** 45,48 **** extern void ExecChooseHashTableSize(double ntuples, int tupwidth,
--- 45,50 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern int ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue);
+ 
  #endif   /* NODEHASH_H */
*** a/src/include/nodes/execnodes.h
--- b/src/include/nodes/execnodes.h
***************
*** 1389,1394 **** typedef struct MergeJoinState
--- 1389,1395 ----
   *		hj_NeedNewOuter			true if need new outer tuple on next call
   *		hj_MatchedOuter			true if found a join match for current outer
   *		hj_OuterNotEmpty		true if outer relation known not empty
+  *		hj_OuterTupleIMBucketNo	IM bucket# for the current outer tuple
   * ----------------
   */
  
***************
*** 1414,1419 **** typedef struct HashJoinState
--- 1415,1421 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int			hj_OuterTupleIMBucketNo;
  } HashJoinState;
  
  
*** a/src/include/nodes/primnodes.h
--- b/src/include/nodes/primnodes.h
***************
*** 121,128 **** typedef struct Expr
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  * The code doesn't really need varnoold/varoattno, but they are very useful
!  * for debugging and interpreting completed plans, so we keep them around.
   */
  #define    INNER		65000
  #define    OUTER		65001
--- 121,132 ----
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  *
!  * For the most part, the code doesn't really need varnoold/varoattno, but
!  * they are very useful for debugging and interpreting completed plans, so we
!  * keep them around.  As of PostgreSQL 8.4, these values are also used by
!  * ExecHashJoinDetectSkew to fetch MCV statistics when performing multi-batch
!  * hash joins.
   */
  #define    INNER		65000
  #define    OUTER		65001
***************
*** 142,148 **** typedef struct Var
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno, for debugging */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;
--- 146,152 ----
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;

Index: src/backend/executor/nodeHash.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHash.c,v
retrieving revision 1.117
diff -c -r1.117 nodeHash.c
*** src/backend/executor/nodeHash.c	1 Jan 2009 17:23:41 -0000	1.117
--- src/backend/executor/nodeHash.c	2 Mar 2009 07:52:25 -0000
***************
*** 53,58 ****
--- 53,219 ----
  	return NULL;
  }
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashGetIMBucket
+  *
+  *  	Returns the index of the in-memory bucket for this
+  *		hashvalue, or IM_INVALID_BUCKET if the hashvalue is not
+  *		associated with any unfrozen bucket (or if skew
+  *		optimization is not being used).
+  *
+  *		It is possible for a tuple whose join attribute value is
+  *		not a MCV to hash to an in-memory bucket due to the limited
+  * 		number of hash values but it is unlikely and everything
+  *		continues to work even if it does happen. We would
+  *		accidentally cache some less optimal tuples in memory
+  *		but the join result would still be accurate.
+  *
+  *		hashtable->imBucket is an open addressing hashtable of
+  *		in-memory buckets (HashJoinIMBucket).
+  * ----------------------------------------------------------------
+  */
+ int
+ ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue)
+ {
+ 	int bucket;
+ 	
+ 	/* Modulo the hashvalue (using bitmask) to find the IM bucket. */
+ 	bucket = hashvalue & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * While we have not hit a hole in the hashtable and have not hit the
+ 	 * actual bucket we have collided in the hashtable so try the next
+ 	 * bucket location.
+ 	 */
+ 	while (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 		bucket = (bucket + 1) & (hashtable->nIMBuckets - 1);
+ 
+ 	/*
+ 	 * If the bucket exists and has been correctly determined return
+ 	 * the bucket index.
+ 	 */
+ 	if (hashtable->imBucket[bucket] != NULL
+ 		&& hashtable->imBucket[bucket]->hashvalue == hashvalue)
+ 		return bucket;
+ 
+ 	/*
+ 	 * Must have run into an empty location or a frozen bucket which means the
+ 	 * tuple with this hashvalue is not to be handled as if it matches with an
+ 	 * in-memory bucket.
+ 	 */
+ 	return IM_INVALID_BUCKET;
+ }
+ 
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashFreezeNextIMBucket
+  *
+  *		Freeze the tuples of the next in-memory bucket by pushing
+  *		them into the main hashtable.  Buckets are frozen in order
+  *		so that the best tuples are kept in memory the longest.
+  * ----------------------------------------------------------------
+  */
+ static bool
+ ExecHashFreezeNextIMBucket(HashJoinTable hashtable)
+ {
+ 	int bucketToFreeze;
+ 	int bucketno;
+ 	int batchno;
+ 	uint32 hashvalue;
+ 	HashJoinTuple hashTuple;
+ 	HashJoinTuple nextHashTuple;
+ 	HashJoinIMBucket *bucket;
+ 	MinimalTuple mintuple;
+ 
+ 	/* Calculate the imBucket index of the bucket to freeze. */
+ 	bucketToFreeze = hashtable->imBucketFreezeOrder
+ 		[hashtable->nUsedIMBuckets - 1 - hashtable->nIMBucketsFrozen];
+ 
+ 	/* Grab a pointer to the actual IM bucket. */
+ 	bucket = hashtable->imBucket[bucketToFreeze];
+ 	hashvalue = bucket->hashvalue;
+ 
+ 	/*
+ 	 * Grab a pointer to the first tuple in the soon to be frozen IM bucket.
+ 	 */
+ 	hashTuple = bucket->tuples;
+ 
+ 	/*
+ 	 * Calculate which bucket and batch the tuples belong to in the main
+ 	 * non-IM hashtable.
+ 	 */
+ 	ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+ 
+ 	/* until we have read all tuples from this bucket */
+ 	while (hashTuple != NULL)
+ 	{
+ 		/*
+ 		 * Some of this code is very similar to that of ExecHashTableInsert.
+ 		 * We do not call ExecHashTableInsert directly as
+ 		 * ExecHashTableInsert expects a TupleTableSlot and we already have
+ 		 * HashJoinTuples.
+ 		 */
+ 		mintuple = HJTUPLE_MINTUPLE(hashTuple);
+ 
+ 		/* Decide whether to put the tuple in the hash table or a temp file. */
+ 		if (batchno == hashtable->curbatch)
+ 		{
+ 			/* Put the tuple in hash table. */
+ 			nextHashTuple = hashTuple->next;
+ 			hashTuple->next = hashtable->buckets[bucketno];
+ 			hashtable->buckets[bucketno] = hashTuple;
+ 			hashTuple = nextHashTuple;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 		}
+ 		else
+ 		{
+ 			/* Put the tuples into a temp file for later batches. */
+ 			Assert(batchno > hashtable->curbatch);
+ 			ExecHashJoinSaveTuple(mintuple, hashvalue,
+ 								  &hashtable->innerBatchFile[batchno]);
+ 			/*
+ 			 * Some memory has been freed up. This must be done before we
+ 			 * pfree the hashTuple of we lose access to the tuple size.
+ 			 */
+ 			hashtable->spaceUsed -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			hashtable->spaceUsedIM -= HJTUPLE_OVERHEAD + mintuple->t_len;
+ 			nextHashTuple = hashTuple->next;
+ 			pfree(hashTuple);
+ 			hashTuple = nextHashTuple;
+ 		}
+ 	}
+ 
+ 	/*
+ 	 * Free the memory the bucket struct was using as it is not necessary
+ 	 * any more.  All code treats a frozen in-memory bucket the same as one
+ 	 * that did not exist; by setting the pointer to null the rest of the code
+ 	 * will function as if we had not created this bucket at all.
+ 	 */
+ 	pfree(bucket);
+ 	hashtable->imBucket[bucketToFreeze] = NULL;
+ 	hashtable->spaceUsed -= IM_BUCKET_OVERHEAD;
+ 	hashtable->spaceUsedIM -= IM_BUCKET_OVERHEAD;
+ 	hashtable->nIMBucketsFrozen++;
+ 
+ 	/*
+ 	 * All buckets have been frozen and deleted from memory so turn off
+ 	 * skew aware partitioning and remove the structs from memory as they are
+ 	 * just wasting space from now on.
+ 	 */
+ 	if (hashtable->nUsedIMBuckets == hashtable->nIMBucketsFrozen)
+ 	{
+ 		hashtable->enableSkewOptimization = false;
+ 		pfree(hashtable->imBucket);
+ 		pfree(hashtable->imBucketFreezeOrder);
+ 		hashtable->spaceUsed -= hashtable->spaceUsedIM;
+ 		hashtable->spaceUsedIM = 0;
+ 	}
+ 
+ 	return true;
+ }
+ 
  /* ----------------------------------------------------------------
   *		MultiExecHash
   *
***************
*** 69,74 ****
--- 230,237 ----
  	TupleTableSlot *slot;
  	ExprContext *econtext;
  	uint32		hashvalue;
+ 	MinimalTuple mintuple;
+ 	int bucketNumber;
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 87,108 ****
  	econtext = node->ps.ps_ExprContext;
  
  	/*
! 	 * get all inner tuples and insert into the hash table (or temp files)
  	 */
! 	for (;;)
! 	{
! 		slot = ExecProcNode(outerNode);
! 		if (TupIsNull(slot))
! 			break;
! 		/* We have to compute the hash value */
! 		econtext->ecxt_innertuple = slot;
! 		if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false, false,
! 								 &hashvalue))
  		{
! 			ExecHashTableInsert(hashtable, slot, hashvalue);
! 			hashtable->totalTuples += 1;
  		}
- 	}
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
--- 250,346 ----
  	econtext = node->ps.ps_ExprContext;
  
  	/*
! 	 * Get all inner tuples and insert into the hash table (or temp
! 	 * files).
! 	 * 
! 	 * Only incur the overhead of the skew tests if skew
! 	 * optimization is actually being used.
  	 */
! 	if (hashtable->enableSkewOptimization)
! 		for (;;)
  		{
! 			slot = ExecProcNode(outerNode);
! 			if (TupIsNull(slot))
! 				break;
! 			/* We have to compute the hash value */
! 			econtext->ecxt_innertuple = slot;
! 			if (ExecHashGetHashValue(hashtable, econtext, hashkeys
! 									, false, false, &hashvalue))
! 			{
! 				/*
! 				 * Skew optimization may have been subsequently
! 				 * turned off if all the IM buckets were frozen so
! 				 * make sure we handle
! 				 * that cleanly.
! 				 */
! 				if (hashtable->enableSkewOptimization)
! 					bucketNumber = 
! 						ExecHashGetIMBucket(hashtable, hashvalue);
! 				else
! 					bucketNumber = IM_INVALID_BUCKET;
! 
! 				/*
! 				 * handle tuples not destined for an in-memory
! 				 * bucket normally
! 				 */
! 				if (bucketNumber == IM_INVALID_BUCKET)
! 					ExecHashTableInsert(hashtable, slot, hashvalue);
! 				else
! 				{
! 					HashJoinTuple hashTuple;
! 					int			hashTupleSize;
! 					
! 					/* get the HashJoinTuple */
! 					mintuple = ExecFetchSlotMinimalTuple(slot);
! 					hashTupleSize = 
! 						HJTUPLE_OVERHEAD + mintuple->t_len;
! 					hashTuple = (HashJoinTuple)
! 						MemoryContextAlloc(hashtable->batchCxt
! 								, hashTupleSize);
! 					hashTuple->hashvalue = hashvalue;
! 					memcpy(HJTUPLE_MINTUPLE(hashTuple), mintuple
! 								, mintuple->t_len);
! 
! 					/*
! 					 * Push the HashJoinTuple onto the front of the
! 					 * IM bucket.
! 					 */
! 					hashTuple->next = 
! 						hashtable->imBucket[bucketNumber]->tuples;
! 					hashtable->imBucket[bucketNumber]->tuples = 
! 						hashTuple;
! 					
! 					/*
! 					 * More memory is now in use so make sure we are not
! 					 * over spaceAllowedIM.
! 					 */
! 					hashtable->spaceUsed += hashTupleSize;
! 					hashtable->spaceUsedIM += hashTupleSize;
! 					while (hashtable->spaceUsedIM >hashtable->spaceAllowedIM
! 						&& ExecHashFreezeNextIMBucket(hashtable))
! 						;
! 					/* Guarantee we are not over the spaceAllowed. */
! 					if (hashtable->spaceUsed > hashtable->spaceAllowed)
! 						ExecHashIncreaseNumBatches(hashtable);
! 				}
! 				hashtable->totalTuples++;
! 			}
! 		}
! 	else
! 		for (;;)
! 		{
! 			slot = ExecProcNode(outerNode);
! 			if (TupIsNull(slot))
! 				break;
! 			/* We have to compute the hash value */
! 			econtext->ecxt_innertuple = slot;
! 			if (ExecHashGetHashValue(hashtable, econtext, hashkeys, false,
! 									 false, &hashvalue))
! 			{
! 				ExecHashTableInsert(hashtable, slot, hashvalue);
! 				hashtable->totalTuples++;
! 			}
  		}
  
  	/* must provide our own instrumentation support */
  	if (node->ps.instrument)
***************
*** 269,274 ****
--- 507,521 ----
  	hashtable->outerBatchFile = NULL;
  	hashtable->spaceUsed = 0;
  	hashtable->spaceAllowed = work_mem * 1024L;
+ 	/* Initialize skew optimization related hashtable variables. */
+ 	hashtable->spaceUsedIM = 0;
+ 	hashtable->spaceAllowedIM
+ 		= hashtable->spaceAllowed * IM_WORK_MEM_PERCENT / 100;
+ 	hashtable->enableSkewOptimization = false;
+ 	hashtable->nUsedIMBuckets = 0;
+ 	hashtable->nIMBuckets = 0;
+ 	hashtable->imBucket = NULL;
+ 	hashtable->nIMBucketsFrozen = 0;
  
  	/*
  	 * Get info about the hash functions to be used for each hash key. Also
***************
*** 815,825 ****
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
  	 */
! 	if (hashTuple == NULL)
! 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
! 	else
  		hashTuple = hashTuple->next;
  
  	while (hashTuple != NULL)
  	{
--- 1062,1078 ----
  	/*
  	 * hj_CurTuple is NULL to start scanning a new bucket, or the address of
  	 * the last tuple returned from the current bucket.
+ 	 *
+ 	 * If the tuple hashed to an IM bucket then scan the IM bucket
+ 	 * otherwise scan the standard hashtable bucket.
  	 */
! 	if (hashTuple != NULL)
  		hashTuple = hashTuple->next;
+ 	else if (hjstate->hj_OuterTupleIMBucketNo != IM_INVALID_BUCKET)
+ 		hashTuple = hashtable->imBucket[hjstate->hj_OuterTupleIMBucketNo]
+ 			->tuples;
+ 	else
+ 		hashTuple = hashtable->buckets[hjstate->hj_CurBucketNo];
  
  	while (hashTuple != NULL)
  	{
Index: src/backend/executor/nodeHashjoin.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/executor/nodeHashjoin.c,v
retrieving revision 1.97
diff -c -r1.97 nodeHashjoin.c
*** src/backend/executor/nodeHashjoin.c	1 Jan 2009 17:23:41 -0000	1.97
--- src/backend/executor/nodeHashjoin.c	2 Mar 2009 22:26:38 -0000
***************
*** 20,25 ****
--- 20,29 ----
  #include "executor/nodeHash.h"
  #include "executor/nodeHashjoin.h"
  #include "utils/memutils.h"
+ #include "utils/syscache.h"
+ #include "utils/lsyscache.h"
+ #include "parser/parsetree.h"
+ #include "catalog/pg_statistic.h"
  
  
  /* Returns true for JOIN_LEFT and JOIN_ANTI jointypes */
***************
*** 34,39 ****
--- 38,244 ----
  						  TupleTableSlot *tupleSlot);
  static int	ExecHashJoinNewBatch(HashJoinState *hjstate);
  
+ /*
+  * ----------------------------------------------------------------
+  *		ExecHashJoinDetectSkew
+  *
+  *		If MCV statistics can be found for the join attribute of
+  *		this hashjoin then create a hash table of buckets. Each
+  *		bucket will correspond to an MCV hashvalue and will be
+  *		filled with inner relation tuples whose join attribute
+  *		hashes to the same value as that MCV.  If a join
+  *		attribute value is a MCV for the join attribute in the
+  *		outer (probe) relation, tuples with this value in the
+  *		inner (build) relation are more likely to join with outer
+  *		relation tuples and a benefit can be gained by keeping
+  *		them in memory while joining the first batch of tuples.
+  * ----------------------------------------------------------------
+  */
+ static void
+ ExecHashJoinDetectSkew(EState *estate, HashJoinState *hjstate, int tupwidth)
+ {
+ 	HeapTupleData	*statsTuple;
+ 	FuncExprState	*clause;
+ 	ExprState		*argstate;
+ 	Var				*variable;
+ 	HashJoinTable	hashtable;
+ 	Datum			*values;
+ 	int				nvalues;
+ 	float4			*numbers;
+ 	int				nnumbers;
+ 	Oid				relid;
+ 	Oid				atttype;
+ 	int				i;
+ 	int				mcvsToUse;
+ 
+ 	/* Only use statistics if there is a single join attribute. */
+ 	if (hjstate->hashclauses->length != 1)
+ 		return; /* Histojoin is not defined for more than one join key */
+ 	
+ 	hashtable = hjstate->hj_HashTable;
+ 	
+ 	/*
+ 	 * Estimate the number of IM buckets that will fit in
+ 	 * the memory allowed for IM buckets.
+ 	 *
+ 	 * hashtable->imBucket will have up to 8 times as many HashJoinIMBucket
+ 	 * pointers as the number of MCV hashvalues. A uint16 index in
+ 	 * hashtable->imBucketFreezeOrder will be created for each IM bucket. One
+ 	 * actual HashJoinIMBucket struct will be created for each
+ 	 * unique MCV hashvalue so up to one struct per MCV.
+ 	 *
+ 	 * It is also estimated that each IM bucket will have a single build
+ 	 * tuple stored in it after partitioning the build relation input.  This
+ 	 * estimate could be high if tuples are filtered out before this join but
+ 	 * in that case the extra memory is used by the regular hashjoin batch.
+ 	 * This estimate could be low if it is a many to many join but in that
+ 	 * case IM buckets will be frozen to free up memory as needed
+ 	 * during the inner relation partitioning phase.
+ 	 */
+ 	mcvsToUse = hashtable->spaceAllowedIM / (
+ 		/* size of a hash tuple */
+ 		HJTUPLE_OVERHEAD + MAXALIGN(sizeof(MinimalTupleData))
+ 			+ MAXALIGN(tupwidth)
+ 		/* max size of hashtable pointers per MCV */
+ 		+ (8 * sizeof(HashJoinIMBucket*))
+ 		+ sizeof(uint16) /* size of imBucketFreezeOrder entry */
+ 		+ IM_BUCKET_OVERHEAD /* size of IM bucket struct */
+ 		);
+ 	if (mcvsToUse == 0)
+ 		return;	/* No point in considering this any further. */
+ 
+ 	/*
+ 	 * Determine the relation id and attribute id of the single join
+ 	 * attribute of the probe relation.
+ 	 */
+ 	clause = (FuncExprState *) lfirst(list_head(hjstate->hashclauses));
+ 	argstate = (ExprState *) lfirst(list_head(clause->args));
+ 
+ 	/*
+ 	 * Do not try to exploit stats if the join attribute is an expression
+ 	 * instead of just a simple attribute.
+ 	 */		
+ 	if (argstate->expr->type != T_Var)
+ 		return;
+ 
+ 	variable = (Var *) argstate->expr;
+ 	relid = getrelid(variable->varnoold, estate->es_range_table);
+ 	atttype = variable->vartype;
+ 
+ 	statsTuple = SearchSysCache(STATRELATT, ObjectIdGetDatum(relid),
+ 		Int16GetDatum(variable->varoattno), 0, 0);
+ 	if (!HeapTupleIsValid(statsTuple))
+ 		return;
+ 
+ 	/* Look for MCV statistics for the attribute. */
+ 	if (get_attstatsslot(statsTuple, atttype, variable->vartypmod,
+ 		STATISTIC_KIND_MCV, InvalidOid, &values, &nvalues,
+ 		&numbers, &nnumbers))
+ 	{
+ 		FmgrInfo   *hashfunctions;
+ 		int nbuckets = 2;
+ 		double frac = 0;
+ 
+ 		/*
+ 		 * IM buckets (imBucket) is an open addressing hashtable with a
+ 		 * power of 2 size that is greater than the number of MCV values.
+ 		 */
+ 		if (mcvsToUse > nvalues)
+ 			mcvsToUse = nvalues;
+ 		/*
+ 		 * Calculate the expected percent of probe relation to join
+ 		 * with IM buckets
+ 		 */
+ 		for (i = 0; i < mcvsToUse; i++)
+ 			frac += numbers[i];
+ 		/*
+ 		 * Don't enable skew optimization if the benefits are really
+ 		 * small.
+ 		 */
+ 		if (frac < IM_MIN_BENEFIT_PERCENT)
+ 		{
+ 			free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 			ReleaseSysCache(statsTuple);
+ 			return;
+ 		}
+ 		while (nbuckets <= mcvsToUse)
+ 			nbuckets <<= 1;
+ 		/* use two more bits just to help avoid collisions */
+ 		nbuckets <<= 2;
+ 		hashtable->nIMBuckets = nbuckets;
+ 		hashtable->enableSkewOptimization = true;
+ 
+ 		/*
+ 		 * Allocate the bucket memory in the hashtable's batch context
+ 		 * because it is only relevant and necessary during the first batch
+ 		 * and will be nicely removed once the first batch is done.
+ 		 */
+ 		hashtable->imBucket =
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				nbuckets * sizeof(HashJoinIMBucket*));
+ 		hashtable->imBucketFreezeOrder =
+ 			MemoryContextAllocZero(hashtable->batchCxt,
+ 				mcvsToUse * sizeof(uint16));
+ 		/* Count the overhead of the IM pointers immediately. */
+ 		hashtable->spaceUsed += nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 		hashtable->spaceUsedIM +=  nbuckets * sizeof(HashJoinIMBucket*)
+ 			+ mcvsToUse * sizeof(uint16);
+ 
+ 		/*
+ 		 * Grab the hash functions as we will be generating the hashvalues
+ 		 * in this section.
+ 		 */
+ 		hashfunctions = hashtable->outer_hashfunctions;
+ 
+ 		/* Create the buckets */
+ 		for (i = 0; i < mcvsToUse; i++)
+ 		{
+ 			uint32 hashvalue = DatumGetUInt32(
+ 				FunctionCall1(&hashfunctions[0], values[i]));
+ 			int bucket = hashvalue & (nbuckets - 1);
+ 
+ 			/*
+ 			 * While we have not hit a hole in the hashtable and have not hit
+ 			 * a bucket with the same hashvalue we have collided in the
+ 			 * hashtable so try the next bucket location (remember it is an
+ 			 * open addressing hashtable).
+ 			 */
+ 			while (hashtable->imBucket[bucket] != NULL
+ 				&& hashtable->imBucket[bucket]->hashvalue != hashvalue)
+ 				bucket = (bucket + 1) & (nbuckets - 1);
+ 
+ 			/*
+ 			 * Leave bucket alone if it has the same hashvalue as current
+ 			 * MCV. We only want one bucket per hashvalue. Even if two MCV
+ 			 * values hash to the same bucket we are fine.
+ 			 */
+ 			if (hashtable->imBucket[bucket] == NULL)
+ 			{
+ 				/*
+ 				 * Allocate the actual bucket structure in the hashtable's batch
+ 				 * context because it is only relevant and necessary during
+ 				 * the first batch and will be nicely removed once the first
+ 				 * batch is done.
+ 				 */
+ 				hashtable->imBucket[bucket]
+ 					= MemoryContextAllocZero(hashtable->batchCxt,
+ 						sizeof(HashJoinIMBucket));
+ 				hashtable->imBucket[bucket]->hashvalue = hashvalue;
+ 				hashtable->imBucketFreezeOrder[hashtable->nUsedIMBuckets]
+ 					= bucket;
+ 				hashtable->nUsedIMBuckets++;
+ 				/* Count the overhead of the IM bucket struct */
+ 				hashtable->spaceUsed += IM_BUCKET_OVERHEAD;
+ 				hashtable->spaceUsedIM += IM_BUCKET_OVERHEAD;
+ 			}
+ 		}
+ 
+ 		free_attstatsslot(atttype, values, nvalues, numbers, nnumbers);
+ 	}
+ 
+ 	ReleaseSysCache(statsTuple);
+ }
  
  /* ----------------------------------------------------------------
   *		ExecHashJoin
***************
*** 147,152 ****
--- 352,362 ----
  										node->hj_HashOperators);
  		node->hj_HashTable = hashtable;
  
+ 		/* Use skew optimization only when there is more than one batch. */
+ 		if (hashtable->nbatch > 1)
+ 			ExecHashJoinDetectSkew(estate, node,
+ 				(outerPlan((Hash *) hashNode->ps.plan))->plan_width);
+ 
  		/*
  		 * execute the Hash node, to build the hash table
  		 */
***************
*** 172,177 ****
--- 382,394 ----
  		 * again.)
  		 */
  		node->hj_OuterNotEmpty = false;
+ 
+ 		/*
+ 		 * Initialize OuterTupleIMBucketNo as this is always the value when
+ 		 * skew optimization is turned off and it will be set properly later
+ 		 * if skew optimization is on.
+ 		 */
+ 		node->hj_OuterTupleIMBucketNo = IM_INVALID_BUCKET;
  	}
  
  	/*
***************
*** 205,216 ****
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but this tuple may not belong to the current batch.
  			 */
! 			if (batchno != hashtable->curbatch)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
--- 422,444 ----
  			ExecHashGetBucketAndBatch(hashtable, hashvalue,
  									  &node->hj_CurBucketNo, &batchno);
  			node->hj_CurTuple = NULL;
+ 			
+ 			/*
+ 			 * Does the outer tuple match an IM bucket? Make sure we
+ 			 * don't incur the cost of a function call if skew
+ 			 * optimization is turned off.
+ 			 */
+ 			if (hashtable->enableSkewOptimization)
+ 				node->hj_OuterTupleIMBucketNo = 
+ 					ExecHashGetIMBucket(hashtable, hashvalue);
  
  			/*
  			 * Now we've got an outer tuple and the corresponding hash bucket,
! 			 * but in might not belong to the current batch, or it might need
! 			 * to go into an in-memory bucket.
  			 */
! 			if (batchno != hashtable->curbatch
! 				&& node->hj_OuterTupleIMBucketNo == IM_INVALID_BUCKET)
  			{
  				/*
  				 * Need to postpone this outer tuple to a later batch. Save it
***************
*** 641,647 ****
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
--- 869,895 ----
  	nbatch = hashtable->nbatch;
  	curbatch = hashtable->curbatch;
  
! 	/* if we just finished the first batch */
! 	if (curbatch == 0)
! 	{
! 		/*
! 		 * Reset some of the skew optimization state variables. IM buckets are
! 		 * no longer being used as of this point because they are only
! 		 * necessary while joining the first batch (before the cleanup phase).
! 		 *
! 		 * Especially need to make sure ExecHashGetIMBucket returns
! 		 * IM_INVALID_BUCKET quickly for all subsequent calls.
! 		 *
! 		 * IM buckets are only taking up memory if this is a multi-batch join
! 		 * and in that case ExecHashTableReset is about to be called which
! 		 * will free all memory currently used by IM buckets and tuples when
! 		 * it deletes hashtable->batchCxt.  If this is a single batch join
! 		 * then imBucket and imBucketFreezeOrder are already NULL and empty.
! 		 */
! 		hashtable->enableSkewOptimization = false;
! 		hashtable->spaceUsedIM = 0;
! 	}
! 	else if (curbatch > 0)
  	{
  		/*
  		 * We no longer need the previous outer batch file; close it right
Index: src/include/executor/hashjoin.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/hashjoin.h,v
retrieving revision 1.49
diff -c -r1.49 hashjoin.h
*** src/include/executor/hashjoin.h	1 Jan 2009 17:23:59 -0000	1.49
--- src/include/executor/hashjoin.h	2 Mar 2009 22:27:19 -0000
***************
*** 72,77 ****
--- 72,97 ----
  #define HJTUPLE_MINTUPLE(hjtup)  \
  	((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
  
+ /*
+  * Stores a hashvalue and linked list of tuples that share that hashvalue.
+  *
+  * When processing MCVs to detect skew in the probe relation of a hash join
+  * the hashvalue is generated and stored before any tuples have been read
+  * (see ExecHashJoinDetectSkew).
+  *
+  * Build tuples that hash to the same hashvalue are placed in the bucket while
+  * reading the build relation.
+  */
+ typedef struct HashJoinIMBucket
+ {
+ 	uint32 hashvalue;
+ 	HashJoinTuple tuples;
+ } HashJoinIMBucket;
+ 
+ #define IM_INVALID_BUCKET -1
+ #define IM_WORK_MEM_PERCENT 2
+ #define IM_MIN_BENEFIT_PERCENT .01
+ #define IM_BUCKET_OVERHEAD MAXALIGN(sizeof(HashJoinIMBucket))
  
  typedef struct HashJoinTableData
  {
***************
*** 113,121 ****
--- 133,162 ----
  
  	Size		spaceUsed;		/* memory space currently used by tuples */
  	Size		spaceAllowed;	/* upper limit for space used */
+ 	/* memory space currently used by IM buckets and tuples */
+ 	Size		spaceUsedIM;
+ 	/* upper limit for space used by IM buckets and tuples */
+ 	Size		spaceAllowedIM;
  
  	MemoryContext hashCxt;		/* context for whole-hash-join storage */
  	MemoryContext batchCxt;		/* context for this-batch-only storage */
+ 	
+ 	/* will the join optimize memory usage when probe relation is skewed */
+ 	bool enableSkewOptimization;
+ 	HashJoinIMBucket **imBucket; /* hashtable of IM buckets */
+ 	/*
+ 	 * array of imBucket indexes to the created IM buckets sorted
+ 	 * in the opposite order that they would be frozen to disk
+ 	 */
+ 	uint16 *imBucketFreezeOrder;
+ 	int nIMBuckets; /* # of buckets in the IM buckets hashtable */
+ 	/*
+ 	 * # of used buckets in the IM buckets hashtable and length of
+ 	 * imBucketFreezeOrder array
+ 	 */
+ 	int nUsedIMBuckets;
+ 	/* # of IM buckets that have already been frozen to disk */
+ 	int nIMBucketsFrozen;
  } HashJoinTableData;
  
  #endif   /* HASHJOIN_H */
Index: src/include/executor/nodeHash.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/executor/nodeHash.h,v
retrieving revision 1.46
diff -c -r1.46 nodeHash.h
*** src/include/executor/nodeHash.h	1 Jan 2009 17:23:59 -0000	1.46
--- src/include/executor/nodeHash.h	1 Mar 2009 03:17:51 -0000
***************
*** 45,48 ****
--- 45,50 ----
  						int *numbuckets,
  						int *numbatches);
  
+ extern int ExecHashGetIMBucket(HashJoinTable hashtable, uint32 hashvalue);
+ 
  #endif   /* NODEHASH_H */
Index: src/include/nodes/execnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/execnodes.h,v
retrieving revision 1.201
diff -c -r1.201 execnodes.h
*** src/include/nodes/execnodes.h	12 Jan 2009 05:10:45 -0000	1.201
--- src/include/nodes/execnodes.h	1 Mar 2009 03:17:51 -0000
***************
*** 1389,1394 ****
--- 1389,1395 ----
   *		hj_NeedNewOuter			true if need new outer tuple on next call
   *		hj_MatchedOuter			true if found a join match for current outer
   *		hj_OuterNotEmpty		true if outer relation known not empty
+  *		hj_OuterTupleIMBucketNo	IM bucket# for the current outer tuple
   * ----------------
   */
  
***************
*** 1414,1419 ****
--- 1415,1421 ----
  	bool		hj_NeedNewOuter;
  	bool		hj_MatchedOuter;
  	bool		hj_OuterNotEmpty;
+ 	int			hj_OuterTupleIMBucketNo;
  } HashJoinState;
  
  
Index: src/include/nodes/primnodes.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/include/nodes/primnodes.h,v
retrieving revision 1.146
diff -c -r1.146 primnodes.h
*** src/include/nodes/primnodes.h	25 Feb 2009 03:30:37 -0000	1.146
--- src/include/nodes/primnodes.h	1 Mar 2009 03:17:51 -0000
***************
*** 121,128 ****
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  * The code doesn't really need varnoold/varoattno, but they are very useful
!  * for debugging and interpreting completed plans, so we keep them around.
   */
  #define    INNER		65000
  #define    OUTER		65001
--- 121,132 ----
   * subplans; for example, in a join node varno becomes INNER or OUTER and
   * varattno becomes the index of the proper element of that subplan's target
   * list.  But varnoold/varoattno continue to hold the original values.
!  *
!  * For the most part, the code doesn't really need varnoold/varoattno, but
!  * they are very useful for debugging and interpreting completed plans, so we
!  * keep them around.  As of PostgreSQL 8.4, these values are also used by
!  * ExecHashJoinDetectSkew to fetch MCV statistics when performing multi-batch
!  * hash joins.
   */
  #define    INNER		65000
  #define    OUTER		65001
***************
*** 142,148 ****
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno, for debugging */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;
--- 146,152 ----
  	Index		varlevelsup;	/* for subquery variables referencing outer
  								 * relations; 0 in a normal var, >0 means N
  								 * levels up */
! 	Index		varnoold;		/* original value of varno */
  	AttrNumber	varoattno;		/* original value of varattno */
  	int			location;		/* token location, or -1 if unknown */
  } Var;