Push the runtime filter to table am.

Author: yjhjstz

src/backend/executor/nodeHash.c

@@ -2584,6 +2584,7 @@ ExecHashTableExplainEnd(PlanState *planstate, struct StringInfoData *buf)
     Instrumentation    *jinstrument = hjstate->js.ps.instrument;
     int                 total_buckets;
     int                 i;
+    HashState *hashState = (HashState *) innerPlanState(hjstate);
 
     if (!hashtable ||
         !hashtable->stats ||
@@ -2598,11 +2599,13 @@ ExecHashTableExplainEnd(PlanState *planstate, struct StringInfoData *buf)
 
 	if (!hashtable->eagerlyReleased)
 	{		
-		HashState *hashState = (HashState *) innerPlanState(hjstate);
-
 		/* Report on batch in progress, in case the join is being ended early. */
 		ExecHashTableExplainBatchEnd(hashState, hashtable);
 	}
+	if (gp_enable_runtime_filter_pushdown && hashState->filters)
+	{
+		ExecRFExplainEnd(hashState, buf);
+	}
 
     /* Report actual work_mem high water mark. */
     jinstrument->workmemused = Max(jinstrument->workmemused, stats->workmem_max);
@@ -4161,52 +4164,83 @@ PushdownRuntimeFilter(HashState *node)
 		scankeys = NIL;
 
 		attr_filter = lfirst(lc);
-		if (attr_filter->empty ||
+		if (attr_filter->empty || attr_filter->hasnulls ||
 			(!IsA(attr_filter->target, SeqScanState) &&
 			 !IsA(attr_filter->target, DynamicSeqScanState)))
 			continue;
 
+		SeqScanState *sss = castNode(SeqScanState, attr_filter->target);
 		/* 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);
+		sk->sk_collation = attr_filter->collation;
 		scankeys = lappend(scankeys, sk);
 
+		if (attr_filter->n_distinct > 0)
+		{
+			int64 range = attr_filter->max - attr_filter->min + 1;
+			if ((range / attr_filter->n_distinct) > gp_runtime_filter_selectivity_threshold)
+			{
+				/* push previous scankeys */
+				sss->filters = list_concat(sss->filters, scankeys);
+				continue;
+			}
+		}
 		/* 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_subtype  = attr_filter->vartype;
 		sk->sk_argument = attr_filter->min;
+		sk->sk_collation = attr_filter->collation;
 		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_subtype  = attr_filter->vartype;
 		sk->sk_argument = attr_filter->max;
+		sk->sk_collation = attr_filter->collation;
 		scankeys = lappend(scankeys, sk);
 
 		/* append new runtime filters to target node */
 		if (IsA(attr_filter->target, SeqScanState))
 		{
 			SeqScanState *sss = castNode(SeqScanState, attr_filter->target);
-			sss->filters = list_concat(sss->filters, scankeys);
+			if (sss->ss.ss_currentScanDesc != NULL)
+			{
+				/* if seqscan is started, we can't pushdown the runtime filter */
+				list_free_deep(scankeys);
+			}
+			else
+			{
+				sss->filters = list_concat(sss->filters, scankeys);
+			}
 		}
 		else if (IsA(attr_filter->target, DynamicSeqScanState))
 		{
 			DynamicSeqScanState *dsss = castNode(DynamicSeqScanState, attr_filter->target);
-			dsss->filters = list_concat(dsss->filters, scankeys);
+			if (dsss->ss.ss_currentScanDesc != NULL)
+			{
+				/* if dynamic seqscan is started, we can't pushdown the runtime filter */
+				list_free_deep(scankeys);
+			}
+			else
+			{
+				dsss->filters = list_concat(dsss->filters, scankeys);
+			}
 		}
 		else
 		{
 			/* never reach here */
 			pg_unreachable();
 		}
+
 	}
 }
 
@@ -4221,10 +4255,15 @@ AddTupleValuesIntoRF(HashState *node, TupleTableSlot *slot)
 	foreach (lc, node->filters)
 	{
 		attr_filter = (AttrFilter *) lfirst(lc);
+		if (attr_filter->hasnulls)
+			continue;
 
 		val = slot_getattr(slot, attr_filter->rattno, &isnull);
 		if (isnull)
+		{
+			attr_filter->hasnulls = true;
 			continue;
+		}
 
 		attr_filter->empty = false;
 
@@ -4274,6 +4313,7 @@ ResetRuntimeFilter(HashState *node)
 	{
 		attr_filter = lfirst(lc);
 		attr_filter->empty  = true;
+		attr_filter->hasnulls = false;
 
 		if (IsA(attr_filter->target, SeqScanState))
 		{
@@ -4304,7 +4344,7 @@ ResetRuntimeFilter(HashState *node)
 
 		attr_filter->blm_filter = bloom_create_aggresive(node->ps.plan->plan_rows,
 														 work_mem,
-														 random());
+														 gp_session_id);
 
 		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)");

src/backend/executor/nodeHashjoin.c

@@ -110,6 +110,8 @@
 
 #include "access/htup_details.h"
 #include "access/parallel.h"
+#include "catalog/pg_statistic.h"
+#include "catalog/pg_namespace.h"
 #include "executor/executor.h"
 #include "executor/hashjoin.h"
 #include "executor/instrument.h"	/* Instrumentation */
@@ -118,9 +120,12 @@
 #include "executor/nodeRuntimeFilter.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "utils/datum.h"
 #include "utils/guc.h"
 #include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
 #include "utils/memutils.h"
+#include "utils/rel.h"
 #include "utils/sharedtuplestore.h"
 
 #include "cdb/cdbvars.h"
@@ -168,10 +173,10 @@ static bool IsEqualOp(Expr *expr);
 static bool CheckEqualArgs(Expr *expr, AttrNumber *lattno, AttrNumber *rattno);
 static bool CheckTargetNode(PlanState *node,
 							AttrNumber attno,
-							AttrNumber *lattno);
+							AttrNumber *lattno, Oid *collation, Oid *var_type);
 static List *FindTargetNodes(HashJoinState *hjstate,
 							 AttrNumber attno,
-							 AttrNumber *lattno);
+							 AttrNumber *lattno, Oid *collation, Oid *var_type);
 static AttrFilter *CreateAttrFilter(PlanState *target,
 									AttrNumber lattno,
 									AttrNumber rattno,
@@ -2192,6 +2197,8 @@ CreateRuntimeFilter(HashJoinState* hjstate)
 	AttrFilter	*attr_filter;
 	ListCell	*lc;
 	List		*targets;
+	Oid			var_type;
+	Oid			collation;
 
 	/*
 	 * A build-side Bloom filter tells us if a row is definitely not in the build
@@ -2232,7 +2239,7 @@ CreateRuntimeFilter(HashJoinState* hjstate)
 		if (lattno < 1 || rattno < 1)
 			continue;
 
-		targets = FindTargetNodes(hjstate, lattno, &lattno);
+		targets = FindTargetNodes(hjstate, lattno, &lattno, &collation, &var_type);
 		if (lattno == -1 || targets == NULL)
 			continue;
 
@@ -2243,6 +2250,8 @@ CreateRuntimeFilter(HashJoinState* hjstate)
 
 			attr_filter = CreateAttrFilter(target, lattno, rattno,
 					hstate->ps.plan->plan_rows);
+			attr_filter->vartype = var_type;
+			attr_filter->collation = collation;
 			if (attr_filter->blm_filter)
 				hstate->filters = lappend(hstate->filters, attr_filter);
 			else
@@ -2329,7 +2338,7 @@ CheckEqualArgs(Expr *expr, AttrNumber *lattno, AttrNumber *rattno)
 }
 
 static bool
-CheckTargetNode(PlanState *node, AttrNumber attno, AttrNumber *lattno)
+CheckTargetNode(PlanState *node, AttrNumber attno, AttrNumber *lattno, Oid *collation, Oid *var_type)
 {
 	Var *var;
 	TargetEntry *te;
@@ -2360,6 +2369,8 @@ CheckTargetNode(PlanState *node, AttrNumber attno, AttrNumber *lattno)
 		return false;
 
 	*lattno = var->varattno;
+	*collation = var->varcollid;
+	*var_type = var->vartype;
 
 	return true;
 }
@@ -2372,7 +2383,7 @@ CheckTargetNode(PlanState *node, AttrNumber attno, AttrNumber *lattno)
  *          SeqScan <- target
  */
 static List *
-FindTargetNodes(HashJoinState *hjstate, AttrNumber attno, AttrNumber *lattno)
+FindTargetNodes(HashJoinState *hjstate, AttrNumber attno, AttrNumber *lattno, Oid *collation, Oid *var_type)
 {
 	Var *var;
 	PlanState *child, *parent;
@@ -2396,7 +2407,7 @@ FindTargetNodes(HashJoinState *hjstate, AttrNumber attno, AttrNumber *lattno)
 			 *   [result]
 			 *     seqscan | dynamicseqscan
 			 */
-			if (!CheckTargetNode(child, attno, lattno))
+			if (!CheckTargetNode(child, attno, lattno, collation, var_type))
 				return NULL;
 
 			targetNodes = lappend(targetNodes, child);
@@ -2414,7 +2425,7 @@ FindTargetNodes(HashJoinState *hjstate, AttrNumber attno, AttrNumber *lattno)
 			for (int i = 0; i < as->as_nplans; i++)
 			{
 				child = as->appendplans[i];
-				if (!CheckTargetNode(child, attno, lattno))
+				if (!CheckTargetNode(child, attno, lattno, collation, var_type))
 					return NULL;
 
 				targetNodes = lappend(targetNodes, child);
@@ -2462,12 +2473,25 @@ CreateAttrFilter(PlanState *target, AttrNumber lattno, AttrNumber rattno,
 {
 	AttrFilter *attr_filter = palloc0(sizeof(AttrFilter));
 	attr_filter->empty  = true;
+	attr_filter->hasnulls = false;
 	attr_filter->target = target;
 
 	attr_filter->lattno = lattno;
 	attr_filter->rattno = rattno;
+	attr_filter->n_distinct = 0.0;
 
-	attr_filter->blm_filter = bloom_create_aggresive(plan_rows, work_mem, random());
+	attr_filter->blm_filter = bloom_create_aggresive(plan_rows, work_mem, gp_session_id);
+
+	if (target && IsA(target, SeqScanState))
+	{
+		HeapTuple statstuple;
+		SeqScanState *scan = (SeqScanState *)target;
+		statstuple = get_att_stats(RelationGetRelid(scan->ss.ss_currentRelation), lattno);
+		if (HeapTupleIsValid(statstuple))
+		{
+			attr_filter->n_distinct = ((Form_pg_statistic) GETSTRUCT(statstuple))->stadistinct;
+		}
+	}
 
 	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)");

src/backend/executor/nodeRuntimeFilter.c

@@ -28,14 +28,18 @@
 
 #include "postgres.h"
 
+#include "access/tableam.h"
 #include "catalog/pg_type.h"
+#include "catalog/pg_statistic.h"
+#include "catalog/pg_namespace.h"
 #include "executor/executor.h"
 #include "executor/hashjoin.h"
 #include "executor/nodeRuntimeFilter.h"
 #include "lib/bloomfilter.h"
 #include "miscadmin.h"
 #include "nodes/nodeFuncs.h"
 #include "nodes/pg_list.h"
+#include "utils/datum.h"
 #include "utils/lsyscache.h"
 
 #include "cdb/cdbvars.h"
@@ -325,4 +329,28 @@ RFFillTupleValues(RuntimeFilterState *rfstate, List *values)
 		rfstate->value_buf[idx] = *dp;
 		idx++;
 	}
-}
+}
+
+void
+ExecRFExplainEnd(HashState *hashState, struct StringInfoData *buf)
+{
+	ListCell	*lc;
+	AttrFilter	*attr_filter;
+	SeqScanState *sss;
+
+	if (!hashState->filters)
+		return;
+
+	foreach (lc, hashState->filters)
+	{
+		attr_filter = lfirst(lc);
+		if (attr_filter->empty || attr_filter->hasnulls)
+			continue;
+
+		sss = castNode(SeqScanState, attr_filter->target);
+		appendStringInfo(buf, "RF: %s attrno: %d, range[%ld, %ld], n_distinct: %.2f\n",
+						 RelationGetRelationName(sss->ss.ss_currentRelation),
+						 attr_filter->lattno, (int64_t) attr_filter->min,
+						 (int64_t) attr_filter->max, attr_filter->n_distinct);
+	}
+}