limit_batches.patch

49c49
< static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable);
---
> static bool ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable);
896a897,901
> 		/* If memory overhead of maintaining batches is expected to be larger than size of hash table itself,
> 		 * then try to split available memory equaly between hash table and batches
> 		 */
> 		if (dbatch*HASH_CHUNK_SIZE > work_mem*1024L)
> 			dbatch = (work_mem*1024LL + dbatch*HASH_CHUNK_SIZE) / HASH_CHUNK_SIZE;
1021a1027,1031
> 
> 	/* Prevent increasing number of batches if their memory overhead is too large */
> 	if (nbatch*HASH_CHUNK_SIZE > work_mem*1024L)
> 		return;
> 
1166c1176
< static void
---
> static bool
1172a1183,1186
> 	/* Prevent increasing number of batches if their memory overhead is too large */
> 	if (hashtable->nbatch*2*HASH_CHUNK_SIZE > work_mem*1024L)
> 		return false;
> 
1374a1389
> 	return true;
1831d1845
< 		HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
2234a2249
> 	Assert(BarrierParticipants(&batch->batch_barrier) == 1);
3105,3106c3120,3127
< 			ExecParallelHashIncreaseNumBatches(hashtable);
< 		else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
---
> 		{
> 			if (ExecParallelHashIncreaseNumBatches(hashtable))
> 				return NULL;
> 			LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
> 			pstate->growth = PHJ_GROWTH_DISABLED;
> 		}
> 		else
> 		{
3109,3110c3130,3132
< 		/* The caller must retry. */
< 		return NULL;
---
> 			/* The caller must retry. */
> 			return NULL;
> 		}
3654c3676
< 			ExecParallelHashIncreaseNumBatches(hashtable);
---
> 			return !ExecParallelHashIncreaseNumBatches(hashtable);
3655a3678
> 		{
3657,3658c3680,3681
< 
< 		return false;
---
> 			return false;
> 		}