From 0fb9ec964369c2af246cfc53e3866e3beb2b7ecc Mon Sep 17 00:00:00 2001
From: Lyu Pan <lyup@amazon.com>
Date: Fri, 16 Sep 2022 21:28:52 +0000
Subject: [PATCH 3/5] Support semijoin filter in the executor for parallel
 mergejoin.

Each worker process create its own bloom filter, and updates its own bloom filter during the outer/left plan scan (no lock is needed). After all processes finish the scan, the bloom filtered are merged together (by performing OR operations on the bit arrays) into a shared bloom filter in the Dynamic Shared Memory area (use lock for synchronization). After this is completed, the merged bloom filter will be copied back to all the worker processes, and be used in the inner/right plan scan for filtering.
---
 src/backend/executor/execScan.c      |   3 +-
 src/backend/executor/nodeMergejoin.c | 221 +++++++++++++++++++++++++--
 src/backend/executor/nodeSeqscan.c   | 173 ++++++++++++++++++++-
 src/backend/lib/bloomfilter.c        |  74 +++++++++
 src/include/executor/nodeMergejoin.h |   4 +-
 src/include/lib/bloomfilter.h        |   7 +
 src/include/nodes/execnodes.h        |  31 ++++
 7 files changed, 495 insertions(+), 18 deletions(-)

diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 4b97c39455..954d99392b 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -215,7 +215,8 @@ ExecScan(ScanState *node,
 			{
 				SemiJoinFilterFinishScan(
 										 ((SeqScanState *) node)->semiJoinFilters,
-										 node->ss_currentRelation->rd_id);
+										 node->ss_currentRelation->rd_id,
+										 node->ps.state->es_query_dsa);
 			}
 			if (projInfo)
 				return ExecClearTuple(projInfo->pi_state.resultslot);
diff --git a/src/backend/executor/nodeMergejoin.c b/src/backend/executor/nodeMergejoin.c
index 81c253aad5..d8fb352313 100644
--- a/src/backend/executor/nodeMergejoin.c
+++ b/src/backend/executor/nodeMergejoin.c
@@ -99,6 +99,10 @@
 #include "executor/nodeMergejoin.h"
 #include "lib/bloomfilter.h"
 #include "miscadmin.h"
+#include "storage/dsm.h"
+#include "storage/lwlock.h"
+#include "storage/shm_toc.h"
+#include "utils/dsa.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 
@@ -659,6 +663,26 @@ ExecMergeJoin(PlanState *pstate)
 				outerTupleSlot = ExecProcNode(outerPlan);
 				node->mj_OuterTupleSlot = outerTupleSlot;
 
+				/*
+				 * Check if outer plan has an SJF and the inner plan does not.
+				 * This case will only arise during parallel execution, when
+				 * the outer plan is initialized with the SJF but the inner
+				 * plan does not because it is not included in the memory
+				 * copied over during worker creation. If this is the case,
+				 * then push down the filter to the inner plan level to
+				 * correct this error and then proceed as normal.
+				 */
+				if (GetSemiJoinFilter(outerPlan, pstate->plan->plan_node_id)
+					&& !GetSemiJoinFilter(innerPlan,
+										  pstate->plan->plan_node_id))
+				{
+					SemiJoinFilter *sjf = GetSemiJoinFilter(outerPlan,
+															pstate->plan->plan_node_id);
+
+					PushDownFilter(innerPlan, NULL, sjf, &sjf->checkingId,
+								   NULL);
+				}
+
 				/* Compute join values and check for unmatchability */
 				switch (MJEvalOuterValues(node))
 				{
@@ -1819,11 +1843,14 @@ PushDownDirection(PlanState *node)
 	}
 }
 
-/* Recursively pushes down the filter until an appropriate SeqScan node is reached. Then, it
- * verifies if that SeqScan node is the one we want to push the filter to, and if it is, then
- * appends the SJF to the node. */
+/*
+ * Recursively pushes down the filter until an appropriate SeqScan node is
+ * reached. Then, it verifies if that SeqScan node is the one we want to push
+ * the filter to, and if it is, then appends the SJF to the node.
+ */
 void
-PushDownFilter(PlanState *node, Plan *plan, SemiJoinFilter * sjf, Oid *relId, int64 *nodeRows)
+PushDownFilter(PlanState *node, Plan *plan, SemiJoinFilter * sjf, Oid *relId,
+			   int64 *nodeRows)
 {
 	if (node == NULL)
 	{
@@ -1838,8 +1865,8 @@ PushDownFilter(PlanState *node, Plan *plan, SemiJoinFilter * sjf, Oid *relId, in
 		Assert(IsA(scan, SeqScanState));
 
 		/*
-		 * found the right Scan node that we want to apply the filter onto via
-		 * matching relId
+		 * Found the right Scan node that we want to apply the filter onto via
+		 * matching relId.
 		 */
 		if (scan->ss.ss_currentRelation->rd_id == *relId)
 		{
@@ -1876,13 +1903,15 @@ PushDownFilter(PlanState *node, Plan *plan, SemiJoinFilter * sjf, Oid *relId, in
 }
 
 /*
- * If this table is the building-side table for the SemiJoinFilter, adds the element to
- * the bloom filter and always returns true. If this table is the checking-side table for the SemiJoinFilter,
- * then checks the element against the bloom filter and returns true if the element is (probably) in the set,
- * and false if the element is not in the bloom filter.
+ * If this table is the building-side table for the SemiJoinFilter, adds the
+ * element to the bloom filter and always returns true. If this table is the
+ * checking-side table for the SemiJoinFilter, then checks the element
+ * against the bloom filter and returns true if the element is (probably)
+ * in the set, and false if the element is not in the bloom filter.
  */
 bool
-SemiJoinFilterExamineSlot(List *semiJoinFilters, TupleTableSlot *slot, Oid tableId)
+SemiJoinFilterExamineSlot(List *semiJoinFilters,
+						  TupleTableSlot *slot, Oid tableId)
 {
 	ListCell   *cell;
 
@@ -1903,7 +1932,8 @@ SemiJoinFilterExamineSlot(List *semiJoinFilters, TupleTableSlot *slot, Oid table
 			 * include multiple join keys.
 			 */
 			val = slot->tts_values[sjf->buildingAttr];
-			bloom_add_element(sjf->filter, (unsigned char *) &val, sizeof(val));
+			bloom_add_element(sjf->filter, (unsigned char *) &val,
+							  sizeof(val));
 		}
 		else if (sjf->doneBuilding && tableId == sjf->checkingId)
 		{
@@ -1912,7 +1942,8 @@ SemiJoinFilterExamineSlot(List *semiJoinFilters, TupleTableSlot *slot, Oid table
 			slot_getsomeattrs(slot, sjf->checkingAttr + 1);
 			sjf->elementsChecked++;
 			val = slot->tts_values[sjf->checkingAttr];
-			if (bloom_lacks_element(sjf->filter, (unsigned char *) &val, sizeof(val)))
+			if (bloom_lacks_element(sjf->filter,
+									(unsigned char *) &val, sizeof(val)))
 			{
 				sjf->elementsFiltered++;
 				return false;
@@ -1923,7 +1954,8 @@ SemiJoinFilterExamineSlot(List *semiJoinFilters, TupleTableSlot *slot, Oid table
 }
 
 void
-SemiJoinFilterFinishScan(List *semiJoinFilters, Oid tableId)
+SemiJoinFilterFinishScan(List *semiJoinFilters, Oid tableId,
+						 dsa_area *parallel_area)
 {
 	ListCell   *cell;
 
@@ -1933,7 +1965,166 @@ SemiJoinFilterFinishScan(List *semiJoinFilters, Oid tableId)
 
 		if (!sjf->doneBuilding && tableId == sjf->buildingId)
 		{
-			sjf->doneBuilding = true;
+			if (!sjf->isParallel)
+			{
+				/*
+				 * Not parallel, so only one process running and that process
+				 * is now complete.
+				 */
+				sjf->doneBuilding = true;
+			}
+			else
+			{
+				/* parallel, so need to sync with the other processes */
+				SemiJoinFilterParallelState *parallelState =
+				(SemiJoinFilterParallelState *) dsa_get_address(
+																parallel_area, sjf->parallelState);
+				bloom_filter *shared_bloom =
+				(bloom_filter *) dsa_get_address(
+												 parallel_area, parallelState->bloom_dsa_address);
+
+				/*
+				 * This process takes control of the lock and updates the
+				 * shared bloom filter. These locks are created by the
+				 * SemiJoinFilterParallelState and are unique to that struct.
+				 */
+				LWLockAcquire(&parallelState->lock, LW_EXCLUSIVE);
+				parallelState->elementsAdded += sjf->elementsAdded;
+				add_to_filter(shared_bloom, sjf->filter);
+				parallelState->workersDone++;
+				LWLockRelease(&parallelState->lock);
+
+				/*
+				 * We need to wait until all threads have had their chance to
+				 * update the shared bloom filter, since our next step is to
+				 * copy the finished bloom filter back into all of the
+				 * separate processes.
+				 */
+				if (parallelState->workersDone == parallelState->numProcesses)
+				{
+					LWLockUpdateVar(&parallelState->secondlock,
+									&parallelState->lockStop, 1);
+				}
+				LWLockWaitForVar(&parallelState->secondlock,
+								 &parallelState->lockStop, 0,
+								 &parallelState->lockStop);
+
+				/*
+				 * Now the shared Bloom filter is fully updated, so each
+				 * individual process copies the finished Bloom filter to the
+				 * local SemiJoinFilter.
+				 */
+				LWLockAcquire(&parallelState->lock, LW_EXCLUSIVE);
+				replace_bitset(sjf->filter, shared_bloom);
+				sjf->elementsAdded = parallelState->elementsAdded;
+				sjf->doneBuilding = true;
+				parallelState->workersDone++;
+				LWLockRelease(&parallelState->lock);
+
+				/*
+				 * Again, we need to wait for all processes to finish copying
+				 * the completed bloom filter because the main process will
+				 * free the shared memory afterwards.
+				 */
+				if (parallelState->workersDone ==
+					2 * parallelState->numProcesses)
+				{
+					LWLockUpdateVar(&parallelState->secondlock,
+									&parallelState->lockStop, 2);
+				}
+				LWLockWaitForVar(&parallelState->secondlock,
+								 &parallelState->lockStop, 1,
+								 &parallelState->lockStop);
+				/* release allocated shared memory in main process */
+				if (!sjf->isWorker)
+				{
+					LWLockRelease(&parallelState->secondlock);
+					bloom_free_in_dsa(parallel_area,
+									  parallelState->bloom_dsa_address);
+					dsa_free(parallel_area, sjf->parallelState);
+				}
+			}
 		}
 	}
 }
+
+dsa_pointer
+CreateFilterParallelState(dsa_area *area, SemiJoinFilter * sjf,
+						  int sjf_num)
+{
+	dsa_pointer bloom_dsa_address =
+	bloom_create_in_dsa(area, sjf->num_elements, sjf->work_mem, sjf->seed);
+	dsa_pointer parallel_address =
+	dsa_allocate0(area, sizeof(SemiJoinFilterParallelState));
+	SemiJoinFilterParallelState *parallelState =
+	(SemiJoinFilterParallelState *) dsa_get_address(area,
+													parallel_address);
+
+	/* copy over information to parallel state */
+	parallelState->doneBuilding = sjf->doneBuilding;
+	parallelState->seed = sjf->seed;
+	parallelState->num_elements = sjf->num_elements;
+	parallelState->work_mem = sjf->work_mem;
+	parallelState->buildingId = sjf->buildingId;
+	parallelState->checkingId = sjf->checkingId;
+	parallelState->buildingAttr = sjf->buildingAttr;
+	parallelState->checkingAttr = sjf->checkingAttr;
+	parallelState->bloom_dsa_address = bloom_dsa_address;
+	parallelState->sjf_num = sjf_num;
+	parallelState->mergejoin_plan_id = sjf->mergejoin_plan_id;
+	/* initialize locks */
+	LWLockInitialize(&parallelState->lock, LWLockNewTrancheId());
+	LWLockInitialize(&parallelState->secondlock, LWLockNewTrancheId());
+	/* should be main process that acquires lock */
+	LWLockAcquire(&parallelState->secondlock, LW_EXCLUSIVE);
+	return parallel_address;
+}
+
+/*
+ * Checks a side of the execution tree and fetches an SJF if its mergejoin
+ * plan ID matches that of the method's mergejoin ID. Used during parallel
+ * execution, where SJF information is lost during information copying to
+ * the worker.
+ */
+SemiJoinFilter *
+GetSemiJoinFilter(PlanState *node, int plan_id)
+{
+	if (node == NULL)
+	{
+		return NULL;
+	}
+	check_stack_depth();
+	if (node->type == T_SeqScanState)
+	{
+		SeqScanState *scan = (SeqScanState *) node;
+
+		Assert(IsA(scan, SeqScanState));
+		if (scan->applySemiJoinFilter)
+		{
+			ListCell   *lc;
+
+			foreach(lc, scan->semiJoinFilters)
+			{
+				SemiJoinFilter *sjf = (SemiJoinFilter *) lfirst(lc);
+
+				if (sjf->mergejoin_plan_id == plan_id)
+				{
+					return sjf;
+				}
+			}
+			return NULL;
+		}
+	}
+	if (PushDownDirection(node) == 1)
+	{
+		/* check both children and return the non-null one */
+		return GetSemiJoinFilter(node->lefttree, plan_id) != NULL ?
+			GetSemiJoinFilter(node->lefttree, plan_id) :
+			GetSemiJoinFilter(node->righttree, plan_id);
+	}
+	if (PushDownDirection(node) == 0)
+	{
+		return GetSemiJoinFilter(node->lefttree, plan_id);
+	}
+	return NULL;
+}
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index e43ce3f8d0..c796d0c416 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -30,9 +30,18 @@
 #include "access/relscan.h"
 #include "access/tableam.h"
 #include "executor/execdebug.h"
+#include "executor/nodeMergejoin.h"
 #include "executor/nodeSeqscan.h"
+#include "storage/lwlock.h"
+#include "storage/shm_toc.h"
 #include "utils/rel.h"
 
+/*
+ * Magic number for location of shared dsa pointer if scan is using a semi-join
+ * filter.
+ */
+#define DSA_LOCATION_KEY_FOR_SJF	UINT64CONST(0xE00000000000FFFF)
+
 static TupleTableSlot *SeqNext(SeqScanState *node);
 
 /* ----------------------------------------------------------------
@@ -157,7 +166,8 @@ ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 	/* and create slot with the appropriate rowtype */
 	ExecInitScanTupleSlot(estate, &scanstate->ss,
 						  RelationGetDescr(scanstate->ss.ss_currentRelation),
-						  table_slot_callbacks(scanstate->ss.ss_currentRelation));
+						  table_slot_callbacks(
+											   scanstate->ss.ss_currentRelation));
 
 	/*
 	 * Initialize result type and projection.
@@ -262,6 +272,20 @@ ExecSeqScanEstimate(SeqScanState *node,
 												  estate->es_snapshot);
 	shm_toc_estimate_chunk(&pcxt->estimator, node->pscan_len);
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+	/*
+	 * Estimate space for extra dsa_pointer address for when parallel
+	 * sequential scans use a semi-join filter.
+	 */
+	if (node->ss.ps.plan->parallel_aware && node->applySemiJoinFilter)
+	{
+		shm_toc_estimate_keys(&pcxt->estimator, 1);
+		if (node->semiJoinFilters)
+		{
+			shm_toc_estimate_keys(&pcxt->estimator,
+								  sizeof(dsa_pointer) * list_length(node->semiJoinFilters));
+		}
+	}
 }
 
 /* ----------------------------------------------------------------
@@ -277,6 +301,51 @@ ExecSeqScanInitializeDSM(SeqScanState *node,
 	EState	   *estate = node->ss.ps.state;
 	ParallelTableScanDesc pscan;
 
+	/*
+	 * If scan is using a semi-join filter, then initialize dsa pointer of
+	 * shared sjf.
+	 */
+	if (node->applySemiJoinFilter)
+	{
+		int			sjf_num = list_length(node->semiJoinFilters);
+		dsa_pointer *dsa_pointer_address;	/* an array of size sjf_num */
+		ListCell   *lc;
+		int			i = 0;
+
+		dsa_pointer_address = (dsa_pointer *) shm_toc_allocate(pcxt->toc,
+															   sizeof(dsa_pointer) * sjf_num);
+		foreach(lc, node->semiJoinFilters)
+		{
+			SemiJoinFilter *sjf = (SemiJoinFilter *) (lfirst(lc));
+			SemiJoinFilterParallelState *parallelState;
+			dsa_area   *area = node->ss.ps.state->es_query_dsa;
+
+			sjf->parallelState = CreateFilterParallelState(area, sjf, sjf_num);
+			sjf->isParallel = true;
+			/* check if main process always will run */
+			parallelState = (SemiJoinFilterParallelState *) dsa_get_address(area, sjf->parallelState);
+			parallelState->numProcesses = 1;
+			/* update parallelState with built bloom filter */
+			if (sjf->doneBuilding &&
+				node->ss.ss_currentRelation->rd_id == sjf->checkingId)
+			{
+				bloom_filter *parallel_bloom = (bloom_filter *) dsa_get_address(area, parallelState->bloom_dsa_address);
+
+				replace_bitset(parallel_bloom, sjf->filter);
+				LWLockRelease(&parallelState->secondlock);
+			}
+			dsa_pointer_address[i] = sjf->parallelState;
+			i++;
+		}
+
+		/*
+		 * Add plan_id to magic number so this is also unique for each plan
+		 * node.
+		 */
+		shm_toc_insert(pcxt->toc, DSA_LOCATION_KEY_FOR_SJF +
+					   node->ss.ps.plan->plan_node_id, dsa_pointer_address);
+	}
+
 	pscan = shm_toc_allocate(pcxt->toc, node->pscan_len);
 	table_parallelscan_initialize(node->ss.ss_currentRelation,
 								  pscan,
@@ -317,4 +386,106 @@ ExecSeqScanInitializeWorker(SeqScanState *node,
 	pscan = shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, false);
 	node->ss.ss_currentScanDesc =
 		table_beginscan_parallel(node->ss.ss_currentRelation, pscan);
+
+	/*
+	 * Create worker's semi-join filter for merge join, if using it. We first
+	 * need to check shm_toc to see if a sjf exists, then create the local
+	 * backend sjf.
+	 */
+	if (shm_toc_lookup(pwcxt->toc,
+					   DSA_LOCATION_KEY_FOR_SJF + node->ss.ps.plan->plan_node_id, 1))
+	{
+		dsa_pointer *parallel_addresses = (dsa_pointer *)
+		shm_toc_lookup(pwcxt->toc,
+					   DSA_LOCATION_KEY_FOR_SJF + node->ss.ps.plan->plan_node_id, 1);
+
+		/*
+		 * we know that there is at least one sjf, we will update accordingly
+		 * if the parallel state says there is more (this avoids using an
+		 * additional shm_toc allocation).
+		 */
+		int			sjf_num = 1;
+
+		/*
+		 * If a copy of any sjf already exists on the backend, we want to free
+		 * it and create a new one.
+		 */
+		if (node->applySemiJoinFilter)
+		{
+			while (list_length(node->semiJoinFilters) > 0)
+			{
+				SemiJoinFilter *sjf = (SemiJoinFilter *)
+				(list_head(node->semiJoinFilters)->ptr_value);
+
+				node->semiJoinFilters =
+					list_delete_nth_cell(node->semiJoinFilters, 0);
+				FreeSemiJoinFilter(sjf);
+			}
+		}
+
+		/*
+		 * Here, we create the process-local SJF's, which will later be
+		 * combined into the single SJF after all parallel work is done.
+		 */
+		for (int i = 0; i < sjf_num; i++)
+		{
+			dsa_pointer parallel_address = parallel_addresses[i];
+			SemiJoinFilterParallelState *parallelState =
+			(SemiJoinFilterParallelState *)
+			dsa_get_address(node->ss.ps.state->es_query_dsa,
+							parallel_address);
+			SemiJoinFilter *sjf;
+			MemoryContext oldContext;
+
+			sjf_num = parallelState->sjf_num;
+			oldContext = MemoryContextSwitchTo(GetMemoryChunkContext(node));
+			sjf = (SemiJoinFilter *) palloc0(sizeof(SemiJoinFilter));
+			sjf->filter = bloom_create(parallelState->num_elements,
+									   parallelState->work_mem,
+									   parallelState->seed);
+			sjf->buildingId = parallelState->buildingId;
+			sjf->checkingId = parallelState->checkingId;
+			sjf->seed = parallelState->seed;
+			sjf->isParallel = true;
+			sjf->isWorker = true;
+			sjf->doneBuilding = parallelState->doneBuilding;
+			sjf->parallelState = parallel_address;
+			node->applySemiJoinFilter = true;
+			sjf->buildingAttr = parallelState->buildingAttr;
+			sjf->checkingAttr = parallelState->checkingAttr;
+			node->semiJoinFilters =
+				lappend(node->semiJoinFilters, (void *) sjf);
+			sjf->mergejoin_plan_id = parallelState->mergejoin_plan_id;
+			/* copy over bloom filter if already built */
+			if (sjf->doneBuilding &&
+				parallelState->checkingId ==
+				node->ss.ss_currentRelation->rd_id)
+			{
+				SemiJoinFilterParallelState *parallelState =
+				(SemiJoinFilterParallelState *)
+				dsa_get_address(node->ss.ps.state->es_query_dsa,
+								sjf->parallelState);
+				bloom_filter *shared_bloom = (bloom_filter *) dsa_get_address(
+																			  node->ss.ps.state->es_query_dsa,
+																			  parallelState->bloom_dsa_address);
+
+				replace_bitset(sjf->filter, shared_bloom);
+			}
+			else if (!sjf->doneBuilding &&
+					 parallelState->buildingId ==
+					 node->ss.ss_currentRelation->rd_id)
+			{
+				/*
+				 * Add this process to number of scan processes, need to use
+				 * lock in case of multiple workers updating at same time. We
+				 * want to avoid using the planned number of workers because
+				 * that can be wrong.
+				 */
+				LWLockAcquire(&parallelState->lock, LW_EXCLUSIVE);
+				parallelState->numProcesses += 1;
+				LWLockRelease(&parallelState->lock);
+			}
+			MemoryContextSwitchTo(oldContext);
+		}
+	}
 }
diff --git a/src/backend/lib/bloomfilter.c b/src/backend/lib/bloomfilter.c
index 3ef67d35ac..0a05ada9b6 100644
--- a/src/backend/lib/bloomfilter.c
+++ b/src/backend/lib/bloomfilter.c
@@ -128,6 +128,56 @@ bloom_free(bloom_filter *filter)
 	pfree(filter);
 }
 
+/*
+ * Create Bloom filter in dsa shared memory
+ */
+dsa_pointer
+bloom_create_in_dsa(dsa_area *area, int64 total_elems, int bloom_work_mem, uint64 seed)
+{
+	dsa_pointer filter_dsa_address;
+	bloom_filter *filter;
+	int			bloom_power;
+	uint64		bitset_bytes;
+	uint64		bitset_bits;
+
+	/*
+	 * Aim for two bytes per element; this is sufficient to get a false
+	 * positive rate below 1%, independent of the size of the bitset or total
+	 * number of elements.  Also, if rounding down the size of the bitset to
+	 * the next lowest power of two turns out to be a significant drop, the
+	 * false positive rate still won't exceed 2% in almost all cases.
+	 */
+	bitset_bytes = Min(bloom_work_mem * UINT64CONST(1024), total_elems * 2);
+	bitset_bytes = Max(1024 * 1024, bitset_bytes);
+
+	/*
+	 * Size in bits should be the highest power of two <= target.  bitset_bits
+	 * is uint64 because PG_UINT32_MAX is 2^32 - 1, not 2^32
+	 */
+	bloom_power = my_bloom_power(bitset_bytes * BITS_PER_BYTE);
+	bitset_bits = UINT64CONST(1) << bloom_power;
+	bitset_bytes = bitset_bits / BITS_PER_BYTE;
+
+	/* Allocate bloom filter with unset bitset */
+	filter_dsa_address = dsa_allocate0(area, offsetof(bloom_filter, bitset) +
+									   sizeof(unsigned char) * bitset_bytes);
+	filter = (bloom_filter *) dsa_get_address(area, filter_dsa_address);
+	filter->k_hash_funcs = optimal_k(bitset_bits, total_elems);
+	filter->seed = seed;
+	filter->m = bitset_bits;
+
+	return filter_dsa_address;
+}
+
+/*
+ * Free Bloom filter in dsa shared memory
+ */
+void
+bloom_free_in_dsa(dsa_area *area, dsa_pointer filter_dsa_address)
+{
+	dsa_free(area, filter_dsa_address);
+}
+
 /*
  * Add element to Bloom filter
  */
@@ -292,3 +342,27 @@ mod_m(uint32 val, uint64 m)
 
 	return val & (m - 1);
 }
+
+/*
+ * Add secondary filter to main filter, essentially "combining" the two filters together.
+ * This happens in-place, with main_filter being the combined filter.
+ * Both filters must have the same seed and size for this to work.
+ */
+void
+add_to_filter(bloom_filter *main_filter, bloom_filter *to_add)
+{
+	Assert(main_filter->seed == to_add->seed);
+	Assert(main_filter->m == to_add->m);
+	/* m is in bits not bytes */
+	for (int i = 0; i < main_filter->m / BITS_PER_BYTE; i++)
+	{
+		main_filter->bitset[i] = main_filter->bitset[i] | to_add->bitset[i];
+	}
+}
+
+void
+replace_bitset(bloom_filter *main_filter, bloom_filter *overriding_filter)
+{
+	Assert(main_filter->m == overriding_filter->m);
+	memcpy(&main_filter->bitset, &overriding_filter->bitset, main_filter->m / BITS_PER_BYTE);
+}
diff --git a/src/include/executor/nodeMergejoin.h b/src/include/executor/nodeMergejoin.h
index c311c7ed80..d4cc439315 100644
--- a/src/include/executor/nodeMergejoin.h
+++ b/src/include/executor/nodeMergejoin.h
@@ -23,6 +23,8 @@ extern void FreeSemiJoinFilter(SemiJoinFilter * sjf);
 extern int	PushDownDirection(PlanState *node);
 extern void PushDownFilter(PlanState *node, Plan *plan, SemiJoinFilter * sjf, Oid *relId, int64 *nodeRows);
 extern bool SemiJoinFilterExamineSlot(List *semiJoinFilters, TupleTableSlot *slot, Oid tableId);
-extern void SemiJoinFilterFinishScan(List *semiJoinFilters, Oid tableId);
+extern void SemiJoinFilterFinishScan(List *semiJoinFilters, Oid tableId, dsa_area *parallel_area);
+extern dsa_pointer CreateFilterParallelState(dsa_area *area, SemiJoinFilter * sjf, int sjf_num);
+extern SemiJoinFilter * GetSemiJoinFilter(PlanState *node, int plan_id);
 
 #endif							/* NODEMERGEJOIN_H */
diff --git a/src/include/lib/bloomfilter.h b/src/include/lib/bloomfilter.h
index 8146d8e7fd..3b5d1821a5 100644
--- a/src/include/lib/bloomfilter.h
+++ b/src/include/lib/bloomfilter.h
@@ -13,15 +13,22 @@
 #ifndef BLOOMFILTER_H
 #define BLOOMFILTER_H
 
+#include "utils/dsa.h"
+
 typedef struct bloom_filter bloom_filter;
 
 extern bloom_filter *bloom_create(int64 total_elems, int bloom_work_mem,
 								  uint64 seed);
 extern void bloom_free(bloom_filter *filter);
+extern dsa_pointer bloom_create_in_dsa(dsa_area *area, int64 total_elems,
+									   int bloom_work_mem, uint64 seed);
+extern void bloom_free_in_dsa(dsa_area *area, dsa_pointer filter_dsa_address);
 extern void bloom_add_element(bloom_filter *filter, unsigned char *elem,
 							  size_t len);
 extern bool bloom_lacks_element(bloom_filter *filter, unsigned char *elem,
 								size_t len);
 extern double bloom_prop_bits_set(bloom_filter *filter);
+extern void add_to_filter(bloom_filter *main_filter, bloom_filter *to_add);
+extern void replace_bitset(bloom_filter *main_filter, bloom_filter *overriding_filter);
 
 #endif							/* BLOOMFILTER_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 6964462720..6ca6de437b 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -40,6 +40,8 @@
 #include "nodes/tidbitmap.h"
 #include "partitioning/partdefs.h"
 #include "storage/condition_variable.h"
+#include "storage/lwlock.h"
+#include "storage/shm_toc.h"
 #include "utils/hsearch.h"
 #include "utils/queryenvironment.h"
 #include "utils/reltrigger.h"
@@ -2026,6 +2028,10 @@ typedef struct SemiJoinFilter
 	Oid			checkingId;
 	int			checkingAttr;
 	bool		doneBuilding;
+	/* Parallel information */
+	bool		isParallel;
+	bool		isWorker;
+	dsa_pointer parallelState;
 	/* metadata */
 	uint64		seed;
 	int64		num_elements;
@@ -2036,6 +2042,31 @@ typedef struct SemiJoinFilter
 	int			mergejoin_plan_id;
 }			SemiJoinFilter;
 
+typedef struct SemiJoinFilterParallelState
+{
+	/* bloom filter information */
+	uint64		seed;
+	int64		num_elements;
+	int			work_mem;
+	dsa_pointer bloom_dsa_address;
+	/* information to copy over to worker processes */
+	int			numAttr;
+	Oid			buildingId;
+	Oid			checkingId;
+	int			buildingAttr;
+	int			checkingAttr;
+	int			elementsAdded;
+	/* information for parallelization and locking */
+	bool		doneBuilding;
+	int			workersDone;
+	int			numProcesses;
+	uint64		lockStop;
+	LWLock		lock;
+	LWLock		secondlock;
+	int			sjf_num;
+	int			mergejoin_plan_id;
+}			SemiJoinFilterParallelState;
+
 typedef struct MergeJoinState
 {
 	JoinState	js;				/* its first field is NodeTag */
-- 
2.37.1

