From f67551ba46d62b308fbf237a059c3819f4c5766e Mon Sep 17 00:00:00 2001
From: Lyu Pan <lyu.steve.pan@gmail.com>
Date: Wed, 12 Oct 2022 16:35:22 +0000
Subject: [PATCH 2/5] Support semijoin filter in the executor for non-parallel
 mergejoin.

During MergeJoinState initialization, if a semijoin filter should be used in the MergeJoin node (according to the planner), a SemiJoinFilterJoinNodeState struct is initialized and pushed down to the scan node (only SeqScan for now). The bloom filter is always built on the outer/left tree, and used in the inner/right tree.
---
 src/backend/executor/execScan.c      | 116 ++++++++++++++++++++++--
 src/backend/executor/nodeMergejoin.c | 129 ++++++++++++++++++++++++++-
 src/backend/executor/nodeSeqscan.c   |  19 ++++
 src/include/executor/nodeMergejoin.h |   3 +
 src/include/nodes/execnodes.h        |  39 ++++++++
 5 files changed, 298 insertions(+), 8 deletions(-)

diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 043bb83f55..7dd7b07a51 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -20,9 +20,13 @@
 
 #include "executor/executor.h"
 #include "miscadmin.h"
+#include "nodes/execnodes.h"
 #include "utils/memutils.h"
 
 
+bool		ExecScanUsingSemiJoinFilter(SeqScanState *node, ExprContext *econtext);
+void		ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semijoin_scan);
+
 
 /*
  * ExecScanFetch -- check interrupts & fetch next potential tuple
@@ -173,10 +177,11 @@ ExecScan(ScanState *node,
 	/* interrupt checks are in ExecScanFetch */
 
 	/*
-	 * If we have neither a qual to check nor a projection to do, just skip
-	 * all the overhead and return the raw scan tuple.
+	 * If we have neither a qual to check nor a projection to do nor a Bloom
+	 * filter to check, just skip all the overhead and return the raw scan
+	 * tuple.
 	 */
-	if (!qual && !projInfo)
+	if (!qual && !projInfo && !(IsA(node, SeqScanState) && ((SeqScanState *) node)->apply_semijoin_filter))
 	{
 		ResetExprContext(econtext);
 		return ExecScanFetch(node, accessMtd, recheckMtd);
@@ -206,6 +211,10 @@ ExecScan(ScanState *node,
 		 */
 		if (TupIsNull(slot))
 		{
+			if (IsA(node, SeqScanState) && ((SeqScanState *) node)->apply_semijoin_filter)
+			{
+				ExecSemiJoinFilterFinishScan(((SeqScanState *) node)->semijoin_filters, ((SeqScanState *) node)->sj_scan_data);
+			}
 			if (projInfo)
 				return ExecClearTuple(projInfo->pi_state.resultslot);
 			else
@@ -232,16 +241,33 @@ ExecScan(ScanState *node,
 			if (projInfo)
 			{
 				/*
-				 * Form a projection tuple, store it in the result tuple slot
-				 * and return it.
+				 * Form a projection tuple, store it in the result tuple slot,
+				 * check against SemiJoinFilter, then return it.
 				 */
-				return ExecProject(projInfo);
+				TupleTableSlot *projectedSlot = ExecProject(projInfo);
+
+				if (IsA(node, SeqScanState) && ((SeqScanState *) node)->apply_semijoin_filter
+					&& !ExecScanUsingSemiJoinFilter((SeqScanState *) node, econtext))
+				{
+					/* slot did not pass SemiJoinFilter, so skipping it. */
+					ResetExprContext(econtext);
+					continue;
+				}
+				return projectedSlot;
 			}
 			else
 			{
 				/*
-				 * Here, we aren't projecting, so just return scan tuple.
+				 * Here, we aren't projecting, so check against
+				 * SemiJoinFilter, then return tuple.
 				 */
+				if (IsA(node, SeqScanState) && ((SeqScanState *) node)->apply_semijoin_filter
+					&& !ExecScanUsingSemiJoinFilter((SeqScanState *) node, econtext))
+				{
+					/* slot did not pass SemiJoinFilter, so skipping it. */
+					ResetExprContext(econtext);
+					continue;
+				}
 				return slot;
 			}
 		}
@@ -340,3 +366,79 @@ ExecScanReScan(ScanState *node)
 		}
 	}
 }
+
+/*
+ *	ExecScanUsingSemiJoinFilter
+ *
+ *	This must be called within ExecScan function of SeqScanState node. This
+ *	function builds/checks the semijoin filter (bloom filter) in the node. If
+ *	this node is the building node of the bloom filter, it always returns true;
+ *	if this node is the checking node of the bloom filter, it returns true if
+ *	the value is in the bloom filter and returns false if not.
+ */
+bool
+ExecScanUsingSemiJoinFilter(SeqScanState *node, ExprContext *econtext)
+{
+	List	   *sj_scan_data;
+	List	   *sj_filters;
+	int			i;
+	ListCell   *lc;
+	SemiJoinFilterJoinNodeState *sjf;
+	SemiJoinFilterScanNodeState *sj_scan;
+	Datum		data;
+	bool		is_null;
+
+	sj_scan_data = node->sj_scan_data;
+	sj_filters = node->semijoin_filters;
+
+	Assert(list_length(sj_scan_data) == list_length(sj_filters));
+
+	i = 0;
+	foreach(lc, sj_scan_data)
+	{
+		is_null = false;
+		sjf = ((SemiJoinFilterJoinNodeState *) (lfirst(list_nth_cell(sj_filters, i))));
+		sj_scan = (SemiJoinFilterScanNodeState *) lfirst(lc);
+
+		data = ExecEvalExpr(sj_scan->expr_state, econtext, &is_null);
+
+		if (!sjf->done_building && sj_scan->is_building_side && !is_null)
+		{
+			bloom_add_element(sjf->filter, (unsigned char *) &data, sizeof(data));
+			++(sjf->elements_added);
+		}
+		else if (sjf->done_building && !sj_scan->is_building_side && !is_null &&
+				 bloom_lacks_element(sjf->filter, (unsigned char *) &data, sizeof(data)))
+		{
+			++(sjf->elements_filtered);
+			return false;
+		}
+		++i;
+	}
+	return true;
+}
+
+/*
+ *	ExecSemiJoinFilterFinishScan
+ *
+ *	This must be called within ExecScan function of SeqScanState node, and
+ *	represents the steps after the bloom filter is done building/checking.
+ */
+void
+ExecSemiJoinFilterFinishScan(List *semijoin_filters, List *semijoin_scan)
+{
+	ListCell   *cell;
+	int			i = 0;
+
+	foreach(cell, semijoin_filters)
+	{
+		SemiJoinFilterJoinNodeState *sjf = ((SemiJoinFilterJoinNodeState *) (lfirst(cell)));
+		SemiJoinFilterScanNodeState *sj_scan = ((SemiJoinFilterScanNodeState *) (lfirst(list_nth_cell(semijoin_scan, i))));;
+
+		++i;
+		if (!sjf->done_building && sj_scan->is_building_side)
+		{
+			sjf->done_building = true;
+		}
+	}
+}
diff --git a/src/backend/executor/nodeMergejoin.c b/src/backend/executor/nodeMergejoin.c
index fed345eae5..4d84ede243 100644
--- a/src/backend/executor/nodeMergejoin.c
+++ b/src/backend/executor/nodeMergejoin.c
@@ -93,13 +93,14 @@
 #include "postgres.h"
 
 #include "access/nbtree.h"
+#include "common/pg_prng.h"
 #include "executor/execdebug.h"
 #include "executor/nodeMergejoin.h"
+#include "lib/bloomfilter.h"
 #include "miscadmin.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 
-
 /*
  * States of the ExecMergeJoin state machine
  */
@@ -1603,6 +1604,40 @@ ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags)
 											node->mergeNullsFirst,
 											(PlanState *) mergestate);
 
+	/*
+	 * initialize SemiJoinFilter, if planner decided to do so
+	 */
+	if (node->sj_metadata && node->sj_metadata->apply_semijoin_filter)
+	{
+		SemiJoinFilterJoinNodeState *sjf;
+		uint64		seed;
+
+		/* create Bloom filter */
+		sjf = (SemiJoinFilterJoinNodeState *) palloc0(sizeof(SemiJoinFilterJoinNodeState));
+
+		/*
+		 * Push down filter down outer and inner subtrees and apply filter to
+		 * the nodes that correspond to the ones identified during planning.
+		 * We are pushing down first because we need some metadata from the
+		 * scan nodes (i.e. Relation Id's and planner-estimated number of
+		 * rows).
+		 */
+		sjf->building_node_id = ((MergeJoin *) mergestate->js.ps.plan)->sj_metadata->building_node_id;
+		sjf->checking_node_id = ((MergeJoin *) mergestate->js.ps.plan)->sj_metadata->checking_node_id;
+		PushDownFilter(mergestate->js.ps.lefttree, sjf, sjf->building_node_id, &sjf->num_elements);
+		PushDownFilter(mergestate->js.ps.righttree, sjf, sjf->checking_node_id, &sjf->num_elements);
+
+		/* Initialize SJF data and create Bloom filter */
+		seed = pg_prng_uint64(&pg_global_prng_state);
+		sjf->filter = bloom_create(sjf->num_elements, work_mem, seed);
+		sjf->work_mem = work_mem;
+		sjf->seed = seed;
+		sjf->done_building = false;
+
+		sjf->mergejoin_plan_id = mergestate->js.ps.plan->plan_node_id;
+		mergestate->sjf = sjf;
+	}
+
 	/*
 	 * initialize join state
 	 */
@@ -1645,6 +1680,14 @@ ExecEndMergeJoin(MergeJoinState *node)
 	ExecClearTuple(node->js.ps.ps_ResultTupleSlot);
 	ExecClearTuple(node->mj_MarkedTupleSlot);
 
+	/*
+	 * free SemiJoinFilterJoinNodeState
+	 */
+	if (node->sjf)
+	{
+		FreeSemiJoinFilter(node->sjf);
+	}
+
 	/*
 	 * shut down the subplans
 	 */
@@ -1678,3 +1721,87 @@ ExecReScanMergeJoin(MergeJoinState *node)
 	if (innerPlan->chgParam == NULL)
 		ExecReScan(innerPlan);
 }
+
+void
+FreeSemiJoinFilter(SemiJoinFilterJoinNodeState * sjf)
+{
+	bloom_free(sjf->filter);
+	pfree(sjf);
+}
+
+/*
+ * Determines the direction that a pushdown filter can be pushed. This is not very robust, but this
+ * is because we've already done the careful calculations at the plan level. If we end up pushing where
+ * we're not supposed to, it's fine because we've done the verifications in the planner.
+ */
+int
+PushDownDirection(PlanState *node)
+{
+	switch (nodeTag(node))
+	{
+		case T_HashState:
+		case T_MaterialState:
+		case T_GatherState:
+		case T_GatherMergeState:
+		case T_SortState:
+		case T_UniqueState:
+		case T_AggState:
+			{
+				return 0;
+			}
+		case T_NestLoopState:
+		case T_MergeJoinState:
+		case T_HashJoinState:
+			{
+				return 1;
+			}
+		default:
+			{
+				return -1;
+			}
+	}
+}
+
+/* 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, SemiJoinFilterJoinNodeState * sjf, int target_node_id, int64 *plan_rows)
+{
+	if (node == NULL)
+	{
+		return;
+	}
+
+	check_stack_depth();
+	if (node->type == T_SeqScanState)
+	{
+		SeqScanState *scan = (SeqScanState *) node;
+
+		/*
+		 * found the right Scan node that we want to apply the filter onto via
+		 * matching plan id.
+		 */
+		if (scan->ss.ps.plan->plan_node_id == target_node_id)
+		{
+			scan->apply_semijoin_filter = true;
+			scan->semijoin_filters = lappend(scan->semijoin_filters, sjf);
+			/* double row estimate to reduce error rate for Bloom filter */
+			if (plan_rows)
+			{
+				*plan_rows = Max(*plan_rows, scan->ss.ps.plan->plan_rows * 2);
+			}
+		}
+		return;
+	}
+	if (PushDownDirection(node) == 0)
+	{
+		PushDownFilter(node->lefttree, sjf, target_node_id, plan_rows);
+		return;
+	}
+	if (PushDownDirection(node) == 1)
+	{
+		PushDownFilter(node->lefttree, sjf, target_node_id, plan_rows);
+		PushDownFilter(node->righttree, sjf, target_node_id, plan_rows);
+	}
+}
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index 7b58cd9162..8f8cc36bc9 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -171,6 +171,25 @@ ExecInitSeqScan(SeqScan *node, EState *estate, int eflags)
 	scanstate->ss.ps.qual =
 		ExecInitQual(node->scan.plan.qual, (PlanState *) scanstate);
 
+	/*
+	 * Initialize semijoin filter expressions.
+	 */
+	if (node->scan.plan.sj_md_list)
+	{
+		ListCell   *lc;
+
+		scanstate->sj_scan_data = NIL;
+		foreach(lc, node->scan.plan.sj_md_list)
+		{
+			SemijoinFilterScanData *md = (SemijoinFilterScanData *) lfirst(lc);
+			SemiJoinFilterScanNodeState *sj_filter = (SemiJoinFilterScanNodeState *) palloc0(sizeof(SemiJoinFilterScanNodeState));
+
+			sj_filter->is_building_side = md->is_building_node;
+			sj_filter->expr_state = ExecInitExpr(lfirst(list_nth_cell(md->semijoin_keys, 0)), (PlanState *) scanstate);
+			scanstate->sj_scan_data = lappend(scanstate->sj_scan_data, sj_filter);
+		}
+	}
+
 	return scanstate;
 }
 
diff --git a/src/include/executor/nodeMergejoin.h b/src/include/executor/nodeMergejoin.h
index 26ab517508..c922fbe297 100644
--- a/src/include/executor/nodeMergejoin.h
+++ b/src/include/executor/nodeMergejoin.h
@@ -19,5 +19,8 @@
 extern MergeJoinState *ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags);
 extern void ExecEndMergeJoin(MergeJoinState *node);
 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);
 
 #endif							/* NODEMERGEJOIN_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 01b1727fc0..9a4c3a7792 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -32,6 +32,7 @@
 #include "access/tupconvert.h"
 #include "executor/instrument.h"
 #include "fmgr.h"
+#include "lib/bloomfilter.h"
 #include "lib/ilist.h"
 #include "lib/pairingheap.h"
 #include "nodes/params.h"
@@ -1451,6 +1452,10 @@ typedef struct SeqScanState
 {
 	ScanState	ss;				/* its first field is NodeTag */
 	Size		pscan_len;		/* size of parallel heap scan descriptor */
+	/* for use of SemiJoinFilter during merge join */
+	bool		apply_semijoin_filter;
+	List	   *semijoin_filters;	/* SemiJoinFilterJoinNodeState	*/
+	List	   *sj_scan_data;	/* SemiJoinFilterScanNodeState */
 } SeqScanState;
 
 /* ----------------
@@ -1987,6 +1992,39 @@ typedef struct NestLoopState
 	TupleTableSlot *nl_NullInnerTupleSlot;
 } NestLoopState;
 
+/* ----------------
+ *	 SemiJoinFilterScanNodeState information
+ *
+ *		SemiJoinFilter information used in Scan node.
+ * ----------------
+ */
+typedef struct SemiJoinFilterScanNodeState
+{
+	bool		is_building_side;
+	ExprState  *expr_state;
+}			SemiJoinFilterScanNodeState;
+
+/* ----------------
+ *	 SemiJoinFilterJoinNodeState information
+ *
+ *		SemiJoinFilter information used in Scan node.
+ * ----------------
+ */
+typedef struct SemiJoinFilterJoinNodeState
+{
+	bloom_filter *filter;
+	int			building_node_id;
+	int			checking_node_id;
+	bool		done_building;
+	uint64		seed;
+	int64		num_elements;
+	int			work_mem;
+	int			elements_added;
+	int			elements_checked;
+	int			elements_filtered;
+	int			mergejoin_plan_id;
+}			SemiJoinFilterJoinNodeState;
+
 /* ----------------
  *	 MergeJoinState information
  *
@@ -2032,6 +2070,7 @@ typedef struct MergeJoinState
 	TupleTableSlot *mj_NullInnerTupleSlot;
 	ExprContext *mj_OuterEContext;
 	ExprContext *mj_InnerEContext;
+	SemiJoinFilterJoinNodeState *sjf;
 } MergeJoinState;
 
 /* ----------------
-- 
2.37.1

