From 071e35b4fc038cd791479beb6fcbe8b0a4c0bee8 Mon Sep 17 00:00:00 2001
From: Lyu Pan <lyup@amazon.com>
Date: Fri, 16 Sep 2022 03:35:06 +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 SemiJoinFilter struct is initialized, then the relation id and attribute number used to build/check the bloom filter are calculated, related information is 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      |  52 +++++-
 src/backend/executor/nodeMergejoin.c | 259 +++++++++++++++++++++++++++
 src/backend/executor/nodeSeqscan.c   |   6 +
 src/include/executor/nodeMergejoin.h |   5 +
 src/include/nodes/execnodes.h        |  25 +++
 5 files changed, 340 insertions(+), 7 deletions(-)

diff --git a/src/backend/executor/execScan.c b/src/backend/executor/execScan.c
index 043bb83f55..4b97c39455 100644
--- a/src/backend/executor/execScan.c
+++ b/src/backend/executor/execScan.c
@@ -19,8 +19,10 @@
 #include "postgres.h"
 
 #include "executor/executor.h"
+#include "executor/nodeMergejoin.h"
 #include "miscadmin.h"
 #include "utils/memutils.h"
+#include "utils/rel.h"
 
 
 
@@ -173,10 +175,12 @@ 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)->applySemiJoinFilter)
 	{
 		ResetExprContext(econtext);
 		return ExecScanFetch(node, accessMtd, recheckMtd);
@@ -206,6 +210,13 @@ ExecScan(ScanState *node,
 		 */
 		if (TupIsNull(slot))
 		{
+			if (IsA(node, SeqScanState) &&
+				((SeqScanState *) node)->applySemiJoinFilter)
+			{
+				SemiJoinFilterFinishScan(
+										 ((SeqScanState *) node)->semiJoinFilters,
+										 node->ss_currentRelation->rd_id);
+			}
 			if (projInfo)
 				return ExecClearTuple(projInfo->pi_state.resultslot);
 			else
@@ -232,16 +243,43 @@ 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)->applySemiJoinFilter)
+				{
+					if (!SemiJoinFilterExamineSlot(
+												   ((SeqScanState *) node)->semiJoinFilters,
+												   projectedSlot, node->ss_currentRelation->rd_id))
+					{
+						/* 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)->applySemiJoinFilter)
+				{
+					if (!SemiJoinFilterExamineSlot(
+												   ((SeqScanState *) node)->semiJoinFilters, slot,
+												   node->ss_currentRelation->rd_id))
+					{
+						/* slot did not pass SemiJoinFilter, so skipping it. */
+						ResetExprContext(econtext);
+						continue;
+					}
+				}
 				return slot;
 			}
 		}
diff --git a/src/backend/executor/nodeMergejoin.c b/src/backend/executor/nodeMergejoin.c
index fed345eae5..81c253aad5 100644
--- a/src/backend/executor/nodeMergejoin.c
+++ b/src/backend/executor/nodeMergejoin.c
@@ -93,8 +93,11 @@
 #include "postgres.h"
 
 #include "access/nbtree.h"
+#include "common/pg_prng.h"
 #include "executor/execdebug.h"
+#include "executor/execExpr.h"
 #include "executor/nodeMergejoin.h"
+#include "lib/bloomfilter.h"
 #include "miscadmin.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -1603,6 +1606,95 @@ ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags)
 											node->mergeNullsFirst,
 											(PlanState *) mergestate);
 
+	/*
+	 * initialize SemiJoinFilter, if planner decided to do so
+	 */
+	if (((MergeJoin *) mergestate->js.ps.plan)->applySemiJoinFilter)
+	{
+		SemiJoinFilter *sjf;
+		Plan	   *buildingNode;
+		Plan	   *checkingNode;
+		uint64		seed;
+		MergeJoinClause clause;
+
+		/* create Bloom filter */
+		sjf = (SemiJoinFilter *) palloc0(sizeof(SemiJoinFilter));
+
+		/*
+		 * 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).
+		 */
+		buildingNode = ((MergeJoin *) mergestate->js.ps.plan)->buildingNode;
+		checkingNode = ((MergeJoin *) mergestate->js.ps.plan)->checkingNode;
+		sjf->buildingId = -1;
+		sjf->checkingId = -1;
+		PushDownFilter(mergestate->js.ps.lefttree, buildingNode, sjf, &sjf->buildingId, &sjf->num_elements);
+		PushDownFilter(mergestate->js.ps.righttree, checkingNode, sjf, &sjf->checkingId, &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->doneBuilding = false;
+
+		/*
+		 * From the plan level, we already know which mergeclause to build the
+		 * filter on. However, to implement this on the scan level, we look at
+		 * the expression and figure out which slot we need to examine in
+		 * ExecScan to build/check the filter on.
+		 */
+
+		clause = &mergestate->mj_Clauses[((MergeJoin *) mergestate->js.ps.plan)->bestExpr];
+
+		/*
+		 * Look through expression steps to determine which relation attribute
+		 * slot they are comparing and take note of it. All merge join clauses
+		 * eventually fetch tuples from either the outer or inner slot, so we
+		 * just need to check for those specific ExprEvalOp's.
+		 */
+		for (int j = 0; j < clause->lexpr->steps_len; j++)
+		{
+			ExprEvalOp	leftOp = clause->lexpr->steps[j].opcode;
+			int			leftAttr = clause->lexpr->steps[j].d.var.attnum;
+
+			if (leftOp == EEOP_OUTER_FETCHSOME)
+			{
+				/* attribute numbers are 1-indexed */
+				sjf->buildingAttr = leftAttr - 1;
+				break;
+			}
+			else if (leftOp == EEOP_INNER_FETCHSOME)
+			{
+				sjf->checkingAttr = leftAttr - 1;
+				break;
+			}
+		}
+		/* do it again for right expression */
+		for (int j = 0; j < clause->rexpr->steps_len; j++)
+		{
+			ExprEvalOp	rightOp = clause->rexpr->steps[j].opcode;
+			int			rightAttr = clause->rexpr->steps[j].d.var.attnum;
+
+			if (rightOp == EEOP_OUTER_FETCHSOME)
+			{
+				/* attribute numbers are 1-indexed */
+				sjf->buildingAttr = rightAttr - 1;
+				break;
+			}
+			else if (rightOp == EEOP_INNER_FETCHSOME)
+			{
+				sjf->checkingAttr = rightAttr - 1;
+				break;
+			}
+		}
+		sjf->mergejoin_plan_id = mergestate->js.ps.plan->plan_node_id;
+		mergestate->sjf = sjf;
+	}
+
 	/*
 	 * initialize join state
 	 */
@@ -1645,6 +1737,14 @@ ExecEndMergeJoin(MergeJoinState *node)
 	ExecClearTuple(node->js.ps.ps_ResultTupleSlot);
 	ExecClearTuple(node->mj_MarkedTupleSlot);
 
+	/*
+	 * free SemiJoinFilter
+	 */
+	if (node->sjf)
+	{
+		FreeSemiJoinFilter(node->sjf);
+	}
+
 	/*
 	 * shut down the subplans
 	 */
@@ -1678,3 +1778,162 @@ ExecReScanMergeJoin(MergeJoinState *node)
 	if (innerPlan->chgParam == NULL)
 		ExecReScan(innerPlan);
 }
+
+void
+FreeSemiJoinFilter(SemiJoinFilter * 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, Plan *plan, SemiJoinFilter * sjf, Oid *relId, int64 *nodeRows)
+{
+	if (node == NULL)
+	{
+		return;
+	}
+
+	check_stack_depth();
+	if (node->type == T_SeqScanState)
+	{
+		SeqScanState *scan = (SeqScanState *) node;
+
+		Assert(IsA(scan, SeqScanState));
+
+		/*
+		 * found the right Scan node that we want to apply the filter onto via
+		 * matching relId
+		 */
+		if (scan->ss.ss_currentRelation->rd_id == *relId)
+		{
+			scan->applySemiJoinFilter = true;
+			scan->semiJoinFilters = lappend(scan->semiJoinFilters, sjf);
+		}
+
+		/*
+		 * Check if right Scan node, based on matching Plan nodes. This will
+		 * be the most common way of matching Scan nodes to the filter, the
+		 * above use case is only for fringe parallel-execution cases.
+		 */
+		else if (scan->ss.ps.plan == plan)
+		{
+			scan->applySemiJoinFilter = true;
+			scan->semiJoinFilters = lappend(scan->semiJoinFilters, sjf);
+			*relId = scan->ss.ss_currentRelation->rd_id;
+			/* double row estimate to reduce error rate for Bloom filter */
+			*nodeRows = Max(*nodeRows, scan->ss.ps.plan->plan_rows * 2);
+		}
+	}
+	else
+	{
+		if (PushDownDirection(node) == 1)
+		{
+			PushDownFilter(node->lefttree, plan, sjf, relId, nodeRows);
+			PushDownFilter(node->righttree, plan, sjf, relId, nodeRows);
+		}
+		else if (PushDownDirection(node) == 0)
+		{
+			PushDownFilter(node->lefttree, plan, sjf, relId, nodeRows);
+		}
+	}
+}
+
+/*
+ * 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)
+{
+	ListCell   *cell;
+
+	foreach(cell, semiJoinFilters)
+	{
+		SemiJoinFilter *sjf = ((SemiJoinFilter *) (lfirst(cell)));
+
+		/* check if this table's relation ID matches the filter's relation ID */
+		if (!sjf->doneBuilding && tableId == sjf->buildingId)
+		{
+			Datum		val;
+
+			slot_getsomeattrs(slot, sjf->buildingAttr + 1);
+			sjf->elementsAdded++;
+
+			/*
+			 * We are only using one key for now. Later functionality might
+			 * include multiple join keys.
+			 */
+			val = slot->tts_values[sjf->buildingAttr];
+			bloom_add_element(sjf->filter, (unsigned char *) &val, sizeof(val));
+		}
+		else if (sjf->doneBuilding && tableId == sjf->checkingId)
+		{
+			Datum		val;
+
+			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)))
+			{
+				sjf->elementsFiltered++;
+				return false;
+			}
+		}
+	}
+	return true;
+}
+
+void
+SemiJoinFilterFinishScan(List *semiJoinFilters, Oid tableId)
+{
+	ListCell   *cell;
+
+	foreach(cell, semiJoinFilters)
+	{
+		SemiJoinFilter *sjf = ((SemiJoinFilter *) (lfirst(cell)));
+
+		if (!sjf->doneBuilding && tableId == sjf->buildingId)
+		{
+			sjf->doneBuilding = true;
+		}
+	}
+}
diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c
index 7b58cd9162..e43ce3f8d0 100644
--- a/src/backend/executor/nodeSeqscan.c
+++ b/src/backend/executor/nodeSeqscan.c
@@ -207,6 +207,12 @@ ExecEndSeqScan(SeqScanState *node)
 	 */
 	if (scanDesc != NULL)
 		table_endscan(scanDesc);
+
+	/*
+	 * clear out semijoinfilter list, if non-null
+	 */
+	if (node->semiJoinFilters)
+		pfree(node->semiJoinFilters);
 }
 
 /* ----------------------------------------------------------------
diff --git a/src/include/executor/nodeMergejoin.h b/src/include/executor/nodeMergejoin.h
index 26ab517508..c311c7ed80 100644
--- a/src/include/executor/nodeMergejoin.h
+++ b/src/include/executor/nodeMergejoin.h
@@ -19,5 +19,10 @@
 extern MergeJoinState *ExecInitMergeJoin(MergeJoin *node, EState *estate, int eflags);
 extern void ExecEndMergeJoin(MergeJoinState *node);
 extern void ExecReScanMergeJoin(MergeJoinState *node);
+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);
 
 #endif							/* NODEMERGEJOIN_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 01b1727fc0..6964462720 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,9 @@ typedef struct SeqScanState
 {
 	ScanState	ss;				/* its first field is NodeTag */
 	Size		pscan_len;		/* size of parallel heap scan descriptor */
+	/* for use of SemiJoinFilters during merge join */
+	bool		applySemiJoinFilter;
+	List	   *semiJoinFilters;
 } SeqScanState;
 
 /* ----------------
@@ -2012,6 +2016,26 @@ typedef struct NestLoopState
 /* private in nodeMergejoin.c: */
 typedef struct MergeJoinClauseData *MergeJoinClause;
 
+typedef struct SemiJoinFilter
+{
+	bloom_filter *filter;
+	/* Relation that Bloom Filter is built on */
+	Oid			buildingId;
+	int			buildingAttr;
+	/* Relation that Bloom Filter is checked on */
+	Oid			checkingId;
+	int			checkingAttr;
+	bool		doneBuilding;
+	/* metadata */
+	uint64		seed;
+	int64		num_elements;
+	int			work_mem;
+	int			elementsAdded;
+	int			elementsChecked;
+	int			elementsFiltered;
+	int			mergejoin_plan_id;
+}			SemiJoinFilter;
+
 typedef struct MergeJoinState
 {
 	JoinState	js;				/* its first field is NodeTag */
@@ -2032,6 +2056,7 @@ typedef struct MergeJoinState
 	TupleTableSlot *mj_NullInnerTupleSlot;
 	ExprContext *mj_OuterEContext;
 	ExprContext *mj_InnerEContext;
+	SemiJoinFilter *sjf;
 } MergeJoinState;
 
 /* ----------------
-- 
2.37.1

