Pass BF to BPP through TBPS->BPPSeeder

Author: AestheticAkhmad

dbcon/joblist/blockedbloomfilter.cpp

@@ -69,5 +69,10 @@ void BlockedBloomFilter::deserialize(messageqcpp::ByteStream& bs)
     }
 }
 
+size_t BlockedBloomFilter::getSize() const
+{
+    return bloomFilter.size();
+}
+
 } // namespace joblist
 

dbcon/joblist/blockedbloomfilter.h

@@ -24,6 +24,8 @@ class BlockedBloomFilter
         void serialize(messageqcpp::ByteStream& bs) const;
         void deserialize(messageqcpp::ByteStream& bs);
 
+        size_t getSize() const;
+
     private:
         // Member variables
         static constexpr uint8_t HASH_FUNC_COUNT = 8;

dbcon/joblist/distributedenginecomm.cpp

@@ -882,6 +882,7 @@ int32_t DistributedEngineComm::write(uint32_t senderID, const SBS& msg)
       case BATCH_PRIMITIVE_ADD_JOINER:
       case BATCH_PRIMITIVE_END_JOINER:
       case BATCH_PRIMITIVE_ABORT:
+      case BATCH_PRIMITIVE_BLOOM_FILTER:
       case DICT_CREATE_EQUALITY_FILTER:
       case DICT_DESTROY_EQUALITY_FILTER:
         /* XXXPAT: This relies on the assumption that the first pmCount "PMS*"

dbcon/joblist/primitivemsg.h

@@ -167,6 +167,7 @@ enum ISMPACKETCOMMAND
   BATCH_PRIMITIVE_END_JOINER = PRIM_LOCALBASE + 11,
   BATCH_PRIMITIVE_ACK = PRIM_LOCALBASE + 12,
   BATCH_PRIMITIVE_ABORT = PRIM_LOCALBASE + 13,
+  BATCH_PRIMITIVE_BLOOM_FILTER = PRIM_LOCALBASE + 14,
 
   // max of 100-50=50 commands
   COL_RESULTS = PRIM_COLBASE + 0,

dbcon/joblist/primitivestep.h

@@ -1475,6 +1475,7 @@ class TupleBPS : public BatchPrimitive, public TupleDeliveryStep
 
   // Blocked Bloom filter
   std::vector<std::shared_ptr<std::array<std::optional<BlockedBloomFilter>, 2>>> bloomFilters;
+  void serializeBloomFilters();
 
   public:
   void setBloomFilters(std::vector<std::shared_ptr<std::array<std::optional<BlockedBloomFilter>, 2>>>&& bloomFilters);

dbcon/joblist/tuple-bps.cpp

@@ -1480,7 +1480,10 @@ void TupleBPS::run()
     BPPIsAllocated = true;
 
     if (doJoin && tjoiners[0]->inPM())
+    {
+      serializeBloomFilters();
       serializeJoiner();
+    }
 
     prepCasualPartitioning();
     startPrimitiveThread();
@@ -3399,6 +3402,61 @@ void TupleBPS::setBloomFilters(std::vector<std::shared_ptr<std::array<std::optio
   this->bloomFilters = std::move(bloomFilters);
 }
 
+void TupleBPS::serializeBloomFilters()
+{
+  messageqcpp::ByteStream bs;
+  ISMPacketHeader ism{};
+
+  ism.Interleave = 0;
+  ism.Flags = 0;
+  ism.Command = BATCH_PRIMITIVE_BLOOM_FILTER;
+  ism.Type = 2;
+  
+  uint32_t messageSize = sizeof(ISMPacketHeader);
+
+  // How to get rid of iterating over BFs 2 times?
+  for (const auto& joinerBF : bloomFilters)
+  {
+    for (size_t i = 0; i < joinerBF->size(); ++i)
+    {
+      messageSize += sizeof(uint8_t);
+
+      if ((*joinerBF)[i].has_value())
+      {
+        messageSize += (*joinerBF)[i]->getSize() * sizeof(uint64_t);
+      }
+
+    }
+  }
+
+  ism.Size = messageSize;
+  bs.append((uint8_t*)&ism, sizeof(ism));
+
+  bs << txnId();
+  bs << sessionId();
+  bs << static_cast<uint32_t>(stepId());
+  bs << uniqueID;
+
+  for (const auto& joinerBF : bloomFilters)
+  {
+    for (size_t i = 0; i < joinerBF->size(); ++i)
+    {
+      uint8_t hasBloomFilter = (*joinerBF)[i].has_value() ? 1 : 0;
+      bs << hasBloomFilter;
+
+      if (hasBloomFilter)
+      {
+        (*joinerBF)[i]->serialize(bs);
+      }
+    
+    }
+  }
+
+  SBS sbs(new messageqcpp::ByteStream(bs));
+  fDec->write(uniqueID, sbs);
+
+}
+
 template bool TupleBPS::processOneFilterType<int64_t>(int8_t colWidth, int64_t value, uint32_t type) const;
 template bool TupleBPS::processOneFilterType<int128_t>(int8_t colWidth, int128_t value, uint32_t type) const;
 

primitives/primproc/batchprimitiveprocessor.cpp

@@ -2695,4 +2695,34 @@ void BatchPrimitiveProcessor::buildVSSCache(uint32_t loopCount)
       vssCache.insert(make_pair(lbidList[i], vssData[i]));
 }
 
+void BatchPrimitiveProcessor::addBloomFilters([[maybe_unused]] messageqcpp::ByteStream& bs)
+{
+  bs.advance(sizeof(ISMPacketHeader) + 4 * sizeof(uint32_t));
+  if (bloomFilters.empty() && doJoin)
+  {
+    bloomFilters.resize(joinerCount);
+  }
+
+  for (size_t j = 0; j < joinerCount; ++j)
+  {
+    for (size_t i = 0; i < 2; ++i)
+    {
+      uint8_t hasBloomFilter = 0;
+      bs >> hasBloomFilter;
+
+      if (hasBloomFilter)
+      {
+        if (!bloomFilters[j])
+          bloomFilters[j] = std::make_shared<std::array<std::optional<BlockedBloomFilter>, 2>>();
+        
+        (*bloomFilters[j]).at(i).emplace();
+        (*bloomFilters[j]).at(i)->deserialize(bs);
+        
+      }
+      
+    }
+  }
+
+}
+
 }  // namespace primitiveprocessor

primitives/primproc/batchprimitiveprocessor.h

@@ -436,6 +436,12 @@ class BatchPrimitiveProcessor
   bool initiatedByEM_;
   uint32_t weight_;
 
+  // Blocked Bloom Filter
+  std::vector<std::shared_ptr<std::array<std::optional<joblist::BlockedBloomFilter>, 2>>> bloomFilters;
+  public:
+  void addBloomFilters(messageqcpp::ByteStream& bs);
+  private:
+
   uint32_t maxPmJoinResultCount = 1048576;
   friend class Command;
   friend class ColumnCommand;

primitives/primproc/primitiveserver.cpp

@@ -1279,6 +1279,20 @@ struct BPPHandler
     }
   };
 
+  struct BloomFilter : public BPPHandlerFunctor
+  {
+    BloomFilter(boost::shared_ptr<BPPHandler> r, SBS b) : BPPHandlerFunctor(std::move(r), std::move(b))
+    {
+    }
+
+    int operator()() override
+    {
+      utils::setThreadName("PPHandBloomFilter");
+      return rt->addBloomFiltersToBPP(*bs, dieTime);
+    }
+
+  };
+
   int doAbort(ByteStream& bs, const posix_time::ptime& dieTime)
   {
     uint32_t key;
@@ -1486,6 +1500,37 @@ struct BPPHandler
     }
   }
 
+  int addBloomFiltersToBPP(ByteStream& bs, const posix_time::ptime& dieTime)
+  {
+    SBPPV bppv;
+    uint32_t uniqueID;
+    const uint8_t* buf;
+
+    buf = bs.buf();
+    uniqueID = *((const uint32_t*)&buf[sizeof(ISMPacketHeader) + 3 * sizeof(uint32_t)]);
+
+    bppv = grabBPPs(uniqueID);
+
+    if (bppv)
+    {
+      boost::shared_lock<boost::shared_mutex> lk(getDJLock(uniqueID));
+      bppv->get()[0]->addBloomFilters(bs);
+      return 0;
+    }
+    else
+    {
+      if (posix_time::second_clock::universal_time() > dieTime)
+      {
+        cout << "addBloomFilterToBPP: job for id " << uniqueID << " has been killed." << endl;
+        return 0;
+      }
+      else
+      {
+        return -1;
+      }
+    }
+  }
+
   int lastJoinerMsg(ByteStream& bs, const posix_time::ptime& dieTime)
   {
     SBPPV bppv;
@@ -2024,6 +2069,30 @@ struct ReadThread
         fBPPHandler->doAck(*sbs);
         break;
       }
+
+      case BATCH_PRIMITIVE_BLOOM_FILTER:
+      {
+        const uint8_t* buf = sbs->buf();
+        uint32_t pos = sizeof(ISMPacketHeader);
+        const uint32_t txnId = *((uint32_t*)&buf[pos]);
+        [[maybe_unused]] const uint32_t sessionID = *((uint32_t*)&buf[pos + 4]);
+        const uint32_t stepID = *((uint32_t*)&buf[pos + 8]);
+        const uint32_t uniqueID = *((uint32_t*)&buf[pos + 12]);
+        
+        const uint32_t id = 0;
+        const uint32_t weight = threadpool::MetaJobsInitialWeight;
+        const uint32_t priority = 0;
+        
+        boost::shared_ptr<FairThreadPool::Functor> functor;
+        functor.reset(new BPPHandler::BloomFilter(fBPPHandler, sbs));
+
+        PriorityThreadPool::Job job(uniqueID, stepID, txnId, functor, outIos, weight, priority, id);
+        OOBProcPool->addJob(job);
+
+
+        break;
+      }
+
       default:
       {
         std::ostringstream os;