Push the runtime filter from HashJoin down to SeqScan.

+----------+  AttrFilter   +------+  ScanKey   +---------+
| HashJoin | ------------> | Hash | ---------> | SeqScan |
+----------+               +------+            +---------+

If "gp_enable_runtime_filter_pushdown" is on, three steps will be run:

Step 1. In ExecInitHashJoin(), try to find the mapper between the var in
        hashclauses and the var in SeqScan. If found we will save the mapper in
        AttrFilter and push them to Hash node;

Step 2. We will create the range/bloom filters in AttrFilter during building
        hash table, and these filters will be converted to the list of ScanKey
        and pushed down to Seqscan when the building finishes;

Step 3. ScanKeys will be used to filter slot in Seqscan.

TODO:
1. support singlenode or utility mode;
2. support null value filter;
3. support Motion, SharedScan as target node;
4. support join qual like: t1.c1 = t2.c1+5;

Author: zhangyue-hashdata

src/backend/commands/explain.c

@@ -139,6 +139,9 @@ static void show_incremental_sort_info(IncrementalSortState *incrsortstate,
 static void show_hash_info(HashState *hashstate, ExplainState *es);
 static void show_runtime_filter_info(RuntimeFilterState *rfstate,
 									 ExplainState *es);
+static void show_pushdown_runtime_filter_info(const char *qlabel,
+											  PlanState *planstate,
+											  ExplainState *es);
 static void show_memoize_info(MemoizeState *mstate, List *ancestors,
 							  ExplainState *es);
 static void show_hashagg_info(AggState *hashstate, ExplainState *es);
@@ -2495,6 +2498,10 @@ ExplainNode(PlanState *planstate, List *ancestors,
 			/* fall through to print additional fields the same as SeqScan */
 			/* FALLTHROUGH */
 		case T_SeqScan:
+			if (gp_enable_runtime_filter_pushdown && IsA(planstate, SeqScanState))
+				show_pushdown_runtime_filter_info("Rows Removed by Pushdown Runtime Filter",
+												  planstate, es);
+			/* FALLTHROUGH */
 		case T_DynamicSeqScan:
 		case T_ValuesScan:
 		case T_CteScan:
@@ -4365,6 +4372,24 @@ show_instrumentation_count(const char *qlabel, int which,
 	}
 }
 
+/*
+ * If it's EXPLAIN ANALYZE, show instrumentation information with pushdown
+ * runtime filter.
+ */
+static void
+show_pushdown_runtime_filter_info(const char *qlabel,
+								  PlanState *planstate,
+								  ExplainState *es)
+{
+	Assert(gp_enable_runtime_filter_pushdown && IsA(planstate, SeqScanState));
+
+	if (!es->analyze || !planstate->instrument)
+		return;
+
+	if (planstate->instrument->prf_work)
+		ExplainPropertyFloat(qlabel, NULL, planstate->instrument->nfilteredPRF, 0, es);
+}
+
 /*
  * Show extra information for a ForeignScan node.
  */

src/backend/commands/explain_gp.c

@@ -56,6 +56,8 @@ typedef struct CdbExplain_StatInst
 	double		nloops;			/* # of run cycles for this node */
 	double		nfiltered1;
 	double		nfiltered2;
+	bool		prf_work;
+	double		nfilteredPRF;
 	double		execmemused;	/* executor memory used (bytes) */
 	double		workmemused;	/* work_mem actually used (bytes) */
 	double		workmemwanted;	/* work_mem to avoid workfile i/o (bytes) */
@@ -885,6 +887,8 @@ cdbexplain_collectStatsFromNode(PlanState *planstate, CdbExplain_SendStatCtx *ct
 	si->nloops = instr->nloops;
 	si->nfiltered1 = instr->nfiltered1;
 	si->nfiltered2 = instr->nfiltered2;
+	si->prf_work     = instr->prf_work;
+	si->nfilteredPRF = instr->nfilteredPRF;
 	si->workmemused = instr->workmemused;
 	si->workmemwanted = instr->workmemwanted;
 	si->workfileCreated = instr->workfileCreated;
@@ -1188,6 +1192,8 @@ cdbexplain_depositStatsToNode(PlanState *planstate, CdbExplain_RecvStatCtx *ctx)
 		instr->nloops = nodeAcc->nsimax->nloops;
 		instr->nfiltered1 = nodeAcc->nsimax->nfiltered1;
 		instr->nfiltered2 = nodeAcc->nsimax->nfiltered2;
+		instr->prf_work     = nodeAcc->nsimax->prf_work;
+		instr->nfilteredPRF = nodeAcc->nsimax->nfilteredPRF;
 		instr->execmemused = nodeAcc->nsimax->execmemused;
 		instr->workmemused = nodeAcc->nsimax->workmemused;
 		instr->workmemwanted = nodeAcc->nsimax->workmemwanted;

src/backend/executor/nodeHash.c

@@ -101,7 +101,10 @@ static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable,
 										  size_t size);
 static void ExecParallelHashMergeCounters(HashJoinTable hashtable);
 static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable);
-
+static void AddTupleValuesIntoRF(HashState *node, TupleTableSlot *slot);
+static void PushdownRuntimeFilter(HashState *node);
+static void FreeRuntimeFilter(HashState *node);
+static void ResetRuntimeFilter(HashState *node);
 
 /* ----------------------------------------------------------------
  *		ExecHash
@@ -192,7 +195,15 @@ MultiExecPrivateHash(HashState *node)
 	{
 		slot = ExecProcNode(outerNode);
 		if (TupIsNull(slot))
+		{
+			if (gp_enable_runtime_filter_pushdown && node->filters)
+				PushdownRuntimeFilter(node);
 			break;
+		}
+
+		if (gp_enable_runtime_filter_pushdown && node->filters)
+			AddTupleValuesIntoRF(node, slot);
+
 		/* We have to compute the hash value */
 		econtext->ecxt_outertuple = slot;
 		bool hashkeys_null = false;
@@ -334,6 +345,7 @@ MultiExecParallelHash(HashState *node)
 				slot = ExecProcNode(outerNode);
 				if (TupIsNull(slot))
 					break;
+
 				econtext->ecxt_outertuple = slot;
 				if (ExecHashGetHashValue(node, hashtable, econtext, hashkeys,
 										 false, hashtable->keepNulls,
@@ -511,6 +523,9 @@ ExecEndHash(HashState *node)
 	 */
 	outerPlan = outerPlanState(node);
 	ExecEndNode(outerPlan);
+
+	if (node->filters)
+		FreeRuntimeFilter(node);
 }
 
 
@@ -2519,6 +2534,9 @@ ExecReScanHash(HashState *node)
 	 */
 	if (node->ps.lefttree->chgParam == NULL)
 		ExecReScan(node->ps.lefttree);
+
+	if (gp_enable_runtime_filter_pushdown && node->filters)
+		ResetRuntimeFilter(node);
 }
 
 
@@ -4125,3 +4143,142 @@ get_hash_mem(void)
 
 	return (int) mem_limit;
 }
+
+/*
+ * Convert AttrFilter to ScanKeyData and send these runtime filters to the
+ * target node(seqscan).
+ */
+void
+PushdownRuntimeFilter(HashState *node)
+{
+	ListCell	*lc;
+	List		*scankeys;
+	ScanKey		sk;
+	AttrFilter	*attr_filter;
+
+	foreach (lc, node->filters)
+	{
+		scankeys = NIL;
+
+		attr_filter = lfirst(lc);
+		if (!IsA(attr_filter->target, SeqScanState) || attr_filter->empty)
+			continue;
+
+		/* bloom filter */
+		sk = (ScanKey)palloc0(sizeof(ScanKeyData));
+		sk->sk_flags    = SK_BLOOM_FILTER;
+		sk->sk_attno    = attr_filter->lattno;
+		sk->sk_subtype  = INT8OID;
+		sk->sk_argument = PointerGetDatum(attr_filter->blm_filter);
+		scankeys = lappend(scankeys, sk);
+
+		/* range filter */
+		sk = (ScanKey)palloc0(sizeof(ScanKeyData));
+		sk->sk_flags    = 0;
+		sk->sk_attno    = attr_filter->lattno;
+		sk->sk_strategy = BTGreaterEqualStrategyNumber;
+		sk->sk_subtype  = INT8OID;
+		sk->sk_argument = attr_filter->min;
+		scankeys = lappend(scankeys, sk);
+
+		sk = (ScanKey)palloc0(sizeof(ScanKeyData));
+		sk->sk_flags    = 0;
+		sk->sk_attno    = attr_filter->lattno;
+		sk->sk_strategy = BTLessEqualStrategyNumber;
+		sk->sk_subtype  = INT8OID;
+		sk->sk_argument = attr_filter->max;
+		scankeys = lappend(scankeys, sk);
+
+		/* append new runtime filters to target node */
+		SeqScanState *sss = castNode(SeqScanState, attr_filter->target);
+		sss->filters = list_concat(sss->filters, scankeys);
+	}
+}
+
+static void
+AddTupleValuesIntoRF(HashState *node, TupleTableSlot *slot)
+{
+	Datum val;
+	bool  isnull;
+	ListCell	*lc;
+	AttrFilter	*attr_filter;
+
+	foreach (lc, node->filters)
+	{
+		attr_filter = (AttrFilter *) lfirst(lc);
+
+		val = slot_getattr(slot, attr_filter->rattno, &isnull);
+		if (isnull)
+			continue;
+
+		attr_filter->empty = false;
+
+		if ((int64_t)val < (int64_t)attr_filter->min)
+			attr_filter->min = val;
+
+		if ((int64_t)val > (int64_t)attr_filter->max)
+			attr_filter->max = val;
+
+		if (attr_filter->blm_filter)
+			bloom_add_element(attr_filter->blm_filter, (unsigned char *)&val, sizeof(Datum));
+	}
+}
+
+void
+FreeRuntimeFilter(HashState *node)
+{
+	ListCell	*lc;
+	AttrFilter	*attr_filter;
+
+	if (!node->filters)
+		return;
+
+	foreach (lc, node->filters)
+	{
+		attr_filter = lfirst(lc);
+		if (attr_filter->blm_filter)
+			bloom_free(attr_filter->blm_filter);
+	}
+
+	list_free_deep(node->filters);
+	node->filters = NIL;
+}
+
+void
+ResetRuntimeFilter(HashState *node)
+{
+	ListCell		*lc;
+	AttrFilter		*attr_filter;
+	SeqScanState	*sss;
+
+	if (!node->filters)
+		return;
+
+	foreach (lc, node->filters)
+	{
+		attr_filter = lfirst(lc);
+		attr_filter->empty  = true;
+
+		if (IsA(attr_filter->target, SeqScanState))
+		{
+			sss = castNode(SeqScanState, attr_filter->target);
+			if (sss->filters)
+			{
+				list_free_deep(sss->filters);
+				sss->filters = NIL;
+			}
+		}
+
+		if (attr_filter->blm_filter)
+			bloom_free(attr_filter->blm_filter);
+
+		attr_filter->blm_filter = bloom_create_aggresive(node->ps.plan->plan_rows,
+														 work_mem,
+														 random());
+
+		StaticAssertDecl(sizeof(LONG_MAX) == sizeof(Datum), "sizeof(LONG_MAX) should be equal to sizeof(Datum)");
+		StaticAssertDecl(sizeof(LONG_MIN) == sizeof(Datum), "sizeof(LONG_MIN) should be equal to sizeof(Datum)");
+		attr_filter->min    = LONG_MAX;
+		attr_filter->max    = LONG_MIN;
+	}
+}