HPCC-32138 Fix remapping issues

And add updateStatsDelta and updateRemappedStatsDelta template helpers Signed-off-by: Jake Smith <[email protected]>
hpcc-systems · Jul 26, 2024 · 342d6c6 · 342d6c6
1 parent a746d7f
commit 342d6c6
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 7 deletions.
diff --git a/system/jlib/jstats.h b/system/jlib/jstats.h
@@ -839,6 +839,21 @@ void mergeStat(CRuntimeStatisticCollection & stats, INTERFACE * source, Statisti
 template <class INTERFACE>
 void mergeStat(CRuntimeStatisticCollection & stats, const Shared<INTERFACE> & source, StatisticKind kind) { mergeStat(stats, source.get(), kind); }
 
+// helper templates that add delta of previous vs current (from source) to tgtStats (and update prevStats)
+template <class INTERFACE>
+void updateStatsDelta(CRuntimeStatisticCollection & tgtStats, CRuntimeStatisticCollection & prevStats, INTERFACE * source)
+{
+    CRuntimeStatisticCollection curStats(tgtStats.queryMapping());
+    mergeStats(curStats, source);
+    prevStats.updateDelta(tgtStats, curStats); // NB: adds delta to tgtStats, and updates prevStats
+}
+
+template <class INTERFACE>
+void updateStatsDelta(CRuntimeStatisticCollection & tgtStats, CRuntimeStatisticCollection & prevStats, const Shared<INTERFACE> & source)
+{
+    updateStatsDelta(tgtStats, prevStats, source.get());
+}
+
 
 //Some template helper classes for overwriting/setting statistics from external sources.
 
@@ -909,6 +924,21 @@ void mergeRemappedStats(CRuntimeStatisticCollection & stats, const Shared<INTERF
     mergeRemappedStats(stats, source.get(), stats.queryMapping(), remaps);
 }
 
+template <class INTERFACE>
+void updateRemappedStatsDelta(CRuntimeStatisticCollection & tgtStats, CRuntimeStatisticCollection & prevStats, INTERFACE * source, const StatKindMap & remap)
+{
+    CRuntimeStatisticCollection curStats(tgtStats.queryMapping());
+    ::mergeRemappedStats(curStats, source, remap);
+    prevStats.updateDelta(tgtStats, curStats); // NB: adds delta to tgtStats, and updates prevStats
+}
+
+template <class INTERFACE>
+void updateRemappedStatsDelta(CRuntimeStatisticCollection & tgtStats, CRuntimeStatisticCollection & prevStats, const Shared<INTERFACE> & source, const StatKindMap & remap)
+{
+    updateRemappedStatsDelta(tgtStats, prevStats, source.get(), remap);
+}
+
+
 //---------------------------------------------------------------------------------------------------------------------
 
 //A class for minimizing the overhead of collecting timestamps.

diff --git a/thorlcr/activities/nsplitter/thnsplitterslave.cpp b/thorlcr/activities/nsplitter/thnsplitterslave.cpp
@@ -405,7 +405,7 @@ class NSplitterSlaveActivity : public CSlaveActivity, implements ISharedSmartBuf
     {
         PARENT::gatherActiveStats(activeStats);
         if (sharedRowStream)
-            mergeRemappedStats(activeStats, sharedRowStream, diskToTempStatsMap);
+            ::mergeStats(activeStats, sharedRowStream);
     }
 // ISharedSmartBufferCallback impl.
     virtual void paged() { pagedOut = true; }

diff --git a/thorlcr/thorutil/thbuf.cpp b/thorlcr/thorutil/thbuf.cpp
@@ -2485,9 +2485,7 @@ class CSharedFullSpillingWriteAhead : public CInterfaceOf<ISharedRowStreamReader
             outputStream.clear();
             iFileIO->flush();
             tempFileOwner->noteSize(iFileIO->getStatistic(StSizeDiskWrite));
-            CRuntimeStatisticCollection currentFileStats(inactiveStats.queryMapping());
-            ::mergeStats(currentFileStats, iFileIO);
-            previousFileStats.updateDelta(inactiveStats, currentFileStats);
+            updateRemappedStatsDelta(inactiveStats, previousFileStats, iFileIO, diskToTempStatsMap); // NB: also updates prev to current
             previousFileStats.reset();
             iFileIO.clear();
         }
@@ -2543,9 +2541,7 @@ class CSharedFullSpillingWriteAhead : public CInterfaceOf<ISharedRowStreamReader
         outputStream->flush();
         totalInputRowsRead.fetch_add(newRowsWritten);
         tempFileOwner->noteSize(iFileIO->getStatistic(StSizeDiskWrite));
-        CRuntimeStatisticCollection currentFileStats(inactiveStats.queryMapping());
-        ::mergeStats(currentFileStats, iFileIO);
-        previousFileStats.updateDelta(inactiveStats, currentFileStats);
+        updateRemappedStatsDelta(inactiveStats, previousFileStats, iFileIO, diskToTempStatsMap); // NB: also updates prev to current
         // JCSMORE - could track size written, and start new file at this point (e.g. every 100MB),
         // and track their starting points (by row #) in a vector
         // We could then tell if/when the readers catch up, and remove consumed files as they do.