feat: pushdown bloom filter to pax table am

This optimization pushes down Bloom Filter conditions for runtime filters
to the Pax Table AM layer.

By applying the filter earlier than the SeqNext() function, it eliminates
the overhead of converting data from columnar format to TableTupleSlot,
resulting in faster query execution

Author: gongxun0928

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

@@ -372,7 +372,7 @@ TableScanDesc PaxScanDesc::BeginScanExtractColumns(
         && !(flags & SO_TYPE_VECTOR)
 #endif
     ) {
-      filter->InitRowFilter(rel, ps, filter->GetColumnProjection());
+      filter->InitRowFilter(rel, ps, filter->GetColumnProjection(), key, nkeys);
     }
   }
   return BeginScan(rel, snapshot, nkeys, key, parallel_scan, flags,

contrib/pax_storage/src/cpp/comm/cbdb_api.h

@@ -80,6 +80,7 @@ extern "C" {
 #include "commands/progress.h"
 #include "commands/tablecmds.h"
 #include "funcapi.h"
+#include "lib/bloomfilter.h"
 #include "miscadmin.h"
 #include "nodes/bitmapset.h"
 #include "nodes/execnodes.h"

contrib/pax_storage/src/cpp/storage/filter/pax_filter.cc

@@ -43,9 +43,8 @@ namespace pax {
 
 PaxFilter::PaxFilter() : sparse_filter_(nullptr), row_filter_(nullptr) {}
 
-void PaxFilter::InitSparseFilter(Relation relation, List *quals,
-                                 ScanKey key, int nkeys,
-                                 bool allow_fallback_to_pg) {
+void PaxFilter::InitSparseFilter(Relation relation, List *quals, ScanKey key,
+                                 int nkeys, bool allow_fallback_to_pg) {
   Assert(!sparse_filter_);
   sparse_filter_ =
       std::make_shared<PaxSparseFilter>(relation, allow_fallback_to_pg);
@@ -123,10 +122,11 @@ void PaxFilter::SetColumnProjection(const std::vector<int> &cols, int natts) {
 }
 
 void PaxFilter::InitRowFilter(Relation relation, PlanState *ps,
-                              const std::vector<bool> &projection) {
+                              const std::vector<bool> &projection, ScanKey key,
+                              int nkeys) {
   Assert(!row_filter_);
   row_filter_ = std::make_shared<PaxRowFilter>();
-  if (!row_filter_->Initialize(relation, ps, projection)) {
+  if (!row_filter_->Initialize(relation, ps, projection, key, nkeys)) {
     row_filter_ = nullptr;
   }
 }

contrib/pax_storage/src/cpp/storage/filter/pax_filter.h

@@ -73,7 +73,8 @@ class PaxFilter final {
 
   // The row filter
   void InitRowFilter(Relation relation, PlanState *ps,
-                     const std::vector<bool> &projection);
+                     const std::vector<bool> &projection, ScanKey key,
+                     int nkeys);
   std::shared_ptr<PaxRowFilter> GetRowFilter();
 
   void LogStatistics() const;

contrib/pax_storage/src/cpp/storage/filter/pax_row_filter.cc

@@ -26,6 +26,7 @@
  */
 
 #include "storage/filter/pax_row_filter.h"
+
 #include "comm/cbdb_wrappers.h"
 
 namespace paxc {
@@ -45,19 +46,57 @@ static inline void FindAttrsInQual(Node *qual, bool *proj, int ncol,
 }
 
 static bool BuildExecutionFilterForColumns(Relation rel, PlanState *ps,
-                                    pax::ExecutionFilterContext *ctx) {
+                                           pax::ExecutionFilterContext *ctx,
+                                           ScanKey key, int nkeys) {
   List *qual = ps->plan->qual;
   List **qual_list;
   ListCell *lc;
   bool *proj;
   int *qual_atts;
   int natts = RelationGetNumberOfAttributes(rel);
 
-  if (!qual || !IsA(qual, List)) return false;
+  int ret = false;
+
+  if (key && nkeys > 0) {
+    // We don't need to support DynamicSeqScanState here. Even if the plan uses
+    // DynamicSeqScanNode for partitioned tables, it's always a regular SeqScan
+    // on a single table. So this will always be a SeqScanState.
+    if (nodeTag(ps) != T_SeqScanState) {
+      elog(ERROR, "runtime filter only support seqscan state, but got %d",
+           nodeTag(ps));
+    }
+
+    for (int i = 0; i < nkeys; i++) {
+      if (key[i].sk_flags & SK_BLOOM_FILTER) {
+        ctx->runtime_bloom_keys.emplace_back(key[i]);
+        ret = true;
+      }
+    }
+
+    // register bloom filters
+    for (int i = 0; i < (int)ctx->runtime_bloom_keys.size(); ++i) {
+      pax::ExecutionFilterContext::FilterNode node;
+      node.kind = pax::ExecutionFilterContext::FilterKind::kBloom;
+      node.index = i;
+      ctx->filter_nodes.emplace_back(node);
+    }
+
+    if (ps->instrument) {
+      ps->instrument->prf_work = true;
+    }
+    ctx->ps = ps;
+
+    // set filter_in_seqscan to false, so that the filter will not be executed
+    // in SeqNext(), but will be executed in pax_row_filter
+    auto seqscan = (SeqScanState *)ps;
+    seqscan->filter_in_seqscan = false;
+  }
+
+  if (!qual || !IsA(qual, List)) return ret;
 
   if (list_length(qual) == 1 && IsA(linitial(qual), BoolExpr)) {
     auto boolexpr = (BoolExpr *)linitial(qual);
-    if (boolexpr->boolop != AND_EXPR) return false;
+    if (boolexpr->boolop != AND_EXPR) return ret;
     qual = boolexpr->args;
   }
   Assert(IsA(qual, List));
@@ -98,6 +137,11 @@ static bool BuildExecutionFilterForColumns(Relation rel, PlanState *ps,
       if (!qual_list[i]) continue;
       ctx->estates[k] = ExecInitQual(qual_list[i], ps);
       ctx->attnos[k] = i;
+      // register expr filter node (by index k)
+      pax::ExecutionFilterContext::FilterNode node;
+      node.kind = pax::ExecutionFilterContext::FilterKind::kExpr;
+      node.index = k;
+      ctx->filter_nodes.emplace_back(node);
       list_free(qual_list[i]);
       k++;
     }
@@ -108,7 +152,11 @@ static bool BuildExecutionFilterForColumns(Relation rel, PlanState *ps,
     list_free(qual_list[0]);
   }
 
-  Assert(ctx->size > 0 || ctx->estate_final);
+  Assert(ctx->size > 0 || ctx->estate_final ||
+         ctx->runtime_bloom_keys.size() > 0);
+
+  // remove qual from plan state, so that the qual will not be executed in
+  // executor, but will be executed in pax_row_filter
   ps->qual = nullptr;
 
   pfree(proj);
@@ -117,20 +165,19 @@ static bool BuildExecutionFilterForColumns(Relation rel, PlanState *ps,
   return true;
 }
 
-} // namespace paxc
-
+}  // namespace paxc
 
 namespace pax {
 
 PaxRowFilter::PaxRowFilter() {}
 
-bool PaxRowFilter::Initialize(Relation rel, PlanState *ps, const std::vector<bool> &projection) {
+bool PaxRowFilter::Initialize(Relation rel, PlanState *ps,
+                              const std::vector<bool> &projection, ScanKey key,
+                              int nkeys) {
   bool ok = false;
-  
+
   CBDB_WRAP_START;
-  { 
-    ok = paxc::BuildExecutionFilterForColumns(rel, ps, &efctx_); 
-  }
+  { ok = paxc::BuildExecutionFilterForColumns(rel, ps, &efctx_, key, nkeys); }
   CBDB_WRAP_END;
 
   if (ok) {
@@ -140,7 +187,8 @@ bool PaxRowFilter::Initialize(Relation rel, PlanState *ps, const std::vector<boo
   return ok;
 }
 
-void PaxRowFilter::FillRemainingColumns(Relation rel, const std::vector<bool> &projection) {
+void PaxRowFilter::FillRemainingColumns(Relation rel,
+                                        const std::vector<bool> &projection) {
   int natts = RelationGetNumberOfAttributes(rel);
   auto proj_len = projection.size();
   std::vector<bool> atts(natts);
@@ -162,5 +210,4 @@ void PaxRowFilter::FillRemainingColumns(Relation rel, const std::vector<bool> &p
   }
 }
 
-
 }  // namespace pax

contrib/pax_storage/src/cpp/storage/filter/pax_row_filter.h

@@ -42,16 +42,38 @@ struct ExecutionFilterContext {
   ExprState *estate_final = nullptr;
   ExprState **estates;
   AttrNumber *attnos;
+  PlanState *ps;
   int size = 0;
   inline bool HasExecutionFilter() const { return size > 0 || estate_final; }
+
+  // runtime bloom filters pushed down via SeqScanState->filters
+  // (SK_BLOOM_FILTER)
+  std::vector<ScanKeyData> runtime_bloom_keys;
+
+  // unified filter nodes (expr + bloom) for execution ordering
+  enum class FilterKind { kExpr, kBloom };
+  struct FilterNode {
+    FilterKind kind;
+    int index;  // index in estates (for kExpr) or in runtime_bloom_keys (for
+                // kBloom)
+    uint64 tested = 0;   // number of rows tested during sampling
+    uint64 passed = 0;   // number of rows passed during sampling
+    double score = 1.0;  // pass rate used for ordering (lower is better)
+  };
+  std::vector<FilterNode> filter_nodes;
+
+  // sampling control to determine filter order
+  bool sampling = true;
+  uint64 sample_target = 65536;  // number of rows for sampling phase
+  uint64 sample_rows = 0;        // rows seen in sampling
 };
 
 class PaxRowFilter final {
-public:
+ public:
   PaxRowFilter();
 
   bool Initialize(Relation rel, PlanState *ps,
-                const std::vector<bool> &projection);
+                  const std::vector<bool> &projection, ScanKey key, int nkeys);
 
   inline const ExecutionFilterContext *GetExecutionFilterContext() const {
     return &efctx_;
@@ -60,17 +82,16 @@ class PaxRowFilter final {
   inline const std::vector<AttrNumber> &GetRemainingColumns() const {
     return remaining_attnos_;
   }
-  
-private:
+
+ private:
   void FillRemainingColumns(Relation rel, const std::vector<bool> &projection);
 
-private:
+ private:
   ExecutionFilterContext efctx_;
   // all selected columns - single row filting columns
   // before running final cross columns expression filtering, the remaining
   // columns should be filled.
   std::vector<AttrNumber> remaining_attnos_;
 };
 
-
-}; 
+};  // namespace pax

contrib/pax_storage/src/cpp/storage/filter/pax_sparse_pg_path.cc

@@ -50,12 +50,14 @@ void PaxSparseFilter::Initialize(List *quals, ScanKey key, int nkeys) {
 
   // walk scan key and only support min/max filter now
   for (int i = 0; i < nkeys; i++) {
-    // TODO: support bloom filter in PaxFilter
-    // but now just skip it, SeqNext() will check bloom filter in PassByBloomFilter()
+    // Now just skip bloom filter here, it will be handled in PaxRowFilter.
     if (key[i].sk_flags & SK_BLOOM_FILTER) {
       continue;
     }
 
+    // PushdownRuntimeFilter only support BTGreaterEqualStrategyNumber and
+    // BTLessEqualStrategyNumber. If the strategy is not supported, skip orther
+    // strategies.
     if (key[i].sk_strategy != BTGreaterEqualStrategyNumber &&
         key[i].sk_strategy != BTLessEqualStrategyNumber) {
       continue;