feat(sorting): basic dumper works for initial runs of TopNOrderBy.

Author: drrtuy

dbcon/joblist/CMakeLists.txt

@@ -27,6 +27,7 @@ set(joblist_LIB_SRCS
     jobstepassociation.cpp
     lbidlist.cpp
     limitedorderby.cpp
+    disk-based-topnorderby.cpp
     passthrucommand-jl.cpp
     passthrustep.cpp
     pcolscan.cpp

dbcon/joblist/disk-based-topnorderby.cpp

@@ -0,0 +1,58 @@
+/* Copyright (C) 2025 MariaDB Corp.
+
+   This program is free software; you can redistribute it and/or
+   modify it under the terms of the GNU General Public License
+   as published by the Free Software Foundation; version 2 of
+   the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+   MA 02110-1301, USA. */
+
+#include <vector>
+
+#include "dumper.h"
+#include "disk-based-topnorderby.h"
+namespace joblist
+{
+
+// The caller ensures lifetime of dl and rg
+void DiskBasedTopNOrderBy::flushCurrentToDisk(RowGroupDL& dl, rowgroup::RowGroup rg, const size_t numberOfRGs, const bool firstFlush)
+{
+  size_t rgid = (firstFlush) ? numberOfRGs : 0;
+  rowgroup::RGData rgData;
+
+  bool more = dl.next(0, &rgData);
+  while (more)
+  {
+    saveRG(rgid, getGenerationCounter(), rg, &rgData);
+    if (firstFlush)
+    {
+      --rgid;
+    }
+    else
+    {
+      ++rgid;
+    }
+    more = dl.next(0, &rgData);
+  }
+
+  incrementGenerationCounter();
+}
+void DiskBasedTopNOrderBy::diskBasedMergePhaseIfNeeded(std::vector<RowGroupDLSPtr>& dataLists)
+{
+}
+
+std::vector<std::string> DiskBasedTopNOrderBy::getGenerationFileNamesNextBatch(const size_t batchSize)
+{
+  return {};
+}
+
+
+}  // namespace joblist

dbcon/joblist/disk-based-topnorderby.h

@@ -60,18 +60,11 @@ class DiskBasedTopNOrderBy : public rowgroup::RGDumper
   {
     return 0;
   }
-  std::vector<std::string> getGenerationFileNamesNextBatch(const size_t batchSize)
-  {
-    return {};
-  }
+  std::vector<std::string> getGenerationFileNamesNextBatch(const size_t batchSize);
 
-  void flushCurrentToDisk(const bool firstFlush)
-  {
-    incrementGenerationCounter();
-  }
-  void diskBasedMergePhaseIfNeeded(std::vector<RowGroupDLSPtr>& dataLists)
-  {
-  }
+  // The caller ensures lifetime of dl and rg
+  void flushCurrentToDisk(RowGroupDL& dl, rowgroup::RowGroup rg, const size_t numberOfRGs, const bool firstFlush);
+  void diskBasedMergePhaseIfNeeded(std::vector<RowGroupDLSPtr>& dataLists);
 
   //  private:
   uint64_t fGenerationCounter{0};

dbcon/joblist/limitedorderby.cpp

@@ -110,8 +110,8 @@ 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)
@@ -162,90 +162,8 @@ 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;
-
-    copyRow(row, &row1);
-
-    if (fDistinct)
-    {
-      fDistinctMap->erase(orderedRowsQueue.top().fData);
-      fDistinctMap->insert(row1.getPointer());
-    }
-
-    orderedRowsQueue.pop();
-    orderedRowsQueue.push(swapRow);
-  }
-}
-
-void LimitedOrderBy::processRow_(const rowgroup::Row& row)
-{
-  // check if this is a distinct row
-  if (fDistinct && fDistinctMap->find(row.getPointer()) != fDistinctMap->end())
-    return;
-
-  // @bug5312, limit count is 0, do nothing.
-  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;
-  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);
-
-    uint64_t memSizeInc = sizeof(newRow);
-    fUncommitedMemory += memSizeInc;
-    if (fUncommitedMemory >= fMaxUncommited)
-    {
-      if (!fRm->getMemory(fUncommitedMemory, fSessionMemLimit))
-      {
-        cerr << IDBErrorInfo::instance()->errorMsg(fErrorCode) << " @" << __FILE__ << ":" << __LINE__;
-        throw IDBExcept(fErrorCode);
-      }
-      fMemSize += fUncommitedMemory;
-      fUncommitedMemory = 0;
-    }
-
-    // add to the distinct map
-    if (fDistinct)
-      fDistinctMap->insert(fRow0.getPointer());
-
-    fRowGroup.incRowCount();
-    fRow0.nextRow();
-
-    if (fRowGroup.getRowCount() >= fRowsPerRG)
-    {
-      fDataQueue.push(fData);
-      uint64_t newSize = fRowGroup.getSizeWithStrings() - fRowGroup.getHeaderSize();
-
-      if (!fRm->getMemory(newSize, fSessionMemLimit))
-      {
-        cerr << IDBErrorInfo::instance()->errorMsg(fErrorCode) << " @" << __FILE__ << ":" << __LINE__;
-        throw IDBExcept(fErrorCode);
-      }
-      fMemSize += newSize;
-
-      fData.reinit(fRowGroup, fRowsPerRG);
-      fRowGroup.setData(&fData);
-      fRowGroup.resetRowGroup(0);
-      fRowGroup.getRow(0, &fRow0);
-    }
-  }
-  else if (fOrderByCond.size() > 0 && fRule.less(row.getPointer(), orderedRowsQueue.top().fData))
-  {
-    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 row2swap " << row1.toString() << std::endl;
+    // std::cout << "LimitedOrderBy::processRow new row 4 swaping " << row.toString() << std::endl;
 
     copyRow(row, &row1);
 
@@ -277,118 +195,93 @@ void LimitedOrderBy::processRow_(const rowgroup::Row& row)
 //   }
 // }
 
-void LimitedOrderBy::brandNewFinalize()
+void LimitedOrderBy::flushCurrentToDisk_(const bool firstFlush)
 {
-  if (!isDiskBased())
-  {
-    return finalize();
-  }
-
-  // if disk-based
-  // here there are <= inputQueuesNumber files on disk
-  // and potentially some in-memory state
-  // need to merge this together to produce a result
+  // make a queue with rgdatas and hand it to DiskBasedTopNOrderBy
+  auto dl = RowGroupDL(1, 1);
+  auto& orderedRowsQueue = getQueue();
+  size_t rowsOverRG = orderedRowsQueue.size() % fRowsPerRG;
+  size_t numberOfRGs = orderedRowsQueue.size() / fRowsPerRG + static_cast<size_t>(rowsOverRG > 0);
+  std::thread flushThread(&DiskBasedTopNOrderBy::flushCurrentToDisk, this, std::ref(dl), fRowGroup,
+                          numberOfRGs, firstFlush);
 
-  if (fUncommitedMemory > 0)
+  uint32_t rSize = fRow0.getSize();
+  // process leftovers
+  if (rowsOverRG)
   {
-    if (!fRm->getMemory(fUncommitedMemory, fSessionMemLimit))
-    {
-      cerr << IDBErrorInfo::instance()->errorMsg(fErrorCode) << " @" << __FILE__ << ":" << __LINE__;
-      throw logging::OutOfMemoryExcept(fErrorCode);
-    }
-    fMemSize += fUncommitedMemory;
-    fUncommitedMemory = 0;
-  }
-
-  queue<RGData> tempQueue;
-  if (fRowGroup.getRowCount() > 0)
-    fDataQueue.push(fData);
+    fData.reinit(fRowGroup, rowsOverRG);
+    fRowGroup.setData(&fData);
+    fRowGroup.resetRowGroup(0);
+    fRowGroup.getRow(rowsOverRG-1, &fRow0);
+    
+    const OrderByRow& topRow = orderedRowsQueue.top();
+    row1.setData(topRow.fData);
+    copyRow(row1, &fRow0);
+    fRowGroup.incRowCount();
+    fRow0.prevRow(rSize);
+    orderedRowsQueue.pop();
 
-  auto& orderedRowsQueue = getQueue();
+    dl.insert(fData);
+  }
 
   if (orderedRowsQueue.size() > 0)
   {
-    // *DRRTUY Very memory intensive. CS needs to account active
-    // memory only and release memory if needed.
-    uint64_t memSizeInc = fRowGroup.getSizeWithStrings() - fRowGroup.getHeaderSize();
-
-    if (!fRm->getMemory(memSizeInc, fSessionMemLimit))
-    {
-      cerr << IDBErrorInfo::instance()->errorMsg(fErrorCode) << " @" << __FILE__ << ":" << __LINE__;
-      throw logging::OutOfMemoryExcept(fErrorCode);
-    }
-    fMemSize += memSizeInc;
-
-    uint64_t offset = 0;
-    uint64_t i = 0;
-    // Reduce queue size by an offset value if it applicable.
-    uint64_t queueSizeWoOffset = orderedRowsQueue.size() > fStart ? orderedRowsQueue.size() - fStart : 0;
-    list<RGData> tempRGDataList;
-
-    if (fCount <= queueSizeWoOffset)
-    {
-      offset = fCount % fRowsPerRG;
-      if (!offset && fCount > 0)
-        offset = fRowsPerRG;
-    }
-    else
-    {
-      offset = queueSizeWoOffset % fRowsPerRG;
-      if (!offset && queueSizeWoOffset > 0)
-        offset = fRowsPerRG;
-    }
-
-    list<RGData>::iterator tempListIter = tempRGDataList.begin();
-
-    i = 0;
-    uint32_t rSize = fRow0.getSize();
-    uint64_t preLastRowNumb = fRowsPerRG - 1;
     fData.reinit(fRowGroup, fRowsPerRG);
     fRowGroup.setData(&fData);
     fRowGroup.resetRowGroup(0);
-    // *DRRTUY This approach won't work with
-    // OFSET > fRowsPerRG
-    offset = offset != 0 ? offset - 1 : offset;
-    fRowGroup.getRow(offset, &fRow0);
+    fRowGroup.getRow(fRowsPerRG-1, &fRow0);
 
-    while ((orderedRowsQueue.size() > fStart) && (i++ < fCount))
+    while (!orderedRowsQueue.empty())
     {
       const OrderByRow& topRow = orderedRowsQueue.top();
       row1.setData(topRow.fData);
       copyRow(row1, &fRow0);
       fRowGroup.incRowCount();
-      offset--;
       fRow0.prevRow(rSize);
       orderedRowsQueue.pop();
 
-      // if RG has fRowsPerRG rows
-      if (offset == (uint64_t)-1)
+      if (fRowGroup.getRowCount() == fRowsPerRG)
       {
-        tempRGDataList.push_front(fData);
-
-        if (!fRm->getMemory(memSizeInc, fSessionMemLimit))
-        {
-          cerr << IDBErrorInfo::instance()->errorMsg(fErrorCode) << " @" << __FILE__ << ":" << __LINE__;
-          throw logging::OutOfMemoryExcept(fErrorCode);
-        }
-        fMemSize += memSizeInc;
+        dl.insert(fData);
 
         fData.reinit(fRowGroup, fRowsPerRG);
         fRowGroup.setData(&fData);
-        fRowGroup.resetRowGroup(0);  // ?
-        fRowGroup.getRow(preLastRowNumb, &fRow0);
-        offset = preLastRowNumb;
+        fRowGroup.resetRowGroup(0);
+        fRowGroup.getRow(fRowsPerRG-1, &fRow0);
       }
     }
-    // Push the last/only group into the queue.
+
     if (fRowGroup.getRowCount() > 0)
-      tempRGDataList.push_front(fData);
+      dl.insert(fData);
+  }
 
-    for (tempListIter = tempRGDataList.begin(); tempListIter != tempRGDataList.end(); tempListIter++)
-      tempQueue.push(*tempListIter);
+  dl.endOfInput();
 
-    fDataQueue = tempQueue;
+  // clean up the current queue/rgdatas to free mem
+  // fDataQueue
+  // fDistinctMap
+  // orderedRowsQueue
+  queue<RGData> tempQueue;
+  fDataQueue.swap(tempQueue);
+  if (fDistinctMap)
+  {
+    fDistinctMap->clear();
   }
+
+  flushThread.join();
+}
+
+void LimitedOrderBy::brandNewFinalize()
+{
+  if (!isDiskBased())
+  {
+    return finalize();
+  }
+
+  // if disk-based
+  // here there are <= inputQueuesNumber files on disk
+  // and potentially some in-memory state
+  // need to merge this together to produce a result
 }
 
 /*

dbcon/joblist/limitedorderby.h

@@ -54,6 +54,8 @@ class LimitedOrderBy : public ordering::IdbOrderBy, public DiskBasedTopNOrderBy
   }
   const std::string toString() const override;
 
+  void flushCurrentToDisk_(const bool firstFlush);
+  
   void finalize();
   void brandNewFinalize();
   bool getNextRGData(rowgroup::RGData& data);