PaxAccessMethod::ScanFlags support SCAN_SUPPORT_RUNTIME_FILTER.

Add runtime_filter test with minmax.

Author: yjhjstz

contrib/pax_storage/expected/runtime_filter.out

@@ -20,5 +20,6 @@ test: dictionary_encoding
 
 test: cluster
 test: db_size_functions
+test: runtime_filter
 
 test: teardown

contrib/pax_storage/pax_schedule

@@ -0,0 +1,86 @@
+SET optimizer TO on;
+
+-- Test Suit 1: runtime filter main case
+DROP TABLE IF EXISTS fact_rf, dim_rf;
+CREATE TABLE fact_rf (fid int, did int, val int) using pax WITH(minmax_columns='fid,did,val');
+CREATE TABLE dim_rf (did int, proj_id int, filter_val int) using pax WITH(minmax_columns='did,proj_id,filter_val');
+
+-- Generating data, fact_rd.did and dim_rf.did is 80% matched
+INSERT INTO fact_rf SELECT i, i % 8000 + 1, i FROM generate_series(1, 100000) s(i);
+INSERT INTO dim_rf SELECT i, i % 10, i FROM generate_series(1, 10000) s(i);
+ANALYZE fact_rf, dim_rf;
+
+SET gp_enable_runtime_filter_pushdown TO off;
+EXPLAIN ANALYZE SELECT COUNT(*) FROM fact_rf, dim_rf
+    WHERE fact_rf.did = dim_rf.did AND proj_id < 2 AND filter_val <= 1000;
+
+SET gp_enable_runtime_filter_pushdown TO on;
+EXPLAIN ANALYZE SELECT COUNT(*) FROM fact_rf, dim_rf
+    WHERE fact_rf.did = dim_rf.did AND proj_id < 2 AND filter_val <= 1000;
+
+-- Test bad filter rate
+EXPLAIN ANALYZE SELECT COUNT(*) FROM fact_rf, dim_rf
+    WHERE fact_rf.did = dim_rf.did AND proj_id < 7;
+
+-- Test outer join
+-- LeftJoin (eliminated and applicatable)
+EXPLAIN ANALYZE SELECT COUNT(*) FROM
+    fact_rf LEFT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE proj_id < 2 AND filter_val <= 1000;
+
+-- LeftJoin
+EXPLAIN ANALYZE SELECT COUNT(*) FROM
+    fact_rf LEFT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE proj_id IS NULL OR proj_id < 2 AND filter_val <= 1000;
+
+-- RightJoin (applicatable)
+EXPLAIN ANALYZE SELECT COUNT(*) FROM
+    fact_rf RIGHT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE proj_id < 2 AND filter_val <= 1000;
+
+-- SemiJoin
+EXPLAIN ANALYZE SELECT COUNT(*) FROM fact_rf
+    WHERE fact_rf.did IN (SELECT did FROM dim_rf WHERE proj_id < 2 AND filter_val <= 1000);
+
+-- SemiJoin -> InnerJoin and deduplicate
+EXPLAIN ANALYZE SELECT COUNT(*) FROM dim_rf
+    WHERE dim_rf.did IN (SELECT did FROM fact_rf) AND proj_id < 2 AND filter_val <= 1000;
+
+-- Test correctness
+SELECT * FROM fact_rf, dim_rf
+    WHERE fact_rf.did = dim_rf.did AND dim_rf.filter_val = 1
+    ORDER BY fid;
+
+SELECT * FROM
+    fact_rf LEFT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE dim_rf.filter_val = 1
+    ORDER BY fid;
+
+SELECT COUNT(*) FROM
+    fact_rf LEFT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE proj_id < 2 AND filter_val <= 1000;
+
+SELECT COUNT(*) FROM
+    fact_rf LEFT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE proj_id IS NULL OR proj_id < 2 AND filter_val <= 1000;
+
+SELECT COUNT(*) FROM
+    fact_rf RIGHT JOIN dim_rf ON fact_rf.did = dim_rf.did
+    WHERE proj_id < 2 AND filter_val <= 1000;
+
+SELECT COUNT(*) FROM fact_rf
+    WHERE fact_rf.did IN (SELECT did FROM dim_rf WHERE proj_id < 2 AND filter_val <= 1000);
+
+SELECT COUNT(*) FROM dim_rf
+    WHERE dim_rf.did IN (SELECT did FROM fact_rf) AND proj_id < 2 AND filter_val <= 1000;
+
+-- Test contain null values
+INSERT INTO dim_rf VALUES (NULL,1, 1);
+EXPLAIN ANALYZE SELECT COUNT(*) FROM fact_rf, dim_rf
+    WHERE fact_rf.did = dim_rf.did AND proj_id < 2 AND filter_val <= 1000;
+SELECT COUNT(*) FROM fact_rf, dim_rf
+    WHERE fact_rf.did = dim_rf.did AND proj_id < 2 AND filter_val <= 1000;
+
+-- Clean up: reset guc
+SET gp_enable_runtime_filter_pushdown TO off;
+RESET optimizer;

contrib/pax_storage/sql/runtime_filter.sql

@@ -443,6 +443,7 @@ const TupleTableSlotOps *PaxAccessMethod::SlotCallbacks(
 
 uint32 PaxAccessMethod::ScanFlags(Relation relation) {
   uint32 flags = 0;
+  std::vector<int> minmax_columns;
 #ifdef VEC_BUILD
   flags |= SCAN_SUPPORT_VECTORIZATION | SCAN_SUPPORT_COLUMN_ORIENTED_SCAN;
 #else
@@ -452,6 +453,10 @@ uint32 PaxAccessMethod::ScanFlags(Relation relation) {
 #if defined(USE_MANIFEST_API) && !defined(USE_PAX_CATALOG)
   flags |= SCAN_FORCE_BIG_WRITE_LOCK;
 #endif
+  minmax_columns = cbdb::GetMinMaxColumnIndexes(relation);
+  if (!minmax_columns.empty()) {
+    flags |= SCAN_SUPPORT_RUNTIME_FILTER;
+  }
 
   return flags;
 }

contrib/pax_storage/src/cpp/access/pax_access_handle.cc

@@ -4169,7 +4169,6 @@ PushdownRuntimeFilter(HashState *node)
 			 !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;
@@ -4181,6 +4180,7 @@ PushdownRuntimeFilter(HashState *node)
 
 		if (attr_filter->n_distinct > 0)
 		{
+			SeqScanState *sss = castNode(SeqScanState, attr_filter->target);
 			int64 range = attr_filter->max - attr_filter->min + 1;
 			if ((range / attr_filter->n_distinct) > gp_runtime_filter_selectivity_threshold)
 			{

src/backend/executor/nodeHash.c

@@ -2197,8 +2197,8 @@ CreateRuntimeFilter(HashJoinState* hjstate)
 	AttrFilter	*attr_filter;
 	ListCell	*lc;
 	List		*targets;
-	Oid			var_type;
-	Oid			collation;
+	Oid			var_type = InvalidOid;
+	Oid			collation = InvalidOid;
 
 	/*
 	 * A build-side Bloom filter tells us if a row is definitely not in the build

src/backend/executor/nodeHashjoin.c

@@ -347,10 +347,13 @@ ExecRFExplainEnd(HashState *hashState, struct StringInfoData *buf)
 		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",
+		if (IsA(attr_filter->target, SeqScanState))
+		{
+			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);
+		}
 	}
 }