HPCC-32138 Changes following review

Signed-off-by: Shamser Ahmed <[email protected]>
hpcc-systems · Jul 19, 2024 · a746d7f · a746d7f
1 parent 385d8f3
commit a746d7f
Showing 1 changed file with 10 additions and 6 deletions.
diff --git a/thorlcr/thorutil/thbuf.cpp b/thorlcr/thorutil/thbuf.cpp
@@ -2452,6 +2452,7 @@ class CSharedFullSpillingWriteAhead : public CInterfaceOf<ISharedRowStreamReader
     SharedRowStreamReaderOptions options;
     size32_t inMemReadAheadGranularity = 0;
     CRuntimeStatisticCollection inactiveStats;
+    CRuntimeStatisticCollection previousFileStats;
     StringAttr baseTmpFilename;
 
 
@@ -2484,7 +2485,10 @@ class CSharedFullSpillingWriteAhead : public CInterfaceOf<ISharedRowStreamReader
             outputStream.clear();
             iFileIO->flush();
             tempFileOwner->noteSize(iFileIO->getStatistic(StSizeDiskWrite));
-            ::mergeStats(inactiveStats, iFileIO);
+            CRuntimeStatisticCollection currentFileStats(inactiveStats.queryMapping());
+            ::mergeStats(currentFileStats, iFileIO);
+            previousFileStats.updateDelta(inactiveStats, currentFileStats);
+            previousFileStats.reset();
             iFileIO.clear();
         }
     }
@@ -2539,6 +2543,9 @@ 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);
         // 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.
@@ -2552,7 +2559,7 @@ class CSharedFullSpillingWriteAhead : public CInterfaceOf<ISharedRowStreamReader
     explicit CSharedFullSpillingWriteAhead(CActivityBase *_activity, unsigned _numOutputs, IRowStream *_input, bool _inputGrouped, const SharedRowStreamReaderOptions &_options, IThorRowInterfaces *rowIf, const char *_baseTmpFilename, ICompressHandler *_compressHandler)
         : activity(*_activity), numOutputs(_numOutputs), input(_input), inputGrouped(_inputGrouped), options(_options), compressHandler(_compressHandler), baseTmpFilename(_baseTmpFilename),
         meta(rowIf->queryRowMetaData()), serializer(rowIf->queryRowSerializer()), allocator(rowIf->queryRowAllocator()), deserializer(rowIf->queryRowDeserializer()),
-        inactiveStats(spillingWriteAheadStatistics)
+        inactiveStats(spillingWriteAheadStatistics), previousFileStats(spillingWriteAheadStatistics)
     {
         assertex(input);
 
@@ -2716,10 +2723,7 @@ class CSharedFullSpillingWriteAhead : public CInterfaceOf<ISharedRowStreamReader
     }
     virtual unsigned __int64 getStatistic(StatisticKind kind) const override
     {
-        unsigned __int64 v = inactiveStats.getStatisticValue(kind);
-        if (iFileIO)
-            v += iFileIO->getStatistic(kind);
-        return v;
+        return inactiveStats.getStatisticValue(kind);
     }
 };