feat(dumper): dumper interface improvements and some TNS and OB refactoring

Author: drrtuy

dbcon/joblist/disk-based-topnorderby.h

@@ -21,19 +21,26 @@
 #include <string>
 #include <vector>
 
+#include "dumper.h"
 #include "elementtype.h"
+#include "resourcemanager.h"
 namespace joblist
 {
 
-class DiskBasedTopNOrderBy
+class DiskBasedTopNOrderBy : public rowgroup::RGDumper
 {
+  // std::string fTmpDir =
+  //     config::Config::makeConfig()->getTempFileDir(config::Config::TempDirPurpose::Aggregates);
+  // std::string fCompStr = config::Config::makeConfig()->getConfig("RowAggregation", "Compression");
  public:
-  DiskBasedTopNOrderBy() {}
-  ~DiskBasedTopNOrderBy() 
+  // TODO Parametrize compression, tmpdir and memory manager (can be temp)
+  DiskBasedTopNOrderBy(ResourceManager* rm)
+   : RGDumper(compress::getCompressInterfaceByName("LZ4"), std::make_unique<rowgroup::MemManager>(),
+              config::Config::makeConfig()->getTempFileDir(config::Config::TempDirPurpose::Sorting),
+              "Sorting", reinterpret_cast<std::uintptr_t>(this))
   {
-     // clean up left over files
   }
-
+  ~DiskBasedTopNOrderBy() = default;
 
   void incrementGenerationCounter()
   {
@@ -44,16 +51,30 @@ class DiskBasedTopNOrderBy
     return fGenerationCounter;
   }
 
-  bool isDiskBased() const { return fGenerationCounter > 0; }
+  bool isDiskBased() const
+  {
+    return fGenerationCounter > 0;
+  }
 
-  size_t getGenerationFilesNumber() const { return 0; }
-  std::vector<std::string> getGenerationFileNamesNextBatch(const size_t batchSize) { return {}; }
+  size_t getGenerationFilesNumber() const
+  {
+    return 0;
+  }
+  std::vector<std::string> getGenerationFileNamesNextBatch(const size_t batchSize)
+  {
+    return {};
+  }
 
-  void flushCurrentToDisk(const bool firstFlush) { incrementGenerationCounter(); }
-  void diskBasedMergePhaseIfNeeded(std::vector<RowGroupDLSPtr>& dataLists) {}
+  void flushCurrentToDisk(const bool firstFlush)
+  {
+    incrementGenerationCounter();
+  }
+  void diskBasedMergePhaseIfNeeded(std::vector<RowGroupDLSPtr>& dataLists)
+  {
+  }
 
-//  private:
-  uint64_t fGenerationCounter {0};
+  //  private:
+  uint64_t fGenerationCounter{0};
 };
 
-}
+}  // namespace joblist

dbcon/joblist/jlf_tuplejoblist.cpp

@@ -499,12 +499,6 @@ void adjustLastStep(JobStepVector& querySteps, DeliveredTableMap& deliverySteps,
     deliverySteps[CNX_VTABLE_ID] = ws;
   }
 
-  // TODO MCOL-894 we don't need to run sorting|distinct
-  // every time
-  //    if ((jobInfo.limitCount != (uint64_t) - 1) ||
-  //            (jobInfo.constantCol == CONST_COL_EXIST) ||
-  //            (jobInfo.hasDistinct))
-  //    {
   if (jobInfo.annexStep.get() == NULL)
     jobInfo.annexStep.reset(new TupleAnnexStep(jobInfo));
 
@@ -513,20 +507,19 @@ void adjustLastStep(JobStepVector& querySteps, DeliveredTableMap& deliverySteps,
 
   if (jobInfo.orderByColVec.size() > 0)
   {
-    tas->addOrderBy(new LimitedOrderBy());
+    tas->addOrderBy(jobInfo.rm);
     if (jobInfo.orderByThreads > 1)
       tas->setParallelOp();
     tas->setMaxThreads(jobInfo.orderByThreads);
   }
 
+  // TODO decouple TCS from TNS
   if (jobInfo.constantCol == CONST_COL_EXIST)
     tas->addConstant(new TupleConstantStep(jobInfo));
 
   if (jobInfo.hasDistinct)
     tas->setDistinct();
 
-  //    }
-
   if (jobInfo.annexStep)
   {
     TupleDeliveryStep* ds = dynamic_cast<TupleDeliveryStep*>(deliverySteps[CNX_VTABLE_ID].get());

dbcon/joblist/limitedorderby.cpp

@@ -40,7 +40,8 @@ namespace joblist
 const uint64_t LimitedOrderBy::fMaxUncommited = 102400;  // 100KiB - make it configurable?
 
 // LimitedOrderBy class implementation
-LimitedOrderBy::LimitedOrderBy() : fStart(0), fCount(-1), fUncommitedMemory(0)
+LimitedOrderBy::LimitedOrderBy(ResourceManager* rm)
+ : DiskBasedTopNOrderBy(rm), fStart(0), fCount(-1), fUncommitedMemory(0)
 {
   fRule.fIdbCompare = this;
 }
@@ -109,13 +110,12 @@ void LimitedOrderBy::processRow(const rowgroup::Row& row)
   if (fCount == 0)
     return;
 
-  std::cout << "LimitedOrderBy::processRow row " << row.toString() << std::endl; 
-  std::cout << "LimitedOrderBy::processRow fStart " << fStart << " fCount " << fCount << std::endl; 
+  std::cout << "LimitedOrderBy::processRow row " << row.toString() << std::endl;
+  std::cout << "LimitedOrderBy::processRow fStart " << fStart << " fCount " << fCount << std::endl;
   auto& orderedRowsQueue = getQueue();
   // if the row count is less than the limit
   if (orderedRowsQueue.size() < fStart + fCount)
   {
-
     copyRow(row, &fRow0);
     OrderByRow newRow(fRow0, fRule);
     orderedRowsQueue.push(newRow);
@@ -163,7 +163,7 @@ void LimitedOrderBy::processRow(const rowgroup::Row& row)
     OrderByRow swapRow = orderedRowsQueue.top();
     row1.setData(swapRow.fData);
     std::cout << "LimitedOrderBy::processRow row2swap " << row1.toString() << std::endl;
-    std::cout <<"LimitedOrderBy::processRow new row 4 swaping " << row.toString() << std::endl;
+    std::cout << "LimitedOrderBy::processRow new row 4 swaping " << row.toString() << std::endl;
 
     copyRow(row, &row1);
 
@@ -178,7 +178,6 @@ void LimitedOrderBy::processRow(const rowgroup::Row& row)
   }
 }
 
-
 void LimitedOrderBy::processRow_(const rowgroup::Row& row)
 {
   // check if this is a distinct row
@@ -189,13 +188,12 @@ void LimitedOrderBy::processRow_(const rowgroup::Row& row)
   if (fCount == 0)
     return;
 
-
   // TODO copy rules or replace ptrs to real instances in CompareRules
   // auto invertedRule = fRule;
   // invertedRule.revertRules();
 
-  std::cout << "LimitedOrderBy::processRow row " << row.toString() << std::endl; 
-  std::cout << "LimitedOrderBy::processRow fStart " << fStart << " fCount " << fCount << std::endl; 
+  std::cout << "LimitedOrderBy::processRow row " << row.toString() << std::endl;
+  std::cout << "LimitedOrderBy::processRow fStart " << fStart << " fCount " << fCount << std::endl;
   auto& orderedRowsQueue = getQueue();
   // if the row count is less than the limit
   if (orderedRowsQueue.size() < fStart + fCount)
@@ -247,7 +245,7 @@ void LimitedOrderBy::processRow_(const rowgroup::Row& row)
     OrderByRow swapRow = orderedRowsQueue.top();
     row1.setData(swapRow.fData);
     std::cout << "LimitedOrderBy::processRow row2swap " << row1.toString() << std::endl;
-    std::cout <<"LimitedOrderBy::processRow new row 4 swaping " << row.toString() << std::endl;
+    std::cout << "LimitedOrderBy::processRow new row 4 swaping " << row.toString() << std::endl;
 
     copyRow(row, &row1);
 
@@ -268,8 +266,8 @@ void LimitedOrderBy::processRow_(const rowgroup::Row& row)
 
 //   // Skip OFFSET
 //   uint64_t sqlOffset = fStart;
-//   std::cout << "brandNewFinalize offset " << sqlOffset << " orderedRowsQueue.size() " << orderedRowsQueue.size() << std::endl;
-//   while (sqlOffset > 0 && !orderedRowsQueue.empty())
+//   std::cout << "brandNewFinalize offset " << sqlOffset << " orderedRowsQueue.size() " <<
+//   orderedRowsQueue.size() << std::endl; while (sqlOffset > 0 && !orderedRowsQueue.empty())
 //   {
 //     auto r = orderedRowsQueue.top();
 //     row1.setData(r.fData);
@@ -283,7 +281,7 @@ void LimitedOrderBy::brandNewFinalize()
 {
   if (!isDiskBased())
   {
-    return finalize();    
+    return finalize();
   }
 
   // if disk-based
@@ -391,11 +389,9 @@ void LimitedOrderBy::brandNewFinalize()
 
     fDataQueue = tempQueue;
   }
-
-
 }
 
-/* 
+/*
  * The f() copies top element from an ordered queue into a row group. It
  * does this backwards to syncronise sorting orientation with the server.
  * The top row from the queue goes last into the returned set.
@@ -525,7 +521,8 @@ bool LimitedOrderBy::getNextRGData(RGData& data)
   // and the current sorted queue size.
   uint64_t rowsToRetrieve = std::min(fCount - fRowsReturned, fRowsPerRG);
   uint64_t rowsToRetrieveFromQueue = std::min(rowsToRetrieve, orderedRowsQueue.size());
-  std::cout << "getNextRGData rowsToRetrieve " << rowsToRetrieve << " orderedRowsQueue.size() " << orderedRowsQueue.size() << std::endl;
+  std::cout << "getNextRGData rowsToRetrieve " << rowsToRetrieve << " orderedRowsQueue.size() "
+            << orderedRowsQueue.size() << std::endl;
   std::cout << "getNextRGData rowsToRetrieveFromQueue " << rowsToRetrieveFromQueue << std::endl;
   for (; rowsToRetrieveFromQueue > thisRGRowNumber; ++thisRGRowNumber)
   {

dbcon/joblist/limitedorderby.h

@@ -25,6 +25,7 @@
 
 #include "disk-based-topnorderby.h"
 #include "../../utils/windowfunction/idborderby.h"
+#include "resourcemanager.h"
 #include "rowgroup.h"
 
 namespace joblist
@@ -39,7 +40,7 @@ struct JobInfo;
 class LimitedOrderBy : public ordering::IdbOrderBy, public DiskBasedTopNOrderBy
 {
  public:
-  LimitedOrderBy();
+  LimitedOrderBy(ResourceManager* rm = nullptr); // TODO remove default
   ~LimitedOrderBy() override;
   using ordering::IdbOrderBy::initialize;
   void initialize(const rowgroup::RowGroup&, const JobInfo&, bool invertRules = false,

dbcon/joblist/tupleannexstep.cpp

@@ -118,8 +118,7 @@ TupleAnnexStep::TupleAnnexStep(const JobInfo& jobInfo)
  , fEndOfResult(false)
  , fDistinct(false)
  , fParallelOp(false)
- , fOrderBy(NULL)
- , fConstant(NULL)
+ , fConstant(nullptr)
  , fFeInstance(funcexp::FuncExp::instance())
  , fJobList(jobInfo.jobListPtr)
  , fFinishedThreads(0)
@@ -146,10 +145,10 @@ TupleAnnexStep::~TupleAnnexStep()
     fRunnersList.clear();
   }
 
-  if (fOrderBy)
-    delete fOrderBy;
+  // if (fOrderBy)
+  //   delete fOrderBy;
 
-  fOrderBy = NULL;
+  // fOrderBy = NULL;
 
   if (fConstant)
     delete fConstant;

dbcon/joblist/tupleannexstep.h

@@ -70,9 +70,9 @@ class TupleAnnexStep : public JobStep, public TupleDeliveryStep
 
   void initialize(const rowgroup::RowGroup& rgIn, const JobInfo& jobInfo);
 
-  void addOrderBy(LimitedOrderBy* lob)
+  void addOrderBy(ResourceManager* rm)
   {
-    fOrderBy = lob;
+    fOrderBy = std::make_unique<LimitedOrderBy>(rm);
   }
   void addConstant(TupleConstantStep* tcs)
   {
@@ -187,7 +187,7 @@ class TupleAnnexStep : public JobStep, public TupleDeliveryStep
   bool fDistinct;
   bool fParallelOp;
 
-  LimitedOrderBy* fOrderBy;
+  std::unique_ptr<LimitedOrderBy> fOrderBy{nullptr};
   TupleConstantStep* fConstant;
 
   funcexp::FuncExp* fFeInstance;

utils/configcpp/configcpp.cpp

@@ -598,6 +598,7 @@ std::string Config::getTempFileDir(Config::TempDirPurpose what)
   {
     case TempDirPurpose::Joins: return prefix.append("joins/");
     case TempDirPurpose::Aggregates: return prefix.append("aggregates/");
+    case TempDirPurpose::Sorting: return prefix.append("sorting/");
   }
   // NOTREACHED
   return {};

utils/configcpp/configcpp.h

@@ -206,7 +206,8 @@ class Config
   enum class TempDirPurpose
   {
     Joins,      ///< disk joins
-    Aggregates  ///< disk-based aggregation
+    Aggregates, ///< disk-based aggregation
+    Sorting     ///< disk-based sorting
   };
   /** @brief Return temporaru directory path for the specified purpose */
   std::string getTempFileDir(TempDirPurpose what);

utils/rowgroup/dumper.cpp

@@ -24,20 +24,16 @@
 
 namespace common
 {
-  std::string errorString(int errNo)
-  {
-    char tmp[1024];
-    auto* buf = strerror_r(errNo, tmp, sizeof(tmp));
-    return {buf};
-  }
+std::string errorString(int errNo)
+{
+  char tmp[1024];
+  auto* buf = strerror_r(errNo, tmp, sizeof(tmp));
+  return {buf};
+}
 }  // namespace common
 
 namespace rowgroup
 {
-Dumper::Dumper(const compress::CompressInterface* comp, MemManager* mm) : fCompressor(comp), fMM(mm->clone())
-{
-}
-
 int Dumper::write(const std::string& fname, const char* buf, size_t sz)
 {
   if (sz == 0)
@@ -161,8 +157,8 @@ void Dumper::checkBuffer(size_t len)
 std::string RGDumper::makeRGFilename(uint64_t rgid, const uint16_t generation) const
 {
   char buf[PATH_MAX];
-  snprintf(buf, sizeof(buf), "%s/Sorting-p%u-t%ld-rg%lu-g%u", fTmpDir.c_str(), getpid(), fUniqId, rgid,
-           generation);
+  snprintf(buf, sizeof(buf), "%s/%s-p%u-t%ld-rg%lu-g%u", fOperationName.c_str(), fTmpDir.c_str(), getpid(),
+           fUniqId, rgid, generation);
   return buf;
 }
 

utils/rowgroup/dumper.h

@@ -25,7 +25,7 @@
 
 namespace common
 {
-  std::string errorString(int errNo);
+std::string errorString(int errNo);
 }
 
 namespace rowgroup
@@ -34,8 +34,11 @@ namespace rowgroup
 class Dumper
 {
  public:
-  Dumper(const compress::CompressInterface* comp, MemManager* mm);
-
+   Dumper(const compress::CompressInterface* comp, MemManager* mm)
+   : fCompressor(comp), fMM(std::unique_ptr<MemManager>(mm)) {}
+  Dumper(const compress::CompressInterface* comp, std::unique_ptr<MemManager>& mm)
+   : fCompressor(comp), fMM(std::move(mm)) {}
+  ~Dumper() = default;
   int write(const std::string& fname, const char* buf, size_t sz);
   int read(const std::string& fname, std::vector<char>& buf);
   size_t size() const;
@@ -49,17 +52,25 @@ class Dumper
   std::vector<char> fTmpBuf;
 };
 
-class RGDumper: protected Dumper
+class RGDumper : protected Dumper
 {
  public:
-  RGDumper(const compress::CompressInterface* comp, MemManager* mm, const std::string& tmpDir, const uint64_t uniqId) : Dumper(comp, mm), fTmpDir(tmpDir) { }
-  void loadRG(uint64_t rgid, const uint16_t generation, RowGroup& fRowGroupOut, std::unique_ptr<RGData>& rgdata, bool unlinkDump = false);
+  RGDumper(const compress::CompressInterface* comp, std::unique_ptr<MemManager> mm, const std::string& tmpDir,
+           const std::string& operationName, const uint64_t uniqId)
+   : Dumper(comp, mm), fTmpDir(tmpDir), fOperationName(operationName)
+  {
+  }
+  ~RGDumper() = default;
+  void loadRG(uint64_t rgid, const uint16_t generation, RowGroup& fRowGroupOut,
+              std::unique_ptr<RGData>& rgdata, bool unlinkDump = false);
   void saveRG(uint64_t rgid, const uint16_t generation, RowGroup& fRowGroupOut, RGData* rgdata);
 
   std::string makeRGFilename(uint64_t rgid, const uint16_t generation) const;
-private:
+
+ private:
   std::string fTmpDir;
+  std::string fOperationName;
   uint64_t fUniqId;
 };
 
-}
+}  // namespace rowgroup