Merge pull request #9262 from ghalliday/issue16385

HPCC-16385 Add support for number of allocation scans statistic

Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

common/thorhelper/roxierow.cpp

@@ -305,6 +305,11 @@ public:
         return heap->finalizeRow(row);
     }
 
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) override
+    {
+        heap->gatherStats(stats);
+    }
+
 protected:
     Owned<roxiemem::IFixedRowHeap> heap;
 };
@@ -361,6 +366,11 @@ public:
         return newrow;
     }
 
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) override
+    {
+        heap->gatherStats(stats);
+    }
+
 protected:
     Owned<roxiemem::IVariableRowHeap> heap;
 };

ecl/hqlcpp/hqlcppds.cpp

@@ -3025,6 +3025,7 @@ public:
     virtual IOutputRowSerializer *createInternalSerializer(ICodeContext *ctx = NULL) { throwUnexpected(); }
     virtual IOutputRowDeserializer *createInternalDeserializer(ICodeContext *ctx) { throwUnexpected(); }
     virtual IEngineRowAllocator *createChildRowAllocator(const RtlTypeInfo *type) { throwUnexpected(); }
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) override {}
 };
 
 //Use a (constant) transform to map selectors of the form queryActiveTableSelector().field

roxie/ccd/ccdserver.cpp

@@ -349,7 +349,8 @@ protected:
 // General activity statistics
 
 static const StatisticsMapping actStatistics(StWhenFirstRow, StTimeElapsed, StTimeLocalExecute, StTimeTotalExecute, StSizeMaxRowSize,
-                                              StNumRowsProcessed, StNumSlaves, StNumStarted, StNumStopped, StNumStrands, StKindNone);
+                                              StNumRowsProcessed, StNumSlaves, StNumStarted, StNumStopped, StNumStrands,
+                                              StNumScansPerRow, StNumAllocations, StNumAllocationScans, StKindNone);
 static const StatisticsMapping joinStatistics(&actStatistics, StNumAtmostTriggered, StKindNone);
 static const StatisticsMapping keyedJoinStatistics(&joinStatistics, StNumServerCacheHits, StNumIndexSeeks, StNumIndexScans, StNumIndexWildSeeks,
                                                     StNumIndexSkips, StNumIndexNullSkips, StNumIndexMerges, StNumIndexMergeCompares,
@@ -1069,10 +1070,20 @@ public:
         totalCycles.merge(strandCycles);
         stats.merge(strandStats);
     }
-    inline void createRowAllocator()
+    inline void ensureRowAllocator()
     {
         if (!rowAllocator) 
-            rowAllocator = ctx->getRowAllocatorEx(meta.queryOriginal(), activityId, factory->getHeapFlags());
+            rowAllocator = createRowAllocator(meta.queryOriginal());
+    }
+
+    virtual IEngineRowAllocator * createRowAllocator(IOutputMetaData * metadata)
+    {
+        return ctx->getRowAllocatorEx(metadata, activityId, factory->getHeapFlags());
+    }
+
+    virtual IEngineRowAllocator * createRowAllocatorEx(IOutputMetaData * metadata, roxiemem::RoxieHeapFlags extraFlags)
+    {
+        return ctx->getRowAllocatorEx(metadata, activityId, (roxiemem::RoxieHeapFlags)(factory->getHeapFlags()|extraFlags));
     }
 
     inline ICodeContext *queryCodeContext()
@@ -1197,7 +1208,7 @@ public:
         colocalParent = _colocalParent;
         createPending = true;
         if (needsAllocator())
-            createRowAllocator();
+            ensureRowAllocator();
         processed = 0;
         if (factory)
             factory->onCreateChildQueries(ctx, &basehelper, childGraphs, this, probeManager, *this, _numParallel);
@@ -1329,6 +1340,11 @@ public:
                 }
                 if (inputStream)
                     inputStream->stop();
+                if (rowAllocator)
+                {
+                    stats.reset(heapStatistics);
+                    rowAllocator->gatherStats(stats);
+                }
             }
         }
     }
@@ -1651,8 +1667,7 @@ public:
     {
         if (needsAllocator)
         {
-            roxiemem::RoxieHeapFlags extraFlags = _parent.factory->getHeapFlags();
-            rowAllocator = parent.queryContext()->getRowAllocatorEx(parent.queryOutputMeta(), parent.queryId(), (roxiemem::RoxieHeapFlags)(roxiemem::RHFunique|extraFlags));
+            rowAllocator = parent.createRowAllocatorEx(parent.queryOutputMeta(), roxiemem::RHFunique);
         }
         else
             rowAllocator = NULL;
@@ -1674,6 +1689,10 @@ public:
         {
             if (inputStream)
                 inputStream->stop();
+
+            stats.reset(heapStatistics); // Heap stats are always gathered from scratch each time
+            if (rowAllocator)
+                rowAllocator->gatherStats(stats);
             parent.stop();
             parent.mergeStrandStats(processed, totalCycles, stats);
         }
@@ -4241,7 +4260,7 @@ public:
         if (meta.needsSerializeDisk())
         {
             deserializer.setown(meta.createDiskDeserializer(ctx->queryCodeContext(), activity.queryId()));
-            rowAllocator.setown(ctx->queryCodeContext()->getRowAllocator(meta.queryOriginal(), activity.queryId()));
+            rowAllocator.setown(activity.createRowAllocator(meta.queryOriginal()));
         }
         if (ctx->queryDebugContext() && ctx->queryDebugContext()->getExecuteSequentially())
             deferredStart = true;
@@ -7497,9 +7516,9 @@ class CRoxieServerHashDedupActivity : public CRoxieServerActivity
     class HashDedupTable : public SuperHashTable
     {
     public:
-        HashDedupTable(IHThorHashDedupArg & _helper, unsigned _activityId) 
+        HashDedupTable(IHThorHashDedupArg & _helper, CRoxieServerHashDedupActivity & _activity)
             : helper(_helper), 
-              activityId(_activityId),
+              activity(_activity),
               keySize(helper.queryKeySize())
         {
         }
@@ -7529,7 +7548,7 @@ class CRoxieServerHashDedupActivity : public CRoxieServerActivity
 
         void onCreate(IRoxieSlaveContext *ctx)
         {
-            keyRowAllocator.setown(ctx->queryCodeContext()->getRowAllocator(keySize.queryOriginal(), activityId));
+            keyRowAllocator.setown(activity.createRowAllocator(keySize.queryOriginal()));
         }
 
         void reset()
@@ -7553,12 +7572,12 @@ class CRoxieServerHashDedupActivity : public CRoxieServerActivity
         IHThorHashDedupArg & helper;
         CachedOutputMetaData keySize;
         Owned<IEngineRowAllocator> keyRowAllocator;
-        unsigned activityId;
+        CRoxieServerHashDedupActivity & activity;
     } table;
 
 public:
     CRoxieServerHashDedupActivity(IRoxieSlaveContext *_ctx, const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager)
-        : CRoxieServerActivity(_ctx, _factory, _probeManager), helper((IHThorHashDedupArg &)basehelper), table(helper, activityId)
+        : CRoxieServerActivity(_ctx, _factory, _probeManager), helper((IHThorHashDedupArg &)basehelper), table(helper, *this)
     {
         eof = false;
     }
@@ -12228,7 +12247,7 @@ class CRoxieServerJoinActivity : public CRoxieServerTwoInputActivity
         if (!defaultLeft)
         {
             if (!defaultLeftAllocator)
-                defaultLeftAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input->queryOutputMeta(), activityId));
+                defaultLeftAllocator.setown(createRowAllocator(input->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultLeftAllocator);
             size32_t thisSize = helper.createDefaultLeft(rowBuilder);
@@ -12241,7 +12260,7 @@ class CRoxieServerJoinActivity : public CRoxieServerTwoInputActivity
         if (!defaultRight)
         {
             if (!defaultRightAllocator)
-                defaultRightAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input1->queryOutputMeta(), activityId));
+                defaultRightAllocator.setown(createRowAllocator(input1->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultRightAllocator);
             size32_t thisSize = helper.createDefaultRight(rowBuilder);
@@ -16736,8 +16755,8 @@ public:
     {
         CRoxieServerNaryActivity::onCreate(_colocalParent);
         ICodeContext * codectx = ctx->queryCodeContext();
-        inputAllocator.setown(codectx->getRowAllocator(helper.queryInputMeta(), activityId));
-        outputAllocator.setown(codectx->getRowAllocator(helper.queryOutputMeta(), activityId));
+        inputAllocator.setown(createRowAllocator(helper.queryInputMeta()));
+        outputAllocator.setown(createRowAllocator(helper.queryOutputMeta()));
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)
@@ -17230,7 +17249,7 @@ public:
     virtual void onCreate(IHThorArg *_colocalParent)
     {
         CRoxieServerActivity::onCreate(_colocalParent);
-        rightRowAllocator.setown(ctx->queryCodeContext()->getRowAllocator(QUERYINTERFACE(helper.queryRightRecordSize(), IOutputMetaData), activityId));
+        rightRowAllocator.setown(createRowAllocator(helper.queryRightRecordSize()));
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)
@@ -17547,7 +17566,7 @@ public:
     const void *defaultRow()
     {
         if (!rowAllocator)
-            createRowAllocator();      // We delay as often not needed...
+            ensureRowAllocator();      // We delay as often not needed...
         RtlDynamicRowBuilder rowBuilder(rowAllocator);
         size32_t thisSize = helper.createDefault(rowBuilder);
         return rowBuilder.finalizeRowClear(thisSize);
@@ -17756,7 +17775,7 @@ class CRoxieServerSelfJoinActivity : public CRoxieServerActivity
         if (!defaultLeft)
         {
             if (!defaultAllocator)
-                defaultAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input->queryOutputMeta(), activityId));
+                defaultAllocator.setown(createRowAllocator(input->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultAllocator);
             size32_t thisSize = helper.createDefaultLeft(rowBuilder);
@@ -17769,7 +17788,7 @@ class CRoxieServerSelfJoinActivity : public CRoxieServerActivity
         if (!defaultRight)
         {
             if (!defaultAllocator)
-                defaultAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input->queryOutputMeta(), activityId));
+                defaultAllocator.setown(createRowAllocator(input->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultAllocator);
             size32_t thisSize = helper.createDefaultRight(rowBuilder);
@@ -18333,7 +18352,7 @@ private:
         if (!defaultRight)
         {
             if (!defaultRightAllocator)
-                defaultRightAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input1->queryOutputMeta(), activityId));
+                defaultRightAllocator.setown(createRowAllocator(input1->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultRightAllocator);
             size32_t thisSize = helper.createDefaultRight(rowBuilder);
@@ -18857,7 +18876,7 @@ private:
         if (!defaultLeft)
         {
             if (!defaultLeftAllocator)
-                defaultLeftAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input->queryOutputMeta(), activityId));
+                defaultLeftAllocator.setown(createRowAllocator(input->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultLeftAllocator);
             size32_t thisSize = helper.createDefaultLeft(rowBuilder);
@@ -18870,7 +18889,7 @@ private:
         if (!defaultRight)
         {
             if (!defaultRightAllocator)
-                defaultRightAllocator.setown(ctx->queryCodeContext()->getRowAllocator(input1->queryOutputMeta(), activityId));
+                defaultRightAllocator.setown(createRowAllocator(input1->queryOutputMeta()));
 
             RtlDynamicRowBuilder rowBuilder(defaultRightAllocator);
             size32_t thisSize = helper.createDefaultRight(rowBuilder);
@@ -19536,7 +19555,7 @@ protected:
                     ctx->queryDebugContext()->checkBreakpoint(DebugStateLimit, NULL, static_cast<IActivityBase *>(this));
                 if (transformExtra)
                 {
-                    createRowAllocator();
+                    ensureRowAllocator();
                     RtlDynamicRowBuilder rowBuilder(rowAllocator);
                     size32_t outSize = transformExtra->transformOnLimitExceeded(rowBuilder);
                     if (outSize)
@@ -19708,7 +19727,7 @@ protected:
         ReleaseRoxieRows(buff);
         if (createRow)
         {
-            createRowAllocator();
+            ensureRowAllocator();
             RtlDynamicRowBuilder rowBuilder(rowAllocator);
             size32_t outSize = helper.transformOnExceptionCaught(rowBuilder, E);
             if (outSize)
@@ -20891,7 +20910,7 @@ public:
         }
         if (workunit != NULL || (results && protocol->getFlags() & HPCC_PROTOCOL_NATIVE_RAW))
         {
-            createRowAllocator();
+            ensureRowAllocator();
             rowSerializer.setown(rowAllocator->createDiskSerializer(ctx->queryCodeContext()));
         }
         __int64 initialProcessed = processed;
@@ -22730,7 +22749,7 @@ public:
 protected:
     virtual const void * createLimitFailRow(bool isKeyed)
     {
-        createRowAllocator();
+        ensureRowAllocator();
         RtlDynamicRowBuilder rowBuilder(rowAllocator);
         size32_t outSize = isKeyed ? limitTransformExtra->transformOnKeyedLimitExceeded(rowBuilder) : limitTransformExtra->transformOnLimitExceeded(rowBuilder);
         if (outSize)
@@ -22902,7 +22921,7 @@ public:
 
     virtual bool processSingleKey(IKeyIndex *key, IRecordLayoutTranslator * trans)
     {
-        createRowAllocator();
+        ensureRowAllocator();
         remote.injectResult(new LazyLocalKeyReader(*this, key, trans));
         return false;
     }
@@ -24458,7 +24477,7 @@ public:
             partNo = map->mapOffset(rp);
         if (needsRHS)
         {
-            Owned<IEngineRowAllocator> extractAllocator = ctx->queryCodeContext()->getRowAllocator(helper.queryExtractedSize(), activityId);
+            Owned<IEngineRowAllocator> extractAllocator = createRowAllocator(helper.queryExtractedSize());
             RtlDynamicRowBuilder rb(extractAllocator, true);
             unsigned rhsSize = helper.extractJoinFields(rb, row);
             char * block = (char *) remote.getMem(partNo, 0, sizeof(rp) + sizeof(rhsSize) + rhsSize); // MORE - superfiles
@@ -24917,7 +24936,7 @@ public:
     virtual void onCreate(IHThorArg *_colocalArg)
     {
         CRemoteResultAdaptor::onCreate(_colocalArg);
-        ccdRecordAllocator.setown(ctx->queryCodeContext()->getRowAllocator(QUERYINTERFACE(helper.queryJoinFieldsRecordSize(), IOutputMetaData), activityId));
+        ccdRecordAllocator.setown(activity.createRowAllocator(helper.queryJoinFieldsRecordSize()));
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)
@@ -25077,7 +25096,7 @@ public:
     {
         CRoxieServerActivity::onCreate(_colocalParent);
         remote.onCreate(_colocalParent);
-        indexReadAllocator.setown(ctx->queryCodeContext()->getRowAllocator(indexReadMeta, activityId));
+        indexReadAllocator.setown(createRowAllocator(indexReadMeta));
     }
 
     virtual void setInput(unsigned idx, unsigned _sourceIdx, IFinalRoxieInput *_in)
@@ -25396,7 +25415,7 @@ protected:
         if (!defaultRight)
         {
             if (!defaultRightAllocator)
-                defaultRightAllocator.setown(ctx->queryCodeContext()->getRowAllocator(helper.queryJoinFieldsRecordSize(), activityId));
+                defaultRightAllocator.setown(createRowAllocator(helper.queryJoinFieldsRecordSize()));
 
             RtlDynamicRowBuilder rowBuilder(defaultRightAllocator);
             size32_t thisSize = helper.createDefaultRight(rowBuilder);
@@ -25822,7 +25841,7 @@ public:
         CRoxieServerKeyedJoinBase::onCreate(_colocalParent);
         head.onCreate(_colocalParent);
         fetchInputFields.set(helper.queryFetchInputRecordSize());
-        fetchInputAllocator.setown(ctx->queryCodeContext()->getRowAllocator(helper.queryFetchInputRecordSize(), activityId));
+        fetchInputAllocator.setown(createRowAllocator(helper.queryFetchInputRecordSize()));
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)
@@ -25959,11 +25978,11 @@ public:
     virtual void onCreate(IHThorArg *_colocalParent)
     {
         CRoxieServerKeyedJoinBase::onCreate(_colocalParent);
-        indexReadAllocator.setown(ctx->queryCodeContext()->getRowAllocator(indexReadMeta, activityId));
+        indexReadAllocator.setown(createRowAllocator(indexReadMeta));
 
         IOutputMetaData *joinFieldsMeta = helper.queryJoinFieldsRecordSize();
         joinPrefixedMeta.setown(new CPrefixedOutputMeta(KEYEDJOIN_RECORD_SIZE(0), joinFieldsMeta)); // MORE - not sure if we really need this
-        joinFieldsAllocator.setown(ctx->queryCodeContext()->getRowAllocator(joinPrefixedMeta, activityId));
+        joinFieldsAllocator.setown(createRowAllocator(joinPrefixedMeta));
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)

roxie/ccd/ccdserver.hpp

@@ -185,6 +185,7 @@ interface IRoxieServerActivity : extends IActivityBase
     virtual void mergeStats(MemoryBuffer &stats) = 0;
     virtual ISectionTimer * registerTimer(unsigned activityId, const char * name) = 0;
     virtual IRoxieServerActivity * queryChildActivity(unsigned activityId) = 0;
+    virtual IEngineRowAllocator * createRowAllocator(IOutputMetaData * metadata) = 0;
 };
 
 interface IRoxieServerActivityFactory : extends IActivityFactory

roxie/roxiemem/roxiemem.cpp

@@ -2349,6 +2349,8 @@ public:
     {
     }
 
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) override {}
+
     virtual void *allocate();
 
 protected:
@@ -2374,6 +2376,11 @@ public:
         }
     }
 
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) override
+    {
+        heap->gatherStats(stats);
+    }
+
     virtual void *allocate()
     {
         return heap->allocate(allocatorId);
@@ -2440,6 +2447,7 @@ public:
     }
     IMPLEMENT_IINTERFACE
 
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) override {}
     virtual void *allocate(memsize_t size, memsize_t & capacity);
     virtual void *resizeRow(void * original, memsize_t copysize, memsize_t newsize, memsize_t &capacity);
     virtual void *finalizeRow(void *final, memsize_t originalSize, memsize_t finalSize);
@@ -2979,6 +2987,8 @@ public:
     void checkScans();
     virtual void reportScanProblem(unsigned __int64 numScans, const HeapletStats & mergedStats) = 0;
 
+    void gatherStats(CRuntimeStatisticCollection & stats);
+
 protected:
     void * doAllocateRow(unsigned allocatorId, unsigned maxSpillCost);
     unsigned doAllocateRowBlock(unsigned allocatorId, unsigned maxSpillCost, unsigned max, char * * rows);
@@ -5050,6 +5060,31 @@ void CHugeHeap::expandHeap(void * original, memsize_t copysize, memsize_t oldcap
 }
 
 
+void CChunkedHeap::gatherStats(CRuntimeStatisticCollection & result)
+{
+    HeapletStats merged(stats);
+
+    NonReentrantSpinBlock b(heapletLock);
+    Heaplet * start = heaplets;
+    if (start)
+    {
+        Heaplet * finger = start;
+        loop
+        {
+            finger->mergeStats(merged);
+            finger = getNext(finger);
+            if (finger == start)
+                break;
+        }
+    }
+
+    if (merged.totalAllocs)
+    {
+        result.addStatistic(StNumAllocations, merged.totalAllocs);
+        result.addStatistic(StNumAllocationScans, merged.totalDistanceScanned / chunkSize);
+    }
+}
+
 void * CChunkedHeap::doAllocateRow(unsigned allocatorId, unsigned maxSpillCost)
 {
     //Only hold the spinblock while walking the list - so subsequent calls to checkLimit don't deadlock.

roxie/roxiemem/roxiemem.hpp

@@ -381,14 +381,20 @@ private:
     const char * ptr;
 };
 
-interface IFixedRowHeap : extends IInterface
+
+interface IRowHeap : extends IInterface
+{
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) = 0;
+};
+
+interface IFixedRowHeap : extends IRowHeap
 {
     virtual void *allocate() = 0;
     virtual void *finalizeRow(void *final) = 0;
     virtual void emptyCache() = 0;
 };
 
-interface IVariableRowHeap : extends IInterface
+interface IVariableRowHeap : extends IRowHeap
 {
     virtual void *allocate(memsize_t size, memsize_t & capacity) = 0;
     virtual void *resizeRow(void * original, memsize_t copysize, memsize_t newsize, memsize_t &capacity) = 0;

rtl/include/eclhelper.hpp

@@ -236,6 +236,7 @@ public:
 interface IOutputRowSerializer;
 interface IOutputRowDeserializer;
 
+class CRuntimeStatisticCollection;
 interface IEngineRowAllocator : extends IInterface
 {
     virtual byte * * createRowset(unsigned _numItems) = 0;
@@ -261,6 +262,8 @@ interface IEngineRowAllocator : extends IInterface
     virtual IOutputRowSerializer *createInternalSerializer(ICodeContext *ctx = NULL) = 0;
     virtual IOutputRowDeserializer *createInternalDeserializer(ICodeContext *ctx) = 0;
     virtual IEngineRowAllocator *createChildRowAllocator(const RtlTypeInfo *childtype) = 0;
+
+    virtual void gatherStats(CRuntimeStatisticCollection & stats) = 0;
 };
 
 interface IRowSerializerTarget

system/jlib/jstatcodes.h

@@ -180,6 +180,9 @@ enum StatisticKind
     StTimeTotalNested,
     StCycleLocalExecuteCycles,
     StNumCompares,
+    StNumScansPerRow,
+    StNumAllocations,
+    StNumAllocationScans,
 
     StMax,
 

system/jlib/jstats.cpp

@@ -613,6 +613,9 @@ static const StatisticMeta statsMetaData[StMax] = {
     { TIMESTAT(TotalNested) },
     { CYCLESTAT(LocalExecute) },
     { NUMSTAT(Compares) },
+    { NUMSTAT(ScansPerRow) },
+    { NUMSTAT(Allocations) },
+    { NUMSTAT(AllocationScans) },
 };
 
 
@@ -925,6 +928,7 @@ void StatisticsMapping::createMappings()
 }
 
 const StatisticsMapping allStatistics;
+const StatisticsMapping heapStatistics(StNumAllocations, StNumAllocationScans, StKindNone);
 const StatisticsMapping diskLocalStatistics(StCycleDiskReadIOCycles, StSizeDiskRead, StNumDiskReads, StCycleDiskWriteIOCycles, StSizeDiskWrite, StNumDiskWrites, StKindNone);
 const StatisticsMapping diskRemoteStatistics(StTimeDiskReadIO, StSizeDiskRead, StNumDiskReads, StTimeDiskWriteIO, StSizeDiskWrite, StNumDiskWrites, StKindNone);
 const StatisticsMapping diskReadRemoteStatistics(StTimeDiskReadIO, StSizeDiskRead, StNumDiskReads, StKindNone);
@@ -1625,10 +1629,11 @@ CRuntimeStatisticCollection::~CRuntimeStatisticCollection()
     delete nested;
 }
 
-void CRuntimeStatisticCollection::ensureNested()
+CNestedRuntimeStatisticMap & CRuntimeStatisticCollection::ensureNested()
 {
     if (!nested)
         nested = new CNestedRuntimeStatisticMap;
+    return *nested;
 }
 
 void CRuntimeStatisticCollection::merge(const CRuntimeStatisticCollection & other)
@@ -1638,12 +1643,38 @@ void CRuntimeStatisticCollection::merge(const CRuntimeStatisticCollection & othe
         StatisticKind kind = other.getKind(i);
         unsigned __int64 value = other.getStatisticValue(kind);
         if (value)
-            mergeStatistic(kind, other.getStatisticValue(kind));
+            mergeStatistic(kind, value);
     }
     if (other.nested)
     {
-        ensureNested();
-        nested->merge(*other.nested);
+        ensureNested().merge(*other.nested);
+    }
+}
+
+void CRuntimeStatisticCollection::updateDelta(CRuntimeStatisticCollection & target, const CRuntimeStatisticCollection & source)
+{
+    ForEachItemIn(i, source)
+    {
+        StatisticKind kind = source.getKind(i);
+        unsigned __int64 sourceValue = source.getStatisticValue(kind);
+        if (queryMergeMode(kind) == StatsMergeSum)
+        {
+            unsigned __int64 prevValue = getStatisticValue(kind);
+            if (sourceValue != prevValue)
+            {
+                target.mergeStatistic(kind, sourceValue - prevValue);
+                setStatistic(kind, sourceValue);
+            }
+        }
+        else
+        {
+            if (sourceValue)
+                target.mergeStatistic(kind, sourceValue);
+        }
+    }
+    if (source.nested)
+    {
+        ensureNested().updateDelta(target.ensureNested(), *source.nested);
     }
 }
 
@@ -1652,10 +1683,23 @@ void CRuntimeStatisticCollection::mergeStatistic(StatisticKind kind, unsigned __
     queryStatistic(kind).merge(value, queryMergeMode(kind));
 }
 
+void CRuntimeStatisticCollection::reset()
+{
+    unsigned num = mapping.numStatistics();
+    for (unsigned i = 0; i <= num; i++)
+        values[i].clear();
+}
+
+void CRuntimeStatisticCollection::reset(const StatisticsMapping & toClear)
+{
+    unsigned num = toClear.numStatistics();
+    for (unsigned i = 0; i < num; i++)
+        queryStatistic(toClear.getKind(i)).clear();
+}
+
 CRuntimeStatisticCollection & CRuntimeStatisticCollection::registerNested(const StatsScopeId & scope, const StatisticsMapping & mapping)
 {
-    ensureNested();
-    return nested->addNested(scope, mapping).queryStats();
+    return ensureNested().addNested(scope, mapping).queryStats();
 }
 
 void CRuntimeStatisticCollection::rollupStatistics(unsigned numTargets, IContextLogger * const * targets) const
@@ -1751,8 +1795,7 @@ void CRuntimeStatisticCollection::deserialize(MemoryBuffer& in)
     in.read(hasNested);
     if (hasNested)
     {
-        ensureNested();
-        nested->deserializeMerge(in);
+        ensureNested().deserializeMerge(in);
     }
 }
 
@@ -1773,8 +1816,7 @@ void CRuntimeStatisticCollection::deserializeMerge(MemoryBuffer& in)
     in.read(hasNested);
     if (hasNested)
     {
-        ensureNested();
-        nested->deserializeMerge(in);
+        ensureNested().deserializeMerge(in);
     }
 }
 
@@ -1859,10 +1901,11 @@ CRuntimeSummaryStatisticCollection::~CRuntimeSummaryStatisticCollection()
     delete[] derived;
 }
 
-void CRuntimeSummaryStatisticCollection::ensureNested()
+CNestedRuntimeStatisticMap & CRuntimeSummaryStatisticCollection::ensureNested()
 {
     if (!nested)
         nested = new CNestedSummaryRuntimeStatisticMap;
+    return *nested;
 }
 
 void CRuntimeSummaryStatisticCollection::mergeStatistic(StatisticKind kind, unsigned __int64 value)
@@ -1992,6 +2035,11 @@ StringBuffer & CNestedRuntimeStatisticCollection::toXML(StringBuffer &str) const
     return str.append("</Scope>");
 }
 
+void CNestedRuntimeStatisticCollection::updateDelta(CNestedRuntimeStatisticCollection & target, const CNestedRuntimeStatisticCollection & source)
+{
+    stats->updateDelta(*target.stats, *source.stats);
+}
+
 //---------------------------------------------------
 
 CNestedRuntimeStatisticCollection & CNestedRuntimeStatisticMap::addNested(const StatsScopeId & scope, const StatisticsMapping & mapping)
@@ -2048,6 +2096,17 @@ void CNestedRuntimeStatisticMap::merge(const CNestedRuntimeStatisticMap & other)
     }
 }
 
+void CNestedRuntimeStatisticMap::updateDelta(CNestedRuntimeStatisticMap & target, const CNestedRuntimeStatisticMap & source)
+{
+    ForEachItemIn(i, source.map)
+    {
+        CNestedRuntimeStatisticCollection & curSource = source.map.item(i);
+        CNestedRuntimeStatisticCollection & curTarget = target.addNested(curSource.scope, curSource.queryMapping());
+        CNestedRuntimeStatisticCollection & curDelta = addNested(curSource.scope, curSource.queryMapping());
+        curDelta.updateDelta(curTarget, curSource);
+    }
+}
+
 bool CNestedRuntimeStatisticMap::serialize(MemoryBuffer& out) const
 {
     out.appendPacked(map.ordinality());

system/jlib/jstats.h

@@ -271,6 +271,7 @@ protected:
 };
 
 extern const jlib_decl StatisticsMapping allStatistics;
+extern const jlib_decl StatisticsMapping heapStatistics;
 extern const jlib_decl StatisticsMapping diskLocalStatistics;
 extern const jlib_decl StatisticsMapping diskRemoteStatistics;
 extern const jlib_decl StatisticsMapping diskReadRemoteStatistics;
@@ -363,12 +364,8 @@ public:
     {
         return queryStatistic(kind).get();
     }
-    void reset()
-    {
-        unsigned num = mapping.numStatistics();
-        for (unsigned i = 0; i <= num; i++)
-            values[i].clear();
-    }
+    void reset();
+    void reset(const StatisticsMapping & toClear);
 
     CRuntimeStatisticCollection & registerNested(const StatsScopeId & scope, const StatisticsMapping & mapping);
 
@@ -378,9 +375,11 @@ public:
     inline unsigned __int64 getValue(unsigned i) const { return values[i].get(); }
 
     void merge(const CRuntimeStatisticCollection & other);
+    void updateDelta(CRuntimeStatisticCollection & target, const CRuntimeStatisticCollection & source);
     void rollupStatistics(IContextLogger * target) { rollupStatistics(1, &target); }
     void rollupStatistics(unsigned num, IContextLogger * const * targets) const;
 
+
     virtual void recordStatistics(IStatisticGatherer & target) const;
     void getNodeProgressInfo(IPropertyTree &node) const;
 
@@ -395,7 +394,7 @@ public:
 
 
 protected:
-    virtual void ensureNested();
+    virtual CNestedRuntimeStatisticMap & ensureNested();
     void reportIgnoredStats() const;
     void mergeStatistic(StatisticKind kind, unsigned __int64 value, StatsMergeAction mergeAction)
     {
@@ -437,7 +436,7 @@ protected:
     };
 
 protected:
-    virtual void ensureNested() override;
+    virtual CNestedRuntimeStatisticMap & ensureNested() override;
 
 protected:
     DerivedStats * derived;
@@ -465,6 +464,7 @@ public:
     void recordStatistics(IStatisticGatherer & target) const;
     StringBuffer & toStr(StringBuffer &str) const;
     StringBuffer & toXML(StringBuffer &str) const;
+    void updateDelta(CNestedRuntimeStatisticCollection & target, const CNestedRuntimeStatisticCollection & source);
 
 public:
     StatsScopeId scope;
@@ -485,6 +485,8 @@ public:
     void recordStatistics(IStatisticGatherer & target) const;
     StringBuffer & toStr(StringBuffer &str) const;
     StringBuffer & toXML(StringBuffer &str) const;
+    void updateDelta(CNestedRuntimeStatisticMap & target, const CNestedRuntimeStatisticMap & source);
+
 
 protected:
     virtual CRuntimeStatisticCollection * createStats(const StatisticsMapping & mapping);