HPCC-30949 Ensure that duplicate statistics are not added to a CStatsCollection

roxie/ccd/ccdserver.cpp

@@ -390,6 +390,60 @@ class IndirectAgentContext : implements IRoxieAgentContext, public CInterface
 
 //=================================================================================
 
+class MergingStatsGatherer : implements CUnsharedInterfaceOf<IStatisticGatherer>
+{
+public:
+    MergingStatsGatherer(IStatisticGatherer * _gatherer) : gatherer(_gatherer)
+    {
+    }
+
+    virtual void beginScope(const StatsScopeId & id) override
+    {
+        gatherer->beginScope(id);
+    }
+    virtual void beginSubGraphScope(unsigned id) override
+    {
+        gatherer->beginSubGraphScope(id);
+    }
+    virtual void beginActivityScope(unsigned id) override
+    {
+        gatherer->beginActivityScope(id);
+    }
+    virtual void beginEdgeScope(unsigned id, unsigned oid) override
+    {
+        gatherer->beginEdgeScope(id, oid);
+    }
+    virtual void beginChildGraphScope(unsigned id) override
+    {
+        gatherer->beginChildGraphScope(id);
+    }
+    virtual void beginChannelScope(unsigned id) override
+    {
+        gatherer->beginChannelScope(id);
+    }
+    virtual void endScope() override
+    {
+        gatherer->endScope();
+    }
+    virtual void addStatistic(StatisticKind kind, unsigned __int64 value) override
+    {
+        //Always merge rather than add (otherwise it may create duplicate entries)
+        gatherer->updateStatistic(kind, value, queryMergeMode(kind));
+    }
+    virtual void updateStatistic(StatisticKind kind, unsigned __int64 value, StatsMergeAction mergeAction) override
+    {
+        gatherer->updateStatistic(kind, value, mergeAction);
+    }
+    virtual IStatisticCollection * getResult() override
+    {
+        return gatherer->getResult();
+    }
+
+private:
+    IStatisticGatherer * gatherer;
+};
+
+//=================================================================================
 #define RESULT_FLUSH_THRESHOLD 10000u
 
 #ifdef _DEBUG
@@ -16532,8 +16586,9 @@ class CRoxieServerParallelGraphLoopActivity : public CRoxieServerGraphLoopActivi
         resultStream = NULL;
         resultJunction.clear();
 
+        MergingStatsGatherer mergeStats(childStats);
         ForEachItemIn(i, iterationGraphs)
-            iterationGraphs.item(i).gatherStatistics(childStats);
+            iterationGraphs.item(i).gatherStatistics(childStats ? &mergeStats : nullptr);
 
         outputs.kill();
         iterationGraphs.kill(); // must be done after all activities killed
@@ -28506,9 +28561,11 @@ class CProxyActivityGraph : implements IActivityGraph, implements IThorChildGrap
     virtual const char *queryName() const override { throwUnexpected(); }
     virtual void gatherStatistics(IStatisticGatherer * statsBuilder) const override
     {
+        //Merge the stats from multiple graph instances.
+        MergingStatsGatherer mergeStatsBuilder(statsBuilder);
         CriticalBlock b(graphCrit);
         ForEachItemIn(i, stack)
-            stack.item(i).gatherStatistics(statsBuilder);
+            stack.item(i).gatherStatistics(statsBuilder ? &mergeStatsBuilder : nullptr);
     }
     virtual IEclGraphResults * evaluate(unsigned parentExtractSize, const byte * parentExtract) override
     {

system/jlib/jstats.cpp

@@ -1917,8 +1917,15 @@ class CStatisticCollection : public CInterfaceOf<IStatisticCollection>
     }
 
 //other public interface functions
+    // addStatistic() should only be used if it is known that the kind is not already there
     void addStatistic(StatisticKind kind, unsigned __int64 value)
     {
+#ifdef _DEBUG
+        // In debug builds check that a value for this kind has not already been added.
+        // We do not want the O(N) overhead in release mode, especially as N is beginning to get larger
+        unsigned __int64 debugTest;
+        assertex(getStatistic(kind,debugTest)==false);
+#endif
         Statistic s(kind, value);
         stats.append(s);
     }

system/jlib/jstats.h

@@ -156,7 +156,7 @@ interface IStatisticGatherer : public IInterface
     virtual void beginChildGraphScope(unsigned id) = 0;
     virtual void beginChannelScope(unsigned id) = 0;
     virtual void endScope() = 0;
-    virtual void addStatistic(StatisticKind kind, unsigned __int64 value) = 0;
+    virtual void addStatistic(StatisticKind kind, unsigned __int64 value) = 0; // use updateStatistic() if kind could already be defined for the active scope
     virtual void updateStatistic(StatisticKind kind, unsigned __int64 value, StatsMergeAction mergeAction) = 0;
     virtual IStatisticCollection * getResult() = 0;
 };