From e0ddb12f2f031ec5460bf4039b60dd3e7ca397e2 Mon Sep 17 00:00:00 2001
From: Lyu Pan <lyu.steve.pan@gmail.com>
Date: Wed, 12 Oct 2022 17:03:54 +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      |  78 +++++++++++++-
 src/backend/executor/nodeMergejoin.c |  90 ++++++++++++++++
 src/backend/executor/nodeSeqscan.c   | 149 +++++++++++++++++++++++++++
 src/backend/lib/bloomfilter.c        |  74 +++++++++++++
 src/include/executor/nodeMergejoin.h |   2 +
 src/include/lib/bloomfilter.h        |   7 ++
 src/include/nodes/execnodes.h        |  34 ++++++
 7 files changed, 429 insertions(+), 5 deletions(-)

diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 7dd7b07a51..71bc9cdc20 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -25,7 +25,7 @@
 
 
 bool		ExecScanUsingSemiJoinFilter(SeqScanState *node, ExprContext *econtext);
-void		ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semijoin_scan);
+void		ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semijoin_scan, dsa_area *parallel_area);
 
 
 /*
@@ -213,7 +213,8 @@ ExecScan(ScanState *node,
 		{
 			if (IsA(node, SeqScanState) && ((SeqScanState *) node)->apply_semijoin_filter)
 			{
-				ExecSemiJoinFilterFinishScan(((SeqScanState *) node)->semijoin_filters, ((SeqScanState *) node)->sj_scan_data);
+				ExecSemiJoinFilterFinishScan(((SeqScanState *) node)->semijoin_filters, ((SeqScanState *) node)->sj_scan_data,
+											 node->ps.state->es_query_dsa);
 			}
 			if (projInfo)
 				return ExecClearTuple(projInfo->pi_state.resultslot);
@@ -425,7 +426,7 @@ ExecScanUsingSemiJoinFilter(SeqScanState *node, ExprContext *econtext)
  *	represents the steps after the bloom filter is done building/checking.
  */
 void
-ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semijoin_scan)
+ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semiJoinScan, dsa_area *parallel_area)
 {
 	ListCell   *cell;
 	int			i = 0;
@@ -433,12 +434,79 @@ ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semijoin_scan)
 	foreach(cell, semijoin_filters)
 	{
 		SemiJoinFilterJoinNodeState *sjf = ((SemiJoinFilterJoinNodeState *) (lfirst(cell)));
-		SemiJoinFilterScanNodeState *sj_scan = ((SemiJoinFilterScanNodeState *) (lfirst(list_nth_cell(semijoin_scan, i))));;
+		SemiJoinFilterScanNodeState *sj_scan = ((SemiJoinFilterScanNodeState *) (lfirst(list_nth_cell(semiJoinScan, i))));;
 
 		++i;
 		if (!sjf->done_building && sj_scan->is_building_side)
 		{
-			sjf->done_building = true;
+			if (!sjf->is_parallel)
+			{
+				/*
+				 * not parallel, so only one process running and that process
+				 * is now complete
+				 */
+				sjf->done_building = true;
+			}
+			else
+			{
+				/* parallel, so need to sync with the other processes */
+				SemiJoinFilterParallelState *parallel_state =
+				(SemiJoinFilterParallelState *) dsa_get_address(parallel_area, sjf->parallel_state);
+				bloom_filter *shared_bloom = (bloom_filter *) dsa_get_address(parallel_area, parallel_state->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(&parallel_state->lock, LW_EXCLUSIVE);
+				parallel_state->elements_added += sjf->elements_added;
+				add_to_filter(shared_bloom, sjf->filter);
+				parallel_state->workers_done++;
+				LWLockRelease(&parallel_state->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 (parallel_state->workers_done == parallel_state->num_processes)
+				{
+					LWLockUpdateVar(&parallel_state->secondlock, &parallel_state->lock_stop, 1);
+				}
+				LWLockWaitForVar(&parallel_state->secondlock, &parallel_state->lock_stop, 0, &parallel_state->lock_stop);
+
+				/*
+				 * now the shared Bloom filter is fully updated, so each
+				 * individual process copies the finished Bloom filter to the
+				 * local SemiJoinFilterJoinNodeState
+				 */
+				LWLockAcquire(&parallel_state->lock, LW_EXCLUSIVE);
+				replace_bitset(sjf->filter, shared_bloom);
+				sjf->elements_added = parallel_state->elements_added;
+				sjf->done_building = true;
+				parallel_state->workers_done++;
+				LWLockRelease(&parallel_state->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 (parallel_state->workers_done == 2 * parallel_state->num_processes)
+				{
+					LWLockUpdateVar(&parallel_state->secondlock, &parallel_state->lock_stop, 2);
+				}
+				LWLockWaitForVar(&parallel_state->secondlock, &parallel_state->lock_stop, 1, &parallel_state->lock_stop);
+				/* release allocated shared memory in main process */
+				if (!sjf->is_worker)
+				{
+					LWLockRelease(&parallel_state->secondlock);
+					bloom_free_in_dsa(parallel_area, parallel_state->bloom_dsa_address);
+					dsa_free(parallel_area, sjf->parallel_state);
+				}
+			}
 		}
 	}
 }
diff --git a/src/backend/executor/nodeMergejoin.c b/src/backend/executor/nodeMergejoin.c
index 4d84ede243..f96bc2f9b6 100644
--- a/src/backend/executor/nodeMergejoin.c
+++ b/src/backend/executor/nodeMergejoin.c
@@ -98,6 +98,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"
 
@@ -657,6 +661,23 @@ 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))
+				{
+					SemiJoinFilterJoinNodeState *sjf = GetSemiJoinFilter(outerPlan, pstate->plan->plan_node_id);
+
+					PushDownFilter(innerPlan, sjf, sjf->checking_node_id, NULL);
+				}
+
 				/* Compute join values and check for unmatchability */
 				switch (MJEvalOuterValues(node))
 				{
@@ -1805,3 +1826,72 @@ PushDownFilter(PlanState *node, SemiJoinFilterJoinNodeState * sjf, int target_no
 		PushDownFilter(node->righttree, sjf, target_node_id, plan_rows);
 	}
 }
+
+dsa_pointer
+CreateFilterParallelState(dsa_area *area, SemiJoinFilterJoinNodeState * 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 *parallel_state = (SemiJoinFilterParallelState *) dsa_get_address(area, parallel_address);
+
+	/* copy over information to parallel state */
+	parallel_state->done_building = sjf->done_building;
+	parallel_state->seed = sjf->seed;
+	parallel_state->num_elements = sjf->num_elements;
+	parallel_state->work_mem = sjf->work_mem;
+	parallel_state->bloom_dsa_address = bloom_dsa_address;
+	parallel_state->sjf_num = sjf_num;
+	parallel_state->mergejoin_plan_id = sjf->mergejoin_plan_id;
+	/* initialize locks */
+	LWLockInitialize(&parallel_state->lock, LWLockNewTrancheId());
+	LWLockInitialize(&parallel_state->secondlock, LWLockNewTrancheId());
+	/* should be main process that acquires lock */
+	LWLockAcquire(&parallel_state->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.
+ */
+SemiJoinFilterJoinNodeState *
+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->apply_semijoin_filter)
+		{
+			ListCell   *lc;
+
+			foreach(lc, scan->semijoin_filters)
+			{
+				SemiJoinFilterJoinNodeState *sjf = (SemiJoinFilterJoinNodeState *) 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 8f8cc36bc9..0a5268b429 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -30,8 +30,15 @@
 #include "access/relscan.h"
 #include "access/tableam.h"
 #include "executor/execdebug.h"
+#include "executor/nodeMergejoin.h"
 #include "executor/nodeSeqscan.h"
 #include "utils/rel.h"
+#include "storage/lwlock.h"
+#include "storage/shm_toc.h"
+#include <unistd.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);
 
@@ -275,6 +282,19 @@ 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->apply_semijoin_filter)
+	{
+		shm_toc_estimate_keys(&pcxt->estimator, 1);
+		if (node->semijoin_filters)
+		{
+			shm_toc_estimate_keys(&pcxt->estimator, sizeof(dsa_pointer) * list_length(node->semijoin_filters));
+		}
+	}
 }
 
 /* ----------------------------------------------------------------
@@ -290,6 +310,50 @@ 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->apply_semijoin_filter)
+	{
+		int			sjf_num = list_length(node->semijoin_filters);
+		dsa_pointer *dsa_pointer_address;	/* actuallly 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->semijoin_filters)
+		{
+			SemiJoinFilterJoinNodeState *sjf = (SemiJoinFilterJoinNodeState *) (lfirst(lc));
+			SemiJoinFilterParallelState *parallel_state;
+			dsa_area   *area = node->ss.ps.state->es_query_dsa;
+
+			sjf->parallel_state = CreateFilterParallelState(area, sjf, sjf_num);
+			sjf->is_parallel = true;
+			/* check if main process always will run */
+			parallel_state = (SemiJoinFilterParallelState *) dsa_get_address(area, sjf->parallel_state);
+			parallel_state->num_processes = 1;
+			/* update parallel_state with built bloom filter */
+			if (sjf->done_building && node->ss.ps.plan->plan_node_id == sjf->checking_node_id)
+			{
+				bloom_filter *parallel_bloom = (bloom_filter *) dsa_get_address(area, parallel_state->bloom_dsa_address);
+
+				replace_bitset(parallel_bloom, sjf->filter);
+				LWLockRelease(&parallel_state->secondlock);
+			}
+			dsa_pointer_address[i] = sjf->parallel_state;
+			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,
@@ -327,6 +391,91 @@ ExecSeqScanInitializeWorker(SeqScanState *node,
 {
 	ParallelTableScanDesc 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->apply_semijoin_filter)
+		{
+			while (list_length(node->semijoin_filters) > 0)
+			{
+				SemiJoinFilterJoinNodeState *sjf = (SemiJoinFilterJoinNodeState *) (list_head(node->semijoin_filters)->ptr_value);
+
+				node->semijoin_filters = list_delete_nth_cell(node->semijoin_filters, 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 *parallel_state = (SemiJoinFilterParallelState *)
+			dsa_get_address(node->ss.ps.state->es_query_dsa, parallel_address);
+			SemiJoinFilterJoinNodeState *sjf;
+			MemoryContext oldContext;
+
+			sjf_num = parallel_state->sjf_num;
+			oldContext = MemoryContextSwitchTo(GetMemoryChunkContext(node));
+			sjf = (SemiJoinFilterJoinNodeState *) palloc0(sizeof(SemiJoinFilterJoinNodeState));
+			sjf->filter = bloom_create(parallel_state->num_elements, parallel_state->work_mem,
+									   parallel_state->seed);
+			sjf->building_node_id = parallel_state->building_node_id;
+			sjf->checking_node_id = parallel_state->checking_node_id;
+			sjf->seed = parallel_state->seed;
+			sjf->is_parallel = true;
+			sjf->is_worker = true;
+			sjf->done_building = parallel_state->done_building;
+			sjf->parallel_state = parallel_address;
+			node->apply_semijoin_filter = true;
+			node->semijoin_filters = lappend(node->semijoin_filters, (void *) sjf);
+			sjf->mergejoin_plan_id = parallel_state->mergejoin_plan_id;
+			/* copy over bloom filter if already built */
+			if (sjf->done_building && parallel_state->checking_node_id == node->ss.ps.plan->plan_node_id)
+			{
+				SemiJoinFilterParallelState *copy_parallel_state = (SemiJoinFilterParallelState *)
+				dsa_get_address(node->ss.ps.state->es_query_dsa, sjf->parallel_state);
+				bloom_filter *shared_bloom = (bloom_filter *) dsa_get_address(
+																			  node->ss.ps.state->es_query_dsa, copy_parallel_state->bloom_dsa_address);
+
+				replace_bitset(sjf->filter, shared_bloom);
+			}
+			else if (!sjf->done_building && parallel_state->building_node_id == node->ss.ps.plan->plan_node_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(&parallel_state->lock, LW_EXCLUSIVE);
+				parallel_state->num_processes += 1;
+				LWLockRelease(&parallel_state->lock);
+			}
+			MemoryContextSwitchTo(oldContext);
+		}
+	}
+
 	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);
diff --git a/src/backend/lib/bloomfilter.c b/src/backend/lib/bloomfilter.c
index 3ef67d35ac..743bd7388a 100644
--- a/src/backend/lib/bloomfilter.c
+++ b/src/backend/lib/bloomfilter.c
@@ -292,3 +292,77 @@ mod_m(uint32 val, uint64 m)
 
 	return val & (m - 1);
 }
+
+/*
+ * 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 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 c922fbe297..2116cb8fcd 100644
--- a/src/include/executor/nodeMergejoin.h
+++ b/src/include/executor/nodeMergejoin.h
@@ -22,5 +22,7 @@ extern void ExecReScanMergeJoin(MergeJoinState *node);
 extern void FreeSemiJoinFilter(SemiJoinFilterJoinNodeState * sjf);
 extern int	PushDownDirection(PlanState *node);
 extern void PushDownFilter(PlanState *node, SemiJoinFilterJoinNodeState * sjf, int target_node_id, int64 *plan_rows);
+extern dsa_pointer CreateFilterParallelState(dsa_area *area, SemiJoinFilterJoinNodeState * sjf, int sjf_num);
+extern SemiJoinFilterJoinNodeState * 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 9a4c3a7792..6e25dabc0b 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -48,6 +48,8 @@
 #include "utils/sortsupport.h"
 #include "utils/tuplesort.h"
 #include "utils/tuplestore.h"
+#include "storage/shm_toc.h"
+#include "storage/lwlock.h"
 
 struct PlanState;				/* forward references in this file */
 struct ParallelHashJoinState;
@@ -2023,8 +2025,40 @@ typedef struct SemiJoinFilterJoinNodeState
 	int			elements_checked;
 	int			elements_filtered;
 	int			mergejoin_plan_id;
+	/* Parallel information */
+	dsa_pointer parallel_state;
+	bool		is_parallel;
+	bool		is_worker;
 }			SemiJoinFilterJoinNodeState;
 
+/* ----------------
+ *	 SemiJoinFilterParallelState information
+ *
+ *		SemiJoinFilter information used in parallel workers.
+ * ----------------
+ */
+typedef struct SemiJoinFilterParallelState
+{
+	/* bloom filter information */
+	uint64		seed;
+	int64		num_elements;
+	int			work_mem;
+	dsa_pointer bloom_dsa_address;
+	int			elements_added;
+	/* information for parallelization and locking */
+	bool		done_building;
+	int			workers_done;
+	int			num_processes;
+	uint64		lock_stop;
+	LWLock		lock;
+	LWLock		secondlock;
+	int			sjf_num;
+	int			mergejoin_plan_id;
+
+	int			building_node_id;
+	int			checking_node_id;
+}			SemiJoinFilterParallelState;
+
 /* ----------------
  *	 MergeJoinState information
  *
-- 
2.37.1

