HPCC-26365 Rework roxie stats to fully use new stats code

Rework Roxie to use the jstats system properly, and to add the ability to add
and accumulate stats in temporary workunits.

Remove any unused code that implemented Roxie stats in more ad-hoc and
inefficient ways

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

common/workunit/referencedfilelist.cpp

@@ -633,7 +633,7 @@ bool ReferencedFileList::addFilesFromQuery(IConstWorkUnit *cw, const IHpccPackag
     Owned<IConstWUGraphIterator> graphs = &cw->getGraphs(GraphTypeActivities);
     ForEach(*graphs)
     {
-        Owned <IPropertyTree> xgmml = graphs->query().getXGMMLTree(false);
+        Owned <IPropertyTree> xgmml = graphs->query().getXGMMLTree(false, false);
         Owned<IPropertyTreeIterator> iter = xgmml->getElements("//node[att/@name='_*ileName']");
         ForEach(*iter)
         {

common/workunit/workunit.cpp

@@ -182,8 +182,8 @@ void doDescheduleWorkkunit(char const * wuid)
  * Graph progress support
  */
 
-CWuGraphStats::CWuGraphStats(IPropertyTree *_progress, StatisticCreatorType _creatorType, const char * _creator, unsigned wfid, const char * _rootScope, unsigned _id)
-    : progress(_progress), creatorType(_creatorType), creator(_creator), id(_id)
+CWuGraphStats::CWuGraphStats(StatisticCreatorType _creatorType, const char * _creator, unsigned wfid, const char * _rootScope, unsigned _id, bool _merge)
+    : creatorType(_creatorType), creator(_creator), id(_id), merge(_merge)
 {
     StatsScopeId graphScopeId;
     verifyex(graphScopeId.setScopeText(_rootScope));
@@ -195,8 +195,25 @@ CWuGraphStats::CWuGraphStats(IPropertyTree *_progress, StatisticCreatorType _cre
 
 void CWuGraphStats::beforeDispose()
 {
-    Owned<IStatisticCollection> stats = collector->getResult();
+    collector->endScope();
+    StringBuffer tag;
+    tag.append("sg").append(id);
 
+    IPropertyTree &progress = queryProgressTree();
+    if (merge && progress.hasProp(tag))
+    {
+        VStringBuffer statsPath("%s/Stats", tag.str());
+        MemoryBuffer compressed;
+        progress.getPropBin(statsPath, compressed);
+        if (compressed.length())
+        {
+            MemoryBuffer serialized;
+            decompressToBuffer(serialized, compressed);
+            Owned<IStatisticCollection> prevCollection = createStatisticCollection(serialized);
+            prevCollection->mergeInto(*collector);
+        }
+    }
+    Owned<IStatisticCollection> stats = collector->getResult();
     MemoryBuffer compressed;
     {
         MemoryBuffer serialized;
@@ -208,18 +225,15 @@ void CWuGraphStats::beforeDispose()
     unsigned maxActivity = 0;
     stats->getMinMaxActivity(minActivity, maxActivity);
 
-    StringBuffer tag;
-    tag.append("sg").append(id);
-
     //Replace the particular subgraph statistics added by this creator
-    IPropertyTree * subgraph = progress->setPropTree(tag);
+    IPropertyTree * subgraph = progress.setPropTree(tag);
     subgraph->setProp("@c", queryCreatorTypeName(creatorType));
     subgraph->setProp("@creator", creator);
     subgraph->setPropInt("@minActivity", minActivity);
     subgraph->setPropInt("@maxActivity", maxActivity);
     subgraph->setPropBin("Stats", compressed.length(), compressed.toByteArray());
-    if (!progress->getPropBool("@stats", false))
-        progress->setPropBool("@stats", true);
+    if (!progress.getPropBool("@stats", false))
+        progress.setPropBool("@stats", true);
 }
 
 IStatisticGatherer & CWuGraphStats::queryStatsBuilder()
@@ -237,10 +251,10 @@ public:
             progress.setown(createPTree());
         formatVersion = progress->getPropInt("@format", PROGRESS_FORMAT_V);
     }
-    virtual IPropertyTree * getProgressTree()
+    virtual IPropertyTree * getProgressTree(bool doFormat) override
     {
         if (progress->getPropBool("@stats"))
-            return createProcessTreeFromStats(); // Should we cache that?
+            return createProcessTreeFromStats(doFormat); // Should we cache that?
         return LINK(progress);
     }
     virtual unsigned queryFormatVersion()
@@ -253,7 +267,7 @@ protected:
     {
         formatVersion = PROGRESS_FORMAT_V;
     }
-    static void expandStats(IPropertyTree * target, IStatisticCollection & collection)
+    static void expandStats(IPropertyTree * target, IStatisticCollection & collection, bool doFormat)
     {
         StringBuffer formattedValue;
         unsigned numStats = collection.getNumStatistics();
@@ -262,13 +276,20 @@ protected:
             StatisticKind kind;
             unsigned __int64 value;
             collection.getStatistic(kind, value, i);
-            formatStatistic(formattedValue.clear(), value, kind);
-            target->setProp(queryTreeTag(kind), formattedValue);
+            if (doFormat)
+            {
+                formatStatistic(formattedValue.clear(), value, kind);
+                target->setProp(queryTreeTag(kind), formattedValue);
+            }
+            else
+            {
+                target->setPropInt64(queryTreeTag(kind), value);
+            }
         }
     }
-    void expandProcessTreeFromStats(IPropertyTree * rootTarget, IPropertyTree * target, IStatisticCollection * collection)
+    void expandProcessTreeFromStats(IPropertyTree * rootTarget, IPropertyTree * target, IStatisticCollection * collection, bool doFormat)
     {
-        expandStats(target, *collection);
+        expandStats(target, *collection, doFormat);
 
         StringBuffer scopeName;
         Owned<IStatisticCollectionIterator> activityIter = &collection->getScopes(NULL, false);
@@ -299,9 +320,11 @@ protected:
             case SSTworkflow:
             case SSTgraph:
                 // SSTworkflow and SSTgraph may be safely ignored.  They are not required to produce the statistics.
-                expandProcessTreeFromStats(rootTarget, target, &cur);
+                expandProcessTreeFromStats(rootTarget, target, &cur, doFormat);
                 continue;
             case SSTfunction:
+            case SSTchannel:
+            case SSTglobal:
                 //MORE:Should function scopes be included in the graph scope somehow, and if so how?
                 continue;
             default:
@@ -310,11 +333,11 @@ protected:
 
             IPropertyTree * next = curTarget->addPropTree(tag);
             next->setProp("@id", id);
-            expandProcessTreeFromStats(rootTarget, next, &cur);
+            expandProcessTreeFromStats(rootTarget, next, &cur, doFormat);
         }
     }
 
-    IPropertyTree * createProcessTreeFromStats()
+    IPropertyTree * createProcessTreeFromStats(bool doFormat)
     {
         MemoryBuffer compressed;
         MemoryBuffer serialized;
@@ -330,7 +353,7 @@ protected:
                 decompressToBuffer(serialized.clear(), compressed);
                 Owned<IStatisticCollection> collection = createStatisticCollection(serialized);
 
-                expandProcessTreeFromStats(progressTree, progressTree, collection);
+                expandProcessTreeFromStats(progressTree, progressTree, collection, doFormat);
             }
         }
         return progressTree.getClear();
@@ -1608,7 +1631,7 @@ private:
             {
                 //Walk dependencies - should possibly have a different SST e.g., SSTdependency since they do not
                 //share many characteristics with edges - e.g. no flowing records => few/no stats.
-                curGraph.setown(graphIter->query().getXGMMLTree(false));
+                curGraph.setown(graphIter->query().getXGMMLTree(false, false));
                 Owned<IPropertyTreeIterator> treeIter = curGraph->getElements("edge");
                 if (treeIter && treeIter->first())
                 {
@@ -3638,15 +3661,42 @@ extern IConstWorkUnitInfo *createConstWorkUnitInfo(IPropertyTree &p)
     return new CLightweightWorkunitInfo(p);
 }
 
+class CDaliWorkUnit;
 class CDaliWuGraphStats : public CWuGraphStats
 {
 public:
-    CDaliWuGraphStats(IRemoteConnection *_conn, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id)
-        : CWuGraphStats(LINK(_conn->queryRoot()), _creatorType, _creator, _wfid, _rootScope, _id), conn(_conn)
+    CDaliWuGraphStats(const CDaliWorkUnit* _owner, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id, bool _merge)
+        : CWuGraphStats(_creatorType, _creator, _wfid, _rootScope, _id, _merge), owner(_owner), graphName(_rootScope), wfid(_wfid)
     {
     }
 protected:
+    virtual IPropertyTree &queryProgressTree() override;
+    const CDaliWorkUnit *owner;
     Owned<IRemoteConnection> conn;
+    StringAttr graphName;
+    unsigned wfid;
+};
+
+class CLocalWuGraphStats : public CWuGraphStats
+{
+public:
+    CLocalWuGraphStats(IPropertyTree *_p, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id, bool _merge)
+        : CWuGraphStats(_creatorType, _creator, _wfid, _rootScope, _id, _merge), graphName(_rootScope), p(_p)
+    {
+    }
+protected:
+    virtual IPropertyTree &queryProgressTree() override
+    {
+        IPropertyTree *progress = p->queryPropTree("GraphProgress");
+        if (!progress)
+            progress = p->addPropTree("GraphProgress");
+        IPropertyTree *graph = progress->queryPropTree(graphName);
+        if (!graph)
+            graph = progress->addPropTree(graphName);
+        return *graph;
+    }
+    StringAttr graphName;
+    Owned<IPropertyTree> p;
 };
 
 CWorkUnitWatcher::CWorkUnitWatcher(IWorkUnitSubscriber *_subscriber, WUSubscribeOptions flags, const char *wuid) : subscriber(_subscriber)
@@ -3751,6 +3801,7 @@ bool CPersistedWorkUnit::aborting() const
 
 class CDaliWorkUnit : public CPersistedWorkUnit
 {
+    friend class CDaliWuGraphStats;
 public:
     CDaliWorkUnit(IRemoteConnection *_conn, ISecManager *secmgr, ISecUser *secuser)
         : connection(_conn), CPersistedWorkUnit(secmgr, secuser)
@@ -3934,9 +3985,9 @@ public:
             }
         }
     }
-    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph) const override
+    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const override
     {
-        return new CDaliWuGraphStats(getWritableProgressConnection(graphName, _wfid), creatorType, creator, _wfid, graphName, subgraph);
+        return new CDaliWuGraphStats(this, creatorType, creator, _wfid, graphName, subgraph, merge);
     }
     virtual void import(IPropertyTree *wuTree, IPropertyTree *graphProgressTree)
     {
@@ -4031,7 +4082,7 @@ public:
         }
     }
 
-    virtual IConstWorkUnit * unlock()
+    virtual IConstWorkUnit * unlock() override
     {
         c->unlockRemote();
         return c.getClear();
@@ -4246,8 +4297,8 @@ public:
             { c->setGraphState(graphName, wfid, state); }
     virtual void setNodeState(const char *graphName, WUGraphIDType nodeId, WUGraphState state) const
             { c->setNodeState(graphName, nodeId, state); }
-    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph) const override
-            { return c->updateStats(graphName, creatorType, creator, _wfid, subgraph); }
+    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const override
+            { return c->updateStats(graphName, creatorType, creator, _wfid, subgraph, merge); }
     virtual void clearGraphProgress() const
             { c->clearGraphProgress(); }
     virtual IStringVal & getAbortBy(IStringVal & str) const
@@ -4440,6 +4491,12 @@ public:
             { c->setResultDataset(name, sequence, len, val, numRows, extend); }
 };
 
+IPropertyTree &CDaliWuGraphStats::queryProgressTree()
+{
+    conn.setown(owner->getWritableProgressConnection(graphName, wfid));
+    return *conn->queryRoot();
+}
+
 class CLocalWUAssociated : implements IConstWUAssociatedFile, public CInterface
 {
     Owned<IPropertyTree> p;
@@ -7009,6 +7066,12 @@ IWorkUnit& CLocalWorkUnit::lock()
     return lockRemote(true);
 }
 
+IConstWorkUnit *CLocalWorkUnit::unlock()
+{
+    unlockRemote();
+    return this;
+}
+
 const char *CLocalWorkUnit::queryWuid() const
 {
     CriticalBlock block(crit);
@@ -9468,7 +9531,7 @@ IPropertyTree *CLocalWorkUnit::getUnpackedTree(bool includeProgress) const
     ForEach(*graphIter)
     {
         IConstWUGraph &graph  = graphIter->query();
-        Owned<IPropertyTree> graphTree = graph.getXGMMLTree(includeProgress);
+        Owned<IPropertyTree> graphTree = graph.getXGMMLTree(includeProgress, false);
         SCMStringBuffer gName;
         graph.getName(gName);
         StringBuffer xpath("Graphs/Graph[@name=\"");
@@ -9554,7 +9617,7 @@ IStringVal& CLocalWUGraph::getLabel(IStringVal &str) const
     }
     else
     {
-        Owned<IPropertyTree> xgmml = getXGMMLTree(false);
+        Owned<IPropertyTree> xgmml = getXGMMLTree(false, false);
         str.set(xgmml->queryProp("@label"));
         return str;
     }
@@ -9566,9 +9629,9 @@ WUGraphState CLocalWUGraph::getState() const
 }
 
 
-IStringVal& CLocalWUGraph::getXGMML(IStringVal &str, bool mergeProgress) const
+IStringVal& CLocalWUGraph::getXGMML(IStringVal &str, bool mergeProgress, bool doFormatStats) const
 {
-    Owned<IPropertyTree> xgmml = getXGMMLTree(mergeProgress);
+    Owned<IPropertyTree> xgmml = getXGMMLTree(mergeProgress, doFormatStats);
     if (xgmml)
     {
         StringBuffer x;
@@ -9814,17 +9877,16 @@ void CLocalWorkUnit::createGraph(const char * name, const char *label, WUGraphTy
     graphs.append(*q);
 }
 
-IConstWUGraphProgress *CLocalWorkUnit::getGraphProgress(const char *name) const
+IConstWUGraphProgress *CLocalWorkUnit::getGraphProgress(const char *graphName) const
 {
-/*    Owned<IRemoteConnection> conn = getProgressConnection();
-    if (conn)
+    IPTree *graphProgress = p->queryPropTree("GraphProgress");
+    if (graphProgress)
     {
-        IPTree *progress = conn->queryRoot()->queryPropTree(graphName);
+        IPTree *progress = graphProgress->queryPropTree(graphName);
         if (progress)
-            return new CConstGraphProgress(p->queryName(), graphName, progress);
+            return new CConstGraphProgress(queryWuid(), graphName, progress);
     }
-    */
-    return NULL;
+    return nullptr;
 }
 WUGraphState CLocalWorkUnit::queryGraphState(const char *graphName) const
 {
@@ -9842,9 +9904,9 @@ void CLocalWorkUnit::setNodeState(const char *graphName, WUGraphIDType nodeId, W
 {
     throwUnexpected();   // Should only be used for persisted workunits
 }
-IWUGraphStats *CLocalWorkUnit::updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph) const
+IWUGraphStats *CLocalWorkUnit::updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const
 {
-    return new CWuGraphStats(LINK(p), creatorType, creator, _wfid, graphName, subgraph);
+    return new CLocalWuGraphStats(LINK(p), creatorType, creator, _wfid, graphName, subgraph, merge);
 }
 
 void CLocalWUGraph::setName(const char *str)
@@ -9961,7 +10023,7 @@ IPropertyTree * CLocalWUGraph::getXGMMLTreeRaw() const
     return p->getPropTree("xgmml");
 }
 
-IPropertyTree * CLocalWUGraph::getXGMMLTree(bool doMergeProgress) const
+IPropertyTree * CLocalWUGraph::getXGMMLTree(bool doMergeProgress, bool doFormatStats) const
 {
     if (!graph)
     {
@@ -9985,7 +10047,7 @@ IPropertyTree * CLocalWUGraph::getXGMMLTree(bool doMergeProgress) const
         {
             //MORE: Eventually this should directly access the new stats structure
             unsigned progressV = progress->queryFormatVersion();
-            Owned<IPropertyTree> progressTree = progress->getProgressTree();
+            Owned<IPropertyTree> progressTree = progress->getProgressTree(doFormatStats);
             Owned<IPropertyTreeIterator> nodeIterator = copy->getElements("node");
             ForEach (*nodeIterator)
                 mergeProgress(nodeIterator->query(), *progressTree, progressV);
@@ -11692,6 +11754,14 @@ extern WORKUNIT_API ILocalWorkUnit * createLocalWorkUnit(const char *xml)
     return ret;
 }
 
+extern WORKUNIT_API ILocalWorkUnit * createLocalWorkUnitFromPTree(IPropertyTree *ptree)
+{
+    Owned<CLocalWorkUnit> cw = new CLocalWorkUnit((ISecManager *) NULL, NULL);
+    cw->loadPTree(ptree);
+    ILocalWorkUnit* ret = QUERYINTERFACE(&cw->lockRemote(false), ILocalWorkUnit);
+    return ret;
+}
+
 void exportWorkUnitToXMLWithHiddenPasswords(IPropertyTree *p, IIOStream &out, unsigned extraXmlFlags)
 {
     const char *name = p->queryName();
@@ -13511,6 +13581,11 @@ public:
         StatsScopeId scopeId(SSTchildgraph, id);
         beginScope(scopeId);
     }
+    virtual void beginChannelScope(unsigned id)
+    {
+        StatsScopeId scopeId(SSTchannel, id);
+        beginScope(scopeId);
+    }
     virtual void endScope()
     {
         scope.setLength(prevLenStack.popGet());

common/workunit/workunit.hpp

@@ -206,8 +206,8 @@ interface IConstWUGraphMeta : extends IInterface
 
 interface IConstWUGraph : extends IConstWUGraphMeta
 {
-    virtual IStringVal & getXGMML(IStringVal & ret, bool mergeProgress) const = 0;
-    virtual IPropertyTree * getXGMMLTree(bool mergeProgress) const = 0;
+    virtual IStringVal & getXGMML(IStringVal & ret, bool mergeProgress, bool doFormatStats) const = 0;
+    virtual IPropertyTree * getXGMMLTree(bool mergeProgress, bool doFormatStats) const = 0;
     virtual IPropertyTree * getXGMMLTreeRaw() const = 0;
 };
 
@@ -768,7 +768,7 @@ interface IWUGraphStats;
 interface IPropertyTree;
 interface IConstWUGraphProgress : extends IInterface
 {
-    virtual IPropertyTree * getProgressTree() = 0;
+    virtual IPropertyTree * getProgressTree(bool doFormat) = 0;
     virtual unsigned queryFormatVersion() = 0;
 };
 
@@ -1296,7 +1296,7 @@ interface IConstWorkUnit : extends IConstWorkUnitInfo
     virtual WUGraphState queryNodeState(const char *graphName, WUGraphIDType nodeId) const = 0;
     virtual void setGraphState(const char *graphName, unsigned wfid, WUGraphState state) const = 0;
     virtual void setNodeState(const char *graphName, WUGraphIDType nodeId, WUGraphState state) const = 0;
-    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph) const = 0;
+    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const = 0;
     virtual void clearGraphProgress() const = 0;
     virtual IStringVal & getAbortBy(IStringVal & str) const = 0;
     virtual unsigned __int64 getAbortTimeStamp() const = 0;
@@ -1388,6 +1388,7 @@ interface IWorkUnit : extends IConstWorkUnit
     virtual void setResultDecimal(const char *name, unsigned sequence, int len, int precision, bool isSigned, const void *val) = 0;
     virtual void setResultDataset(const char * name, unsigned sequence, size32_t len, const void *val, unsigned numRows, bool extend) = 0;
     virtual void import(IPropertyTree *wuTree, IPropertyTree *graphProgressTree = nullptr) = 0;
+    virtual IConstWorkUnit * unlock() = 0;
 };
 
 
@@ -1419,7 +1420,6 @@ interface ILocalWorkUnit : extends IWorkUnit
 {
     virtual void serialize(MemoryBuffer & tgt) = 0;
     virtual void deserialize(MemoryBuffer & src) = 0;
-    virtual IConstWorkUnit * unlock() = 0;
 };
 
 
@@ -1597,6 +1597,7 @@ extern WORKUNIT_API void setWorkUnitFactory(IWorkUnitFactory *_factory);
 extern WORKUNIT_API IWorkUnitFactory * getWorkUnitFactory();
 extern WORKUNIT_API IWorkUnitFactory * getWorkUnitFactory(ISecManager *secmgr, ISecUser *secuser);
 extern WORKUNIT_API ILocalWorkUnit* createLocalWorkUnit(const char *XML);
+extern WORKUNIT_API ILocalWorkUnit * createLocalWorkUnitFromPTree(IPropertyTree *ptree);  // takes ownership of tree
 extern WORKUNIT_API IConstWorkUnitInfo *createConstWorkUnitInfo(IPropertyTree &p);
 extern WORKUNIT_API StringBuffer &exportWorkUnitToXML(const IConstWorkUnit *wu, StringBuffer &str, bool unpack, bool includeProgress, bool hidePasswords);
 extern WORKUNIT_API void exportWorkUnitToXMLFile(const IConstWorkUnit *wu, const char * filename, unsigned extraXmlFlags, bool unpack, bool includeProgress, bool hidePasswords, bool splitStats);

common/workunit/workunit.ipp

@@ -231,7 +231,7 @@ public:
     virtual void setGraphState(const char *graphName, unsigned wfid, WUGraphState state) const;
     virtual void setNodeState(const char *graphName, WUGraphIDType nodeId, WUGraphState state) const;
     virtual WUGraphState queryNodeState(const char *graphName, WUGraphIDType nodeId) const;
-    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph) const override;
+    virtual IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned _wfid, unsigned subgraph, bool merge) const override;
     void clearGraphProgress() const;
     virtual void import(IPropertyTree *wuTree, IPropertyTree *graphProgressTree) {}; //No GraphProgressTree in CLocalWorkUnit.
 
@@ -280,6 +280,7 @@ public:
     virtual bool isProtected() const;
     virtual bool isPausing() const;
     virtual IWorkUnit& lock();
+    virtual IConstWorkUnit *unlock();
     virtual void requestAbort();
     virtual unsigned calculateHash(unsigned prevHash);
     virtual void copyWorkUnit(IConstWorkUnit *cached, bool copyStats, bool all);
@@ -637,14 +638,14 @@ public:
     IMPLEMENT_IINTERFACE;
     CLocalWUGraph(const CLocalWorkUnit &owner, IPropertyTree *p);
 
-    virtual IStringVal & getXGMML(IStringVal & ret, bool mergeProgress) const override;
+    virtual IStringVal & getXGMML(IStringVal & ret, bool mergeProgress, bool doFormatStats) const override;
     virtual IStringVal & getName(IStringVal & ret) const override;
     virtual IStringVal & getLabel(IStringVal & ret) const override;
     virtual IStringVal & getTypeName(IStringVal & ret) const override;
     virtual WUGraphType getType() const override;
     virtual WUGraphState getState() const override;
     virtual unsigned getWfid() const override;
-    virtual IPropertyTree * getXGMMLTree(bool mergeProgress) const override;
+    virtual IPropertyTree * getXGMMLTree(bool mergeProgress, bool doFormatStats) const override;
     virtual IPropertyTree * getXGMMLTreeRaw() const override;
 
     void setName(const char *str);
@@ -658,15 +659,16 @@ public:
 class WORKUNIT_API CWuGraphStats : public CInterfaceOf<IWUGraphStats>
 {
 public:
-    CWuGraphStats(IPropertyTree *_progress, StatisticCreatorType _creatorType, const char * _creator, unsigned wfid, const char * _rootScope, unsigned _id);
+    CWuGraphStats(StatisticCreatorType _creatorType, const char * _creator, unsigned wfid, const char * _rootScope, unsigned _id, bool _merge);
     virtual void beforeDispose();
     virtual IStatisticGatherer & queryStatsBuilder();
 protected:
-    Owned<IPropertyTree> progress;
+    virtual IPropertyTree &queryProgressTree() = 0;
     Owned<IStatisticGatherer> collector;
     StringAttr creator;
     StatisticCreatorType creatorType;
     unsigned id;
+    bool merge;
 };
 
 class WORKUNIT_API CWorkUnitWatcher : public CInterface, implements IWorkUnitWatcher, implements ISDSSubscription

dali/daliadmin/daadmin.cpp

@@ -2577,7 +2577,7 @@ void dumpProgress(const char *wuid, const char * graph)
     Owned<IConstWUGraphProgress> progress = workunit->getGraphProgress(graph);
     if (!progress)
         return;
-    Owned<IPropertyTree> tree = progress->getProgressTree();
+    Owned<IPropertyTree> tree = progress->getProgressTree(true);
     saveXML("stdout:", tree);
 }
 
@@ -3290,4 +3290,4 @@ void removeOrphanedGlobalVariables(bool dryrun, bool reconstruct)
     }
 }
 
-} // namespace daadmin
+} // namespace daadmin

ecl/agentexec/agentexec.cpp

@@ -67,7 +67,6 @@ CEclAgentExecutionServer::CEclAgentExecutionServer(IPropertyTree *_config) : con
 
     agentName = config->queryProp("@name");
     assertex(agentName);
-    setStatisticsComponentName(SCThthor, agentName, true);
 
     daliServers = config->queryProp("@daliServers");
     assertex(daliServers);
@@ -75,6 +74,12 @@ CEclAgentExecutionServer::CEclAgentExecutionServer(IPropertyTree *_config) : con
     apptype = config->queryProp("@type");
     if (!apptype)
         apptype = "hthor";
+    StatisticCreatorType ctype = SCThthor;
+    if (strieq(apptype, "roxie"))
+        ctype = SCTroxie;
+    else if (strieq(apptype, "thor"))
+        ctype = SCTthor;
+    setStatisticsComponentName(ctype, agentName, true);
 #ifdef _CONTAINERIZED
     unsigned poolSize = config->getPropInt("@maxActive", 100);
     pool.setown(createThreadPool("agentPool", this, NULL, poolSize, INFINITE));

ecl/eclagent/eclagent.cpp

@@ -543,7 +543,7 @@ EclAgent::EclAgent(IConstWorkUnit *wu, const char *_wuid, bool _checkVersion, bo
             Owned<IPropertyTree> graphTree = createPTree("Graph");
             graphTree->addProp("@id", graphName.str());
             Owned<IPropertyTree> xgmmlTree = createPTree("xgmml");
-            Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false);
+            Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false, false);
             xgmmlTree->addPropTree("graph", graphXgmml.getClear());
             graphTree->addPropTree("xgmml", xgmmlTree.getClear());
             destTree->addPropTree("Graph", graphTree.getClear());

ecl/eclagent/eclagent.ipp

@@ -920,83 +920,6 @@ class EclSubGraph : public CInterface, implements ILocalEclGraphResults, public
     friend class EclGraphElement;
 private:
 
-    class LegacyInputProbe : public CInterface, implements IHThorInput, implements IEngineRowStream
-    {
-        IHThorInput  *in;
-        size32_t    maxRowSize;
-        unsigned sourceId;
-        unsigned outputIndex;
-
-        StringAttr edgeId;
-
-    public:
-        IMPLEMENT_IINTERFACE;
-
-        LegacyInputProbe(IHThorInput *_in, unsigned _sourceId, int outputidx)
-            : in(_in), sourceId(_sourceId), outputIndex(outputidx)
-        {
-            StringAttrBuilder edgeIdText(edgeId);
-            edgeIdText.append(_sourceId).append("_").append(outputidx);
-            maxRowSize = 0;
-        }
-
-        IOutputMetaData * queryOutputMeta() const { return in->queryOutputMeta(); }
-
-        void ready()
-        {
-            in->ready();
-        }
-
-        void stop()
-        {
-            in->stop();
-        }
-
-        virtual void resetEOF()
-        {
-            in->resetEOF();
-        }
-
-        IEngineRowStream &queryStream()
-        {
-            return *this;
-        }
-
-        bool isGrouped() { return in->isGrouped(); }
-
-        bool nextGroup(ConstPointerArray & group)
-        {
-            const void * next;
-            while ((next = nextRow()) != NULL)
-                group.append(next);
-            if (group.ordinality())
-                return true;
-            return false;
-        }
-
-        const void *nextRow()
-        {
-            const void *ret = in->nextRow();
-            if (ret)
-            {
-                size32_t size = in->queryOutputMeta()->getRecordSize(ret);
-                if (size > maxRowSize)
-                    maxRowSize = size;
-            }
-            return ret;
-        }
-
-        virtual void updateProgress(IStatisticGatherer &progress) const
-        {
-            {
-                StatsEdgeScope scope(progress, sourceId, outputIndex);
-                progress.addStatistic(StSizeMaxRowSize, maxRowSize);
-            }
-            if (in)
-                in->updateProgress(progress);
-        }
-    };
-
     RedirectedAgentContext subgraphAgentContext;
     class SubGraphCodeContext : public IndirectCodeContext
     {
@@ -1016,11 +939,8 @@ private:
         EclSubGraph * container;
     } subgraphCodeContext;
 
-    friend class LegacyInputProbe;
-    bool probeEnabled;
-
 public:
-    EclSubGraph(IAgentContext & _agent, EclGraph &parent, EclSubGraph * _owner, unsigned subGraphSeqNo, bool enableProbe, CHThorDebugContext * _debugContext, IProbeManager * _probeManager);
+    EclSubGraph(IAgentContext & _agent, EclGraph &parent, EclSubGraph * _owner, unsigned subGraphSeqNo, CHThorDebugContext * _debugContext, IProbeManager * _probeManager);
     IMPLEMENT_IINTERFACE
 
     void createFromXGMML(EclGraph * graph, ILoadedDllEntry * dll, IPropertyTree * xgmml, unsigned & subGraphSeqNo, EclSubGraph * resultsGraph);
@@ -1035,22 +955,6 @@ public:
     void doExecuteChild(const byte * parentExtract);
     IEclLoopGraph * resolveLoopGraph(unsigned id);
 
-    IHThorInput *createLegacyProbe(IHThorInput      *in,
-                             unsigned           sourceId,
-                             unsigned           targetId,
-                             int                outputidx,
-                             IConstWorkUnit     *workunit)
-    {
-        if (probeEnabled)
-        {
-            LegacyInputProbe *probe = new LegacyInputProbe(in, sourceId, outputidx);
-            probes.append(*probe);
-            return probe;
-        }
-        else
-            return in;
-    }
-
 //interface IEclGraphResults
     virtual IHThorGraphResult * queryResult(unsigned id);
     virtual IHThorGraphResult * queryGraphLoopResult(unsigned id);
@@ -1144,7 +1048,7 @@ public:
         aborted = false;
     }
 
-    void createFromXGMML(ILoadedDllEntry * dll, IPropertyTree * xgmml, bool enableProbe);
+    void createFromXGMML(ILoadedDllEntry * dll, IPropertyTree * xgmml);
     void execute(const byte * parentExtract);
     void executeLibrary(const byte * parentExtract, IHThorGraphResults * results);
     IWUGraphStats *updateStats(StatisticCreatorType creatorType, const char * creator, unsigned wfid, unsigned subgraph);

ecl/eclagent/eclgraph.cpp

@@ -499,7 +499,7 @@ void EclGraphElement::createActivity(IAgentContext & agent, EclSubGraph * owner)
             ForEachItemIn(i2, branches)
             {
                 EclGraphElement & input = branches.item(i2);
-                IHThorInput * probe = NULL;
+                IHThorInput * useInput = NULL;
 
                 if (probeManager)
                 {
@@ -512,18 +512,14 @@ void EclGraphElement::createActivity(IAgentContext & agent, EclSubGraph * owner)
                                                         0,//input.id,
                                                         0,//id,
                                                         0);
-                        probe = & dynamic_cast<IHThorInput &> (base->queryInput());
+                        useInput = & dynamic_cast<IHThorInput &> (base->queryInput());
                     }
                 }
                 else
                 {
-                    probe = subgraph->createLegacyProbe(input.queryOutput(branchIndexes.item(i2)),
-                                                    input.id,
-                                                    id,
-                                                    0,
-                                                    agent.queryWorkUnit());
+                    useInput = input.queryOutput(branchIndexes.item(i2));
                 }
-                activity->setInput(i2, probe);
+                activity->setInput(i2, useInput);
             }
             break;
         }
@@ -758,8 +754,8 @@ IHThorException * EclGraphElement::makeWrappedException(IException * e)
 
 //---------------------------------------------------------------------------
 
-EclSubGraph::EclSubGraph(IAgentContext & _agent, EclGraph & _parent, EclSubGraph * _owner, unsigned _seqNo, bool enableProbe, CHThorDebugContext * _debugContext, IProbeManager * _probeManager)
-    : probeEnabled(enableProbe), seqNo(_seqNo), parent(_parent), owner(_owner), debugContext(_debugContext), probeManager(_probeManager), isLoopBody(false)
+EclSubGraph::EclSubGraph(IAgentContext & _agent, EclGraph & _parent, EclSubGraph * _owner, unsigned _seqNo, CHThorDebugContext * _debugContext, IProbeManager * _probeManager)
+    : seqNo(_seqNo), parent(_parent), owner(_owner), debugContext(_debugContext), probeManager(_probeManager), isLoopBody(false)
 {
     executed = false;
     created = false;
@@ -805,7 +801,7 @@ void EclSubGraph::createFromXGMML(EclGraph * graph, ILoadedDllEntry * dll, IProp
             if (probeManager)
                 childProbe.setown(probeManager->startChildGraph(subGraphSeqNo, NULL));
 
-            Owned<EclSubGraph> subgraph = new EclSubGraph(*agent, *graph, this, subGraphSeqNo++, probeEnabled, debugContext, probeManager);
+            Owned<EclSubGraph> subgraph = new EclSubGraph(*agent, *graph, this, subGraphSeqNo++, debugContext, probeManager);
             subgraph->createFromXGMML(graph, dll, &cur, subGraphSeqNo, resultsGraph);
             if (probeManager)
                 probeManager->endChildGraph(childProbe, NULL);
@@ -1170,7 +1166,7 @@ void EclGraph::associateSubGraph(EclSubGraph * subgraph)
     subgraphMap.setValue(subgraph->id, subgraph);
 }
 
-void EclGraph::createFromXGMML(ILoadedDllEntry * dll, IPropertyTree * xgmml, bool enableProbe)
+void EclGraph::createFromXGMML(ILoadedDllEntry * dll, IPropertyTree * xgmml)
 {
     Owned<IPropertyTreeIterator> iter = xgmml->getElements("node");
     unsigned subGraphSeqNo = 0;
@@ -1181,7 +1177,7 @@ void EclGraph::createFromXGMML(ILoadedDllEntry * dll, IPropertyTree * xgmml, boo
         if (probeManager)
             childProbe.setown(probeManager->startChildGraph(subGraphSeqNo, NULL));
 
-        Owned<EclSubGraph> subgraph = new EclSubGraph(*agent, *this, NULL, subGraphSeqNo++, enableProbe, debugContext, probeManager);
+        Owned<EclSubGraph> subgraph = new EclSubGraph(*agent, *this, NULL, subGraphSeqNo++, debugContext, probeManager);
         subgraph->createFromXGMML(this, dll, &iter->query(), subGraphSeqNo, NULL);
         if (probeManager)
             probeManager->endChildGraph(childProbe, NULL);
@@ -1347,7 +1343,7 @@ void EclGraph::updateLibraryProgress()
     {
         EclSubGraph & cur = graphs.item(idx);
         unsigned wfid = cur.parent.queryWfid();
-        Owned<IWUGraphStats> progress = wu->updateStats(queryGraphName(), queryStatisticsComponentType(), queryStatisticsComponentName(), wfid, cur.id);
+        Owned<IWUGraphStats> progress = wu->updateStats(queryGraphName(), queryStatisticsComponentType(), queryStatisticsComponentName(), wfid, cur.id, false);
         cur.updateProgress(progress->queryStatsBuilder());
     }
 }
@@ -1490,7 +1486,7 @@ void GraphResults::setResult(unsigned id, IHThorGraphResult * result)
 
 IWUGraphStats *EclGraph::updateStats(StatisticCreatorType creatorType, const char * creator, unsigned activeWfid, unsigned subgraph)
 {
-    return wu->updateStats (queryGraphName(), creatorType, creator, activeWfid, subgraph);
+    return wu->updateStats (queryGraphName(), creatorType, creator, activeWfid, subgraph, false);
 }
 
 void EclGraph::updateWUStatistic(IWorkUnit *lockedwu, StatisticScopeType scopeType, const char * scope, StatisticKind kind, const char * descr, unsigned __int64 value)
@@ -1538,12 +1534,10 @@ EclGraph * EclAgent::loadGraph(const char * graphName, IConstWorkUnit * wu, ILoa
 {
     Owned<IConstWUGraph> wuGraph = wu->getGraph(graphName);
     assertex(wuGraph);
-    Owned<IPropertyTree> xgmml = wuGraph->getXGMMLTree(false);
-
-    bool probeEnabled = wuRead->getDebugValueBool("_Probe", false);
+    Owned<IPropertyTree> xgmml = wuGraph->getXGMMLTree(false, false);
 
     Owned<EclGraph> eclGraph = new EclGraph(*this, graphName, wu, isLibrary, debugContext, probeManager, wuGraph->getWfid());
-    eclGraph->createFromXGMML(dll, xgmml, probeEnabled);
+    eclGraph->createFromXGMML(dll, xgmml);
     return eclGraph.getClear();
 }
 

esp/services/ws_workunits/ws_workunitsQuerySets.cpp

@@ -185,7 +185,7 @@ bool copyWULogicalFiles(IEspContext &context, IConstWorkUnit &cw, const char *cl
     Owned<IConstWUGraphIterator> graphs = &cw.getGraphs(GraphTypeActivities);
     ForEach(*graphs)
     {
-        Owned <IPropertyTree> xgmml = graphs->query().getXGMMLTree(false);
+        Owned <IPropertyTree> xgmml = graphs->query().getXGMMLTree(false, false);
         Owned<IPropertyTreeIterator> iter = xgmml->getElements(".//node");
         ForEach(*iter)
         {

esp/services/ws_workunits/ws_workunitsService.cpp

@@ -4006,7 +4006,7 @@ bool CWsWorkunitsEx::onWUProcessGraph(IEspContext &context,IEspWUProcessGraphReq
 
         PROGLOG("WUProcessGraph: %s, Graph Name %s", wuid.str(), req.getName());
         StringBuffer xml;
-        Owned<IPropertyTree> xgmml = graph->getXGMMLTree(true); // merge in graph progress information
+        Owned<IPropertyTree> xgmml = graph->getXGMMLTree(true, true); // merge in graph progress information
         toXML(xgmml.get(), xml);
         resp.setTheGraph(xml.str());
     }
@@ -4059,7 +4059,7 @@ void CWsWorkunitsEx::readGraph(IEspContext& context, const char* subGraphId, WUG
             g->setFailed(true);
     }
 
-    Owned<IPropertyTree> xgmml = graph->getXGMMLTree(true);
+    Owned<IPropertyTree> xgmml = graph->getXGMMLTree(true, true);
 
     // New functionality, if a subgraph id is specified and we only want to load the xgmml for that subgraph
     // then we need to conditionally pull a propertytree from the xgmml graph one and use that for the xgmml.

plugins/cassandra/cassandrawu.cpp

@@ -2422,7 +2422,7 @@ public:
                 const char *name=sub.queryName();
                 if (name[0]=='s' && name[1]=='g')
                 {
-                    setGraphProgress(&graph, graphName, atoi(name+2), sub.queryProp("@creator"));
+                    setGraphProgress(&graph, graphName, atoi(name+2), sub.queryProp("@creator"), false);
                 }
                 else if (streq(name, "node"))
                 {
@@ -2430,7 +2430,7 @@ public:
                     if (subid)
                     {
                         if (sub.hasChildren()) // Old format
-                            setGraphProgress(&sub, graphName, subid, sub.queryProp("@creator"));
+                            setGraphProgress(&sub, graphName, subid, sub.queryProp("@creator"), false);
                         if (sub.hasProp("@_state"))
                             setNodeState(graphName, subid, (WUGraphState) sub.getPropInt("@_state"));
                     }
@@ -2745,25 +2745,30 @@ public:
     class CCassandraWuGraphStats : public CWuGraphStats
     {
     public:
-        CCassandraWuGraphStats(const CCassandraWorkUnit *_parent, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id)
-        : CWuGraphStats(createPTree(_rootScope), _creatorType, _creator, _wfid, _rootScope, _id),
-          parent(_parent)
+        CCassandraWuGraphStats(const CCassandraWorkUnit *_parent, StatisticCreatorType _creatorType, const char * _creator, unsigned _wfid, const char * _rootScope, unsigned _id, bool _merge)
+        : CWuGraphStats(_creatorType, _creator, _wfid, _rootScope, _id, _merge),
+          progress(createPTree(_rootScope)), parent(_parent)
         {
         }
-        virtual void beforeDispose()
+        virtual IPropertyTree &queryProgressTree() override
+        {
+            return *progress.get();
+        }
+        virtual void beforeDispose() override
         {
             CWuGraphStats::beforeDispose(); // Sets up progress - should contain a single child tree sqNN where nn==id
-            parent->setGraphProgress(progress, progress->queryName(), id, creator);
+            parent->setGraphProgress(progress, progress->queryName(), id, creator, merge);
         }
 
     protected:
+        Owned<IPropertyTree> progress;
         Linked<const CCassandraWorkUnit> parent;
         StringAttr wuid;
     };
 
-    IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned wfid, unsigned subgraph) const override
+    IWUGraphStats *updateStats(const char *graphName, StatisticCreatorType creatorType, const char * creator, unsigned wfid, unsigned subgraph, bool merge) const override
     {
-        return new CCassandraWuGraphStats(this, creatorType, creator, wfid, graphName, subgraph);
+        return new CCassandraWuGraphStats(this, creatorType, creator, wfid, graphName, subgraph, merge);
     }
 
 
@@ -2858,8 +2863,10 @@ public:
         return p;
     }
 
-    void setGraphProgress(IPropertyTree *progress, const char *gid, unsigned subid, const char *creator) const
+    void setGraphProgress(IPropertyTree *progress, const char *gid, unsigned subid, const char *creator, bool merge) const
     {
+        if (merge)
+            UNIMPLEMENTED;
         const char *wuid=queryWuid();
         CassandraStatement statement(sessionCache->prepareStatement("INSERT INTO wuGraphProgress (partition, wuid, graphID, subgraphID, creator, progress) values (?,?,?,?,?,?);"));
         statement.bindInt32(0, rtlHash32VStr(wuid, 0) % sessionCache->queryPartitions());
@@ -3428,7 +3435,7 @@ public:
                         const char *name=sub.queryName();
                         if (name[0]=='s' && name[1]=='g')
                         {
-                            wu->setGraphProgress(&graph, graphName, atoi(name+2), sub.queryProp("@creator"));
+                            wu->setGraphProgress(&graph, graphName, atoi(name+2), sub.queryProp("@creator"), false);
                         }
                         else if (streq(name, "node"))
                         {
@@ -3436,7 +3443,7 @@ public:
                             if (subid)
                             {
                                 if (sub.hasChildren()) // Old format
-                                    wu->setGraphProgress(&sub, graphName, subid, sub.queryProp("@creator"));
+                                    wu->setGraphProgress(&sub, graphName, subid, sub.queryProp("@creator"), false);
                                 if (sub.hasProp("@_state"))
                                     wu->setNodeState(graphName, subid, (WUGraphState) sub.getPropInt("@_state"));
                             }

roxie/ccd/ccd.hpp

@@ -367,7 +367,6 @@ extern HardwareInfo hdwInfo;
 extern unsigned parallelAggregate;
 extern bool inMemoryKeysEnabled;
 extern unsigned __int64 minFreeDiskSpace;
-extern bool probeAllRows;
 extern bool steppingEnabled;
 extern bool simpleLocalKeyedJoins;
 extern bool enableKeyDiff;
@@ -555,7 +554,7 @@ public:
 extern void putStatsValue(IPropertyTree *node, const char *statName, const char *statType, unsigned __int64 val);
 extern void putStatsValue(StringBuffer &reply, const char *statName, const char *statType, unsigned __int64 val);
 
-extern const StatisticsMapping globalStatistics;
+extern const StatisticsMapping accumulatedStatistics;
 
 class ContextLogger : implements IRoxieContextLogger, public CInterface
 {
@@ -581,7 +580,7 @@ private:
 public:
     IMPLEMENT_IINTERFACE;
 
-    ContextLogger() : stats(globalStatistics, true)
+    ContextLogger() : stats(accumulatedStatistics, true)
     {
         ctxTraceLevel = traceLevel;
         intercept = false;
@@ -691,6 +690,11 @@ public:
         stats.addStatisticAtomic(kind, value);
     }
 
+    virtual void setStatistic(StatisticKind kind, unsigned __int64 value) const
+    {
+        stats.setStatistic(kind, value);
+    }
+
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const
     {
         CriticalBlock block(statsCrit);

roxie/ccd/ccdactivities.cpp

@@ -100,13 +100,16 @@ extern void putStatsValue(StringBuffer &reply, const char *statName, const char
     }
 }
 
-CActivityFactory::CActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode)
+static const StatisticsMapping edgeStatistics({StNumRowsProcessed, StNumStarts, StNumStops, StNumSlaves});
+
+CActivityFactory::CActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode, const StatisticsMapping &_factoryStats)
   : queryFactory(_queryFactory),
     helperFactory(_helperFactory),
     id(_id),
     subgraphId(_subgraphId),
     kind(_kind),
-    mystats(allStatistics)  // We COULD cut down this list but it would complicate the structure, and we do actually track more in the factory than in the activity
+    mystats(_factoryStats),
+    myedgestats(edgeStatistics)
 {
     if (helperFactory)
     {
@@ -142,8 +145,8 @@ class CAgentActivityFactory : public CActivityFactory, implements IAgentActivity
 public:
     IMPLEMENT_IINTERFACE
 
-    CAgentActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory) 
-        : CActivityFactory(_graphNode.getPropInt("@id", 0), _subgraphId, _queryFactory, _helperFactory, getActivityKind(_graphNode), _graphNode)
+    CAgentActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, const StatisticsMapping &_factoryStats)
+        : CActivityFactory(_graphNode.getPropInt("@id", 0), _subgraphId, _queryFactory, _helperFactory, getActivityKind(_graphNode), _graphNode, _factoryStats)
     {
     }
 
@@ -186,15 +189,11 @@ public:
     {
         return CActivityFactory::getKind();
     }
-
-    virtual void getEdgeProgressInfo(unsigned idx, IPropertyTree &edge) const
+    virtual const StatisticsMapping &queryStatsMapping() const
     {
-        CActivityFactory::getEdgeProgressInfo(idx, edge);
-    }
-    virtual void getNodeProgressInfo(IPropertyTree &node) const
-    {
-        CActivityFactory::getNodeProgressInfo(node);
+        return mystats.queryMapping();
     }
+
     virtual void resetNodeProgressInfo()
     {
         CActivityFactory::resetNodeProgressInfo();
@@ -203,6 +202,23 @@ public:
     {
         CActivityFactory::getActivityMetrics(reply);
     }
+    virtual void gatherStats(IStatisticGatherer &builder, int channel, bool reset) const override
+    {
+        //Because subgraphs are flattened in roxie the subgraph needs to be selected for each activity
+        assertex(channel != -1);
+        StatsSubgraphScope sg(builder, subgraphId);
+        StatsActivityScope ac(builder, id);
+        {
+            ChannelActivityScope ch(builder, channel);
+            mystats.recordStatistics(builder, reset);
+        }
+        ForEachItemIn(i, childQueries)
+        {
+            auto child = childQueries.item(i);
+            StatsChildGraphScope cc(builder, childQueryIndexes.item(i));
+            child.gatherStats(builder, channel, reset);
+        }
+    }
     IRoxieAgentContext *createAgentContext(const AgentContextLogger &logctx, IRoxieQueryPacket *packet) const
     {
         return queryFactory.createAgentContext(logctx, packet, childQueries.length()!=0);
@@ -218,7 +234,7 @@ public:
         {
             ForEachItemIn(idx, childQueries)
             {
-                if (!_probeManager) // MORE - the probeAllRows is a hack!
+                if (!_probeManager)
                     _probeManager = queryContext->queryProbeManager();
                 IActivityGraph *childGraph = createActivityGraph(ctx, NULL, childQueryIndexes.item(idx), childQueries.item(idx), NULL, _probeManager, logctx, 1); // MORE - the parent is wrong!
                 childGraphs.append(*childGraph);
@@ -331,14 +347,7 @@ protected:
     virtual void onCreate()
     {
         queryContext.setown(basefactory->createAgentContext(logctx, packet));
-#ifdef _DEBUG
-        // MORE - need to consider debugging....
-        if (probeAllRows)
-            probeManager.setown(createProbeManager());
-        basefactory->createChildQueries(queryContext, childGraphs, basehelper, probeManager, queryContext, logctx);
-#else
         basefactory->createChildQueries(queryContext, childGraphs, basehelper, NULL, queryContext, logctx);
-#endif
         if (meta.needsSerializeDisk())
             serializer.setown(meta.createDiskSerializer(queryContext->queryCodeContext(), basefactory->queryId()));
         if (needsRowAllocator())
@@ -388,10 +397,12 @@ protected:
 
     virtual void beforeDispose() override
     {
-        CRuntimeStatisticCollection merged(allStatistics);
-        logctx.gatherStats(merged);
         if (defaultCollectFactoryStatistics)
+        {
+            CRuntimeStatisticCollection merged(basefactory->queryStatsMapping());
+            logctx.gatherStats(merged);
             basefactory->mergeStats(merged);
+        }
     }
 
 public:
@@ -984,6 +995,8 @@ public:
 
 };
 
+static const StatisticsMapping diskAgentStatistics({StNumDiskRowsRead, StNumDiskSeeks, StNumDiskAccepted, StNumDiskRejected });
+
 class CRoxieDiskBaseActivityFactory : public CAgentActivityFactory
 {
 protected:
@@ -991,7 +1004,7 @@ protected:
     Owned<IInMemoryIndexManager> manager;
 public:
     CRoxieDiskBaseActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
-        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
+        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, diskAgentStatistics)
     {
         Owned<IHThorDiskReadBaseArg> helper = (IHThorDiskReadBaseArg *) helperFactory();
         bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getFlags() & (TDXvarfilename|TDXdynamicfilename)) != 0);
@@ -2402,17 +2415,24 @@ protected:
     IOutputMetaData *projectedMeta = nullptr;
     IOutputMetaData *expectedMeta = nullptr;
 
-    CRoxieKeyedActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
-        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
+    CRoxieKeyedActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, const StatisticsMapping &_factoryStats)
+        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, _factoryStats)
     {
     }
 };
 
+static const StatisticsMapping indexAgentStats({StNumIndexSeeks, StNumIndexScans, StNumIndexWildSeeks,
+                                                StNumIndexSkips, StNumIndexNullSkips, StNumIndexMerges, StNumIndexMergeCompares,
+                                                StNumPreFiltered, StNumPostFiltered, StNumIndexAccepted, StNumIndexRejected,
+                                                StNumBlobCacheHits, StNumLeafCacheHits, StNumNodeCacheHits,
+                                                StNumBlobCacheAdds, StNumLeafCacheAdds, StNumNodeCacheAdds,
+                                                StNumIndexRowsRead});
+
 class CRoxieIndexActivityFactory : public CRoxieKeyedActivityFactory
 {
 public:
     CRoxieIndexActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
-        : CRoxieKeyedActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
+        : CRoxieKeyedActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, indexAgentStats)
     {
     }
 
@@ -3635,7 +3655,7 @@ public:
     Owned<IFileIOArray> fileArray;
 
     CRoxieFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
-        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
+        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, diskAgentStatistics)
     {
         Owned<IHThorFetchBaseArg> helper = (IHThorFetchBaseArg *) helperFactory();
         bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getFetchFlags() & (FFvarfilename|FFdynamicfilename)) != 0);
@@ -4029,7 +4049,7 @@ class CRoxieKeyedJoinIndexActivityFactory : public CRoxieKeyedActivityFactory
 {
 public:
     CRoxieKeyedJoinIndexActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
-        : CRoxieKeyedActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
+        : CRoxieKeyedActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, indexAgentStats)
     {
         Owned<IHThorKeyedJoinArg> helper = (IHThorKeyedJoinArg *) helperFactory();
         bool variableFileName = allFilesDynamic || queryFactory.isDynamic() || ((helper->getJoinFlags() & (JFvarindexfilename|JFdynamicindexfilename|JFindexfromactivity)) != 0);
@@ -4415,7 +4435,7 @@ public:
     Owned<IFileIOArray> files;
 
     CRoxieKeyedJoinFetchActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory)
-        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory)
+        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, _helperFactory, diskAgentStatistics)
     {
         Owned<IHThorKeyedJoinArg> helper = (IHThorKeyedJoinArg *) helperFactory();
         assertex(helper->diskAccessRequired());
@@ -4724,7 +4744,7 @@ class CRoxieRemoteActivityFactory : public CAgentActivityFactory
 
 public:
     CRoxieRemoteActivityFactory(IPropertyTree &graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, unsigned _remoteId)
-        : CAgentActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory), remoteId(_remoteId)
+        : CAgentActivityFactory(graphNode, _subgraphId, _queryFactory, _helperFactory, allStatistics), remoteId(_remoteId)  // MORE is all statistics right? Or is it none?
     {
     }
 
@@ -4756,7 +4776,7 @@ protected:
 
 public:
     CRoxieDummyActivityFactory(IPropertyTree &_graphNode, unsigned _subgraphId, IQueryFactory &_queryFactory, bool isLoadDataOnly)
-        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, NULL)
+        : CAgentActivityFactory(_graphNode, _subgraphId, _queryFactory, NULL, allStatistics)
     {
         if (_graphNode.getPropBool("att[@name='_isSpill']/@value", false) || _graphNode.getPropBool("att[@name='_isSpillGlobal']/@value", false))
             return;  // ignore 'spills'

roxie/ccd/ccdactivities.hpp

@@ -37,14 +37,13 @@ interface IIndexReadActivityInfo;
 interface IActivityFactory : extends IInterface
 {
     virtual void addChildQuery(unsigned id, ActivityArray *childQuery) = 0;
-    virtual void getEdgeProgressInfo(unsigned idx, IPropertyTree &edge) const = 0;
-    virtual void getNodeProgressInfo(IPropertyTree &node) const = 0;
     virtual ActivityArray *queryChildQuery(unsigned idx, unsigned &id) = 0;
     virtual unsigned queryId() const = 0;
     virtual void resetNodeProgressInfo() = 0;
     virtual IQueryFactory &queryQueryFactory() const = 0;
     virtual ThorActivityKind getKind() const = 0;
     virtual void getActivityMetrics(StringBuffer &reply) const = 0;
+    virtual void gatherStats(IStatisticGatherer &builder, int channel, bool reset) const = 0;
     virtual void getXrefInfo(IPropertyTree &reply, const IRoxieContextLogger &logctx) const = 0;
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const = 0;
 };

roxie/ccd/ccdcontext.cpp

@@ -1194,7 +1194,6 @@ protected:
     MapXToMyClass<unsigned, unsigned, IActivityGraph> childGraphs;
     Owned<IActivityGraph> graph;
     StringBuffer authToken;
-    Owned<IPropertyTree> probeQuery;
     unsigned lastWuAbortCheck;
     unsigned startTime;
     std::atomic<unsigned> totAgentsReplyLen = {0};
@@ -1208,7 +1207,6 @@ protected:
     Owned<IRoxieDaliHelper> daliHelperLink;
     Owned<IDistributedFileTransaction> superfileTransaction;
 
-    mutable CriticalSection statsCrit;
     const IRoxieContextLogger &logctx;
 
 
@@ -1307,6 +1305,11 @@ public:
         logctx.noteStatistic(kind, value);
     }
 
+    virtual void setStatistic(StatisticKind kind, unsigned __int64 value) const
+    {
+        logctx.setStatistic(kind, value);
+    }
+
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const
     {
         logctx.mergeStats(from);
@@ -1525,16 +1528,14 @@ public:
             probeManager.setown(createDebugManager(debugContext, graphName));
             debugContext->checkBreakpoint(DebugStateGraphCreate, NULL, graphName);
         }
-        else if (probeAllRows || probeQuery != NULL)
-            probeManager.setown(createProbeManager());
         graph.setown(factory->lookupGraph(this, graphName, probeManager, *this, NULL));
         graph->onCreate(NULL);  // MORE - is that right
         if (debugContext)
             debugContext->checkBreakpoint(DebugStateGraphStart, NULL, graphName);
         if (workUnit)
-            graphStats.setown(workUnit->updateStats(graph->queryName(), SCTroxie, queryStatisticsComponentName(), graph->queryWorkflowId(), 0));
+            graphStats.setown(workUnit->updateStats(graph->queryName(), SCTroxie, queryStatisticsComponentName(), graph->queryWorkflowId(), 0, false));
         else if (statsWu)
-            graphStats.setown(statsWu->updateStats(graph->queryName(), SCTroxie, queryStatisticsComponentName(), graph->queryWorkflowId(), 0));
+            graphStats.setown(statsWu->updateStats(graph->queryName(), SCTroxie, queryStatisticsComponentName(), graph->queryWorkflowId(), 0, true));
     }
 
     IWorkUnit *updateStatsWorkUnit() const
@@ -1606,20 +1607,7 @@ public:
 
     void runGraph()
     {
-        try
-        {
-            graph->execute();
-
-            if (probeQuery)
-                graph->getProbeResponse(probeQuery);
-
-        }
-        catch(...)
-        {
-            if (probeQuery)
-                graph->getProbeResponse(probeQuery);
-            throw;
-        }
+        graph->execute();
     }
 
     virtual void executeGraph(const char * name, bool realThor, size32_t parentExtractSize, const void * parentExtract)
@@ -2353,6 +2341,8 @@ public:
         // NOTE: This is needed to ensure that owned activities are destroyed BEFORE I am,
         // to avoid pure virtual calls when they come to call noteProcessed()
         logctx.mergeStats(globalStats);
+        if (factory)
+            factory->mergeStats(logctx);
         childGraphs.releaseAll();
     }
 
@@ -2581,33 +2571,18 @@ protected:
 
     void doPostProcess()
     {
-        if (workUnit || statsWu)
+        logctx.mergeStats(globalStats);
+        logctx.setStatistic(StTimeTotalExecute, elapsedTimer.elapsedNs());
+        if (factory)
         {
-            WorkunitUpdate w(updateStatsWorkUnit());
-            Owned<IStatisticGatherer> builder = createGlobalStatisticGatherer(w);
-            globalStats.recordStatistics(*builder);
+            factory->mergeStats(logctx);
         }
-        logctx.mergeStats(globalStats);
         globalStats.reset();
         if (!protocol)
             return;
 
         if (!isRaw && !isBlocked)
             protocol->flush();
-
-        if (probeQuery)
-        {
-            // loop through all of the graphs and create a _Probe to output each xgmml
-            Owned<IPropertyTreeIterator> graphs = probeQuery->getElements("Graph");
-            ForEach(*graphs)
-            {
-                IPropertyTree &graph = graphs->query();
-
-                StringBuffer xgmml;
-                _toXML(&graph, xgmml, 0);
-                protocol->appendProbeGraph(xgmml.str());
-            }
-        }
     }
     void addWuException(IException *E)
     {
@@ -2731,23 +2706,13 @@ public:
         sendHeartBeats = enableHeartBeat && isRaw && isBlocked && options.priority==0;
 
         const char *wuid = context->queryProp("@wuid");
-        if (!wuid && options.statsToWorkunit)
+        if (options.statsToWorkunit)
         {
             IRoxieDaliHelper *daliHelper = checkDaliConnection();
-            if (daliHelper)
+            if (daliHelper->connected())
             {
-                // MORE - can we accumulate results from several runs?
-                statsWu.setown(daliHelper->createWorkUnit());
-                // Rather than copying the entire workunit, for example via
-                //    queryExtendedWU(statsWu)->copyWorkUnit(_factory->queryWorkUnit(), false, true);
-                // we create a blank workunit with a reference to the original workunit that allows the graph info to be patched in as needed
-                StringBuffer dllFileName;
-                splitFilename(_factory->queryDll()->queryName(), nullptr, nullptr, &dllFileName, nullptr, false);
-                if (strlen(SharedObjectPrefix))
-                    dllFileName.replaceString(SharedObjectPrefix, "");
-                queryExtendedWU(statsWu)->queryPTree()->setProp("@clonedFromWorkunit", dllFileName);
+                statsWu.setown(daliHelper->createStatsWorkUnit(wuid, _factory->queryDll()->queryName()));
                 WorkunitUpdate wu(&statsWu->lock());
-                addTimeStamp(wu, SSTglobal, NULL, StWhenStarted);
                 wu->setState(WUStateRunning);
                 VStringBuffer jobname("Stats for %s", _factory->queryQueryName());
                 const char *statsID = context->queryProp("@statsId");
@@ -2773,8 +2738,6 @@ public:
             if (debugUID && *debugUID)
                 initDebugMode(breakAtStart, debugUID);
         }
-        else if (context->getPropBool("_Probe", false))
-            probeQuery.setown(_factory->cloneQueryXGMML());
 
         // MORE some of these might be appropriate in wu case too?
         rowManager->setActivityTracking(context->getPropBool("_TraceMemory", false));
@@ -2937,9 +2900,9 @@ public:
         addTimeStamp(w, SSTglobal, NULL, StWhenFinished);
         updateWorkunitTimings(w, myTimer);
         Owned<IStatisticGatherer> gatherer = createGlobalStatisticGatherer(w);
-        CRuntimeStatisticCollection merged(allStatistics);
+        CRuntimeStatisticCollection merged(accumulatedStatistics);
         logctx.gatherStats(merged);
-        merged.recordStatistics(*gatherer);
+        merged.recordStatistics(*gatherer, false);
 
         //MORE: If executed more than once (e.g., scheduled), then TimeElapsed isn't particularly correct.
         gatherer->updateStatistic(StTimeElapsed, elapsedTimer.elapsedNs(), StatsMergeReplace);

roxie/ccd/ccddali.cpp

@@ -462,11 +462,28 @@ public:
         return userdesc;
     }
 
-    virtual IWorkUnit *createWorkUnit() override
+    virtual IConstWorkUnit *createStatsWorkUnit(const char *wuid, const char *dllname) const override
     {
         Owned<IWorkUnitFactory> wuFactory = getWorkUnitFactory();
-        return wuFactory->createWorkUnit("roxie", ""); // NOTE - scope is later overridden by clone operation
-        // MORE - What about security parms?
+        Owned<IWorkUnit> wu;
+        if (!wuid || streq(wuid, "*"))
+        {
+            wu.setown(wuFactory->createWorkUnit("roxie", ""));  // MORE - What about security parms?
+            StringBuffer dllFileName;
+            splitFilename(dllname, nullptr, nullptr, &dllFileName, nullptr, false);
+            if (strlen(SharedObjectPrefix))
+                dllFileName.replaceString(SharedObjectPrefix, "");
+            queryExtendedWU(wu)->queryPTree()->setProp("@clonedFromWorkunit", dllFileName);
+            addTimeStamp(wu, SSTglobal, NULL, StWhenStarted);
+            wu->setState(WUStateRunning);
+        }
+        else
+        {
+            wu.setown(wuFactory->updateWorkUnit(wuid));
+            if (!wu)
+                throw makeStringExceptionV(ROXIE_CONTROL_MSG_ERROR, "Can't open stats WU %s", wuid);
+        }
+        return wu->unlock();
     }
 
     static const char *getQuerySetPath(StringBuffer &buf, const char *id)

roxie/ccd/ccddali.hpp

@@ -62,7 +62,7 @@ interface IRoxieDaliHelper : extends IInterface
     virtual void noteWorkunitRunning(const char *wu, bool running) = 0;
     virtual StringBuffer &getDaliIp(StringBuffer &ip) const = 0;
     virtual IUserDescriptor *queryUserDescriptor() = 0;
-    virtual IWorkUnit *createWorkUnit() = 0;
+    virtual IConstWorkUnit *createStatsWorkUnit(const char *wuid, const char *dllName) const = 0;
 };
 
 

roxie/ccd/ccddebug.cpp

@@ -195,268 +195,6 @@ public:
 };
 
 
-class TraceProbe : public InputProbe
-{
-public:
-    TraceProbe(IFinalRoxieInput *_in, unsigned _sourceId, unsigned _targetId, unsigned _sourceIdx, unsigned _targetIdx, unsigned _iteration, unsigned _channel)
-        : InputProbe(_in, NULL, _sourceId, _sourceIdx, _targetId, _targetIdx, _iteration, _channel)
-    {
-    }
-
-    bool matches(IPropertyTree &edge, bool forNode)
-    {
-        if (forNode)
-        {
-            unsigned id = edge.getPropInt("@id", 0);
-            if (id && (id == sourceId || id == targetId))
-            {
-                return true;
-            }
-        }
-        else
-        {
-            unsigned id = edge.getPropInt("@source", 0);
-            if (id && id == sourceId)
-            {
-                id = edge.getPropInt("@target", 0);
-                if (id && id == targetId)
-                {
-                    unsigned idx = edge.getPropInt("att[@name=\"_sourceIndex\"]/@value", 0);
-                    if (idx == sourceIdx)
-                        return true;
-                }
-            }
-            id = edge.getPropInt("att[@name=\"_sourceActivity\"]/@value");
-            if (id && id == sourceId)
-            {
-                id = edge.getPropInt("att[@name=\"_targetActivity\"]/@value");
-                if (id && id == targetId)
-                {
-                    unsigned idx = edge.getPropInt("att[@name=\"_sourceIndex\"]/@value", 0);
-                    if (idx == sourceIdx)
-                        return true;
-                }
-            }
-        }
-        return false;
-    }
-
-    const void * _next(const void *inputRow)
-    {
-        const byte *ret = (const byte *) inputRow;
-        if (ret && probeAllRows)
-        {
-            CommonXmlWriter xmlwrite(XWFnoindent|XWFtrim|XWFopt);
-            if (inMeta && inMeta->hasXML())
-                inMeta->toXML(ret, xmlwrite);
-            DBGLOG("ROW: [%d->%d] {%p} %s", sourceId, targetId, ret, xmlwrite.str());
-        }
-        return ret;
-    }
-
-    virtual const void * nextRowGE(const void * seek, unsigned numFields, bool &wasCompleteMatch, const SmartStepExtra & stepExtra)
-    {
-        // MORE - should probably only note them when wasCompleteMatch is true?
-        return _next(InputProbe::nextRowGE(seek, numFields, wasCompleteMatch, stepExtra));
-    }
-    virtual const void *nextRow()
-    {
-        return _next(InputProbe::nextRow());
-    }
-
-    void getNodeProgressInfo(IPropertyTree &node)
-    {
-        // node is the input (or possibly output) of this probe edge
-        unsigned started = everStarted;
-        putStatsValue(&node, "_roxieStarted", "sum", started);
-        unsigned id = node.getPropInt("@id", 0);
-        bool isOutput = (id != 0) && (id != sourceId);
-        unsigned totalTime = (unsigned) (cycle_to_nanosec(in->queryTotalCycles())/1000);
-        if (isOutput)
-            totalTime += 10; // Fudge factor - I don't really know the times but this makes the graph more useable than not supplying a totalTime value
-        if (totalTime)
-            putStatsValue(&node, "TimeTotalExecute", "sum", totalTime);
-        unsigned localTime = isOutput ? 10 : (unsigned) (cycle_to_nanosec(in->queryActivity()->queryLocalCycles())/1000); // Fudge factor - I don't really know the times but this makes the graph more useable than not supplying a localTime value
-        if (localTime)
-            putStatsValue(&node, "TimeLocalExecute", "sum", localTime);
-    }
-
-    void getEdgeProgressInfo(IPropertyTree &edge)
-    {
-        putStatsValue(&edge, "_roxieStarted", "sum", hasStarted);
-        if (hasStarted)
-        {
-            putStatsValue(&edge, "NumRowsProcessed", "sum", totalRowCount);
-            putStatsValue(&edge, "SizeMaxRowSize", "max", maxRowSize);
-        }
-    }
-};
-
-class CProbeManager : implements IProbeManager, public CInterface
-{
-    IArrayOf<IFinalRoxieInput> probes; // May want to replace with hash table at some point....
-public:
-    IMPLEMENT_IINTERFACE;
-
-    IRoxieProbe *createProbe(IInputBase *in, IActivityBase *inAct, IActivityBase *outAct, unsigned sourceIdx, unsigned targetIdx, unsigned iteration)
-    {
-        unsigned idIn = inAct->queryId();
-        unsigned idOut = outAct->queryId();
-        TraceProbe *probe = new TraceProbe(static_cast<IFinalRoxieInput*>(in), idIn, idOut, sourceIdx, targetIdx, iteration, 0);
-        probes.append(*probe);
-        return probe;
-    }
-
-    TraceProbe *findProbe(IPropertyTree &edge, bool forNode, unsigned &startat)
-    {
-        // MORE - this is n-squared on number of edges in the graph. Could get painful - recode if needed
-        // However I think that the "startat" cache probably prevents the pain
-        unsigned probeCount = probes.ordinality();
-        unsigned search = probeCount;
-        unsigned idx = startat;
-        while (search--)
-        {
-            idx++;
-            if (idx>=probeCount) idx = 0;
-            TraceProbe &p = static_cast<TraceProbe &> (probes.item(idx));
-            if (p.matches(edge, forNode))
-            {
-                startat = idx;
-                return &p;
-            }
-        }
-        return NULL;
-    }
-
-    virtual void noteSink(IActivityBase *)
-    {
-    }
-
-    virtual IDebugGraphManager *queryDebugManager()
-    {
-        return NULL;
-    }
-
-    virtual void noteDependency(IActivityBase *sourceActivity, unsigned sourceIndex, unsigned controlId, const char *edgeId, IActivityBase *targetActivity)
-    {
-    }
-
-    virtual IProbeManager *startChildGraph(unsigned childGraphId, IActivityBase *parent)
-    {
-        return LINK(this);
-    }
-
-    virtual void endChildGraph(IProbeManager *child, IActivityBase *parent)
-    {
-    }
-
-    virtual void deleteGraph(IArrayOf<IActivityBase> *activities, IArrayOf<IInputBase> *goers)
-    {
-        if (goers)
-        {
-            ForEachItemIn(probeIdx, *goers)
-            {
-                TraceProbe &probe = (TraceProbe &) goers->item(probeIdx);
-                probes.zap(probe);
-            }
-        }
-    }
-
-    virtual void setNodeProperty(IActivityBase *node, const char *propName, const char *propVvalue)
-    {
-        // MORE - we could note these in probe mode too...
-    }
-
-    virtual void setNodePropertyInt(IActivityBase *node, const char *propName, unsigned __int64 propVvalue)
-    {
-        // MORE - we could note these in probe mode too...
-    }
-
-    virtual void getProbeResponse(IPropertyTree *query)
-    {
-        Owned<IPropertyTreeIterator> graphs = query->getElements("Graph");
-        ForEach(*graphs)
-        {
-            IPropertyTree &graph = graphs->query();
-            Owned<IPropertyTreeIterator> subgraphs = graph.getElements("xgmml/graph");
-            ForEach(*subgraphs)
-            {
-                IPropertyTree &subgraph = subgraphs->query();
-                Owned<IPropertyTreeIterator> nodes = subgraph.getElements(".//node");
-                unsigned startat = 0;
-                ForEach(*nodes)
-                {
-                    IPropertyTree &node = nodes->query();
-                    TraceProbe *currentProbe = findProbe(node, true, startat);
-                    if (currentProbe)
-                    {
-                        currentProbe->getNodeProgressInfo(node);
-                    }
-                }
-                Owned<IPropertyTreeIterator> edges = subgraph.getElements(".//edge");
-                startat = 0;
-                ForEach(*edges)
-                {
-                    IPropertyTree &edge = edges->query();
-                    if (edge.getPropInt("att[@name='_dependsOn']/@value", 0) != 0)
-                    {
-                        const char *targetNode = edge.queryProp("att[@name='_targetActivity']/@value");
-                        if (targetNode)
-                        {
-                            StringBuffer xpath;
-                            IPropertyTree *target = query->queryPropTree(xpath.append(".//node[@id='").append(targetNode).append("']"));
-                            if (target)
-                            {
-                                unsigned started = target->getPropInt("att[@name='_roxieStarted']/@value", 0);
-                                IPropertyTree *att = edge.queryPropTree("att[@name=\"_roxieStarted\"]");
-                                if (!att)
-                                {
-                                    att = edge.addPropTree("att");
-                                    att->setProp("@name", "_roxieStarted");
-                                }
-                                else
-                                    started += att->getPropInt("@value");
-                                att->setPropInt("@value", started);
-                            }
-
-                        }
-                    }
-                    else
-                    {
-                        TraceProbe *currentProbe = findProbe(edge, false, startat);
-                        if (currentProbe)
-                        {
-                            currentProbe->getEdgeProgressInfo(edge);
-                        }
-                        else
-                        {
-                            const char *targetNode = edge.queryProp("att[@name='_targetActivity']/@value");
-                            if (targetNode)
-                            {
-                                StringBuffer xpath;
-                                IPropertyTree *target = query->queryPropTree(xpath.append(".//node[@id='").append(targetNode).append("']"));
-                                if (target)
-                                {
-                                    unsigned started = target->getPropInt("att[@name='_roxieStarted']/@value", 0);
-                                    IPropertyTree *att = edge.queryPropTree("att[@name=\"_roxieStarted\"]");
-                                    if (!att)
-                                    {
-                                        att = edge.addPropTree("att");
-                                        att->setProp("@name", "_roxieStarted");
-                                    }
-                                    else
-                                        started += att->getPropInt("@value");
-                                    att->setPropInt("@value", started);
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        }
-    }
-};
-
 typedef const IInterface *CIptr;
 typedef MapBetween<unsigned, unsigned, CIptr, CIptr> ProxyMap;
 static ProxyMap *registeredProxies;
@@ -1001,11 +739,6 @@ public:
     }
 };
 
-extern IProbeManager *createProbeManager()
-{
-    return new CProbeManager;
-}
-
 IDebugGraphManager *createProxyDebugGraphManager(unsigned graphId, unsigned channel, memsize_t remoteGraphId);
 
 class CRoxieDebugGraphManager : extends CBaseDebugGraphManager

roxie/ccd/ccddebug.hpp

@@ -26,7 +26,6 @@ interface IProbeManager;
 interface IDebuggableContext;
 
 extern void doDebugRequest(IRoxieQueryPacket *packet, const IRoxieContextLogger &logctx);
-extern IProbeManager *createProbeManager();
 extern IProbeManager *createDebugManager(IDebuggableContext *debugContext, const char *graphName);
 
 #endif

roxie/ccd/ccdlistener.cpp

@@ -412,9 +412,9 @@ public:
         // So do the query ourselves and in all child threads;
         const char *name = xml->queryName();
         CascadeMergeType mergeType=CascadeMergeNone;
-        if (strstr(name, "querystats"))
+        if (strieq(name, "control:querystats"))
             mergeType=CascadeMergeStats;
-        else if (strstr(name, "queries"))
+        else if (strieq(name, "control:queries"))
             mergeType=CascadeMergeQueries;
         Owned<IPropertyTree> mergedReply;
         if (mergeType!=CascadeMergeNone)

roxie/ccd/ccdmain.cpp

@@ -184,7 +184,6 @@ StringBuffer defaultPlaneDirPrefix;
 bool trapTooManyActiveQueries;
 unsigned maxEmptyLoopIterations;
 unsigned maxGraphLoopIterations;
-bool probeAllRows;
 bool steppingEnabled = true;
 bool simpleLocalKeyedJoins = true;
 bool adhocRoxie = false;

roxie/ccd/ccdprotocol.cpp

@@ -859,24 +859,6 @@ public:
             }
         }
     }
-    virtual void appendProbeGraph(const char *xml)
-    {
-        if (!xml)
-        {
-            if (probe)
-                probe.clear();
-            return;
-        }
-        if (!probe)
-        {
-            probe.setown(new FlushingStringBuffer(client, isBlocked, MarkupFmt_XML, false, isHTTP, logctx));
-            probe->startDataset("_Probe", NULL, (unsigned) -1);  // initialize it
-        }
-
-        probe->append("\n");
-        probe->append(xml);
-    }
-
 };
 
 class CHpccXmlResultsWriter : public CHpccNativeResultsWriter
@@ -1115,11 +1097,6 @@ public:
         ForEachItemIn(i, contentsMap)
             contentsMap.item(i)->flush(true);
     }
-    virtual void appendProbeGraph(const char *xml)
-    {
-        if (results)
-            results->appendProbeGraph(xml);
-    }
 };
 
 class CHpccJsonResponse : public CHpccNativeProtocolResponse
@@ -1658,7 +1635,7 @@ private:
             if (!uid)
                 uid = queryPT->queryProp("_TransactionId");
             isBlind = queryPT->getPropBool("@blind", false) || queryPT->getPropBool("_blind", false);
-            isDebug = queryPT->getPropBool("@debug") || queryPT->getPropBool("_Probe", false);
+            isDebug = queryPT->getPropBool("@debug");
             toXML(queryPT, saniText, 0, isBlind ? (XML_SingleQuoteAttributeValues | XML_Sanitize) : XML_SingleQuoteAttributeValues);
         }
     }

roxie/ccd/ccdquery.cpp

@@ -70,6 +70,14 @@ unsigned ActivityArray::recursiveFindActivityIndex(unsigned id)
     return NotFound;
 }
 
+void ActivityArray::gatherStats(IStatisticGatherer &builder, int channel, bool reset)
+{
+    ForEachItemIn(idx, activities)
+    {
+        activities.item(idx).gatherStats(builder, channel, reset);
+    }
+}
+
 //----------------------------------------------------------------------------------------------
 // Class CQueryDll maps dlls into loadable workunits, complete with caching to ensure that a refresh of the QuerySet 
 // can avoid reloading dlls, and that the same CQueryDll (and the objects it owns) can be shared between server and 
@@ -574,6 +582,7 @@ protected:
 
     mutable CIArrayOf<TerminationCallbackInfo> callbacks;
     mutable CriticalSection callbacksCrit;
+    mutable CRuntimeStatisticCollection stats;
 public:
     static CriticalSection queryCacheCrit;
 
@@ -1131,7 +1140,7 @@ public:
     unsigned channelNo;
 
     CQueryFactory(const char *_id, const IQueryDll *_dll, const IRoxiePackage &_package, hash64_t _hashValue, unsigned _channelNo, ISharedOnceContext *_sharedOnceContext, bool _dynamic)
-        : package(_package), dll(_dll), sharedOnceContext(_sharedOnceContext), id(_id), dynamic(_dynamic), hashValue(_hashValue), channelNo(_channelNo)
+        : package(_package), dll(_dll), sharedOnceContext(_sharedOnceContext), id(_id), dynamic(_dynamic), hashValue(_hashValue), stats(accumulatedStatistics), channelNo(_channelNo)
     {
         package.Link();
         targetClusterType = RoxieCluster;
@@ -1274,7 +1283,7 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
                         Owned<IConstWUGraphIterator> graphs = &wu->getGraphs(GraphTypeActivities);
                         ForEach(*graphs)
                         {
-                            Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false);
+                            Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false, false);
                             Owned<IPropertyTreeIterator> nodes = graphXgmml->getElements(".//node");
                             ForEach(*nodes)
                             {
@@ -1367,7 +1376,7 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
                 {
                     graphs->query().getName(graphNameStr);
                     const char *graphName = graphNameStr.s.str();
-                    Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false);
+                    Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false, false);
                     try
                     {
                         ActivityArray *activities = loadGraph(*graphXgmml, graphName);
@@ -1432,33 +1441,6 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
         return ret.getClear();
     }
 
-    void getGraphStats(StringBuffer &reply, const IPropertyTree &thisGraph) const
-    {
-        Owned<IPropertyTree> graph = createPTreeFromIPT(&thisGraph, ipt_lowmem);
-        Owned<IPropertyTreeIterator> edges = graph->getElements(".//edge");
-        ForEach(*edges)
-        {
-            IPropertyTree &edge = edges->query();
-            IActivityFactory *a = findActivity(edge.getPropInt("@source", 0));
-            if (!a)
-                a = findActivity(edge.getPropInt("att[@name=\"_sourceActivity\"]/@value", 0));
-            if (a)
-            {
-                unsigned sourceOutput = edge.getPropInt("att[@name=\"_sourceIndex\"]/@value", 0);
-                a->getEdgeProgressInfo(sourceOutput, edge);
-            }
-        }
-        Owned<IPropertyTreeIterator> nodes = graph->getElements(".//node");
-        ForEach(*nodes)
-        {
-            IPropertyTree &node = nodes->query();
-            IActivityFactory *a = findActivity(node.getPropInt("@id", 0));
-            if (a)
-                a->getNodeProgressInfo(node);
-        }
-        toXML(graph, reply);
-    }
-
     virtual IPropertyTree* cloneQueryXGMML() const override
     {
         assertex(dll && dll->queryWorkUnit());
@@ -1469,7 +1451,7 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
         {
             graphs->query().getName(graphNameStr);
             const char *graphName = graphNameStr.s.str();
-            Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false);
+            Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false, false);
             IPropertyTree *newGraph = tree->addPropTree("Graph");
             newGraph->setProp("@id", graphName);
             newGraph->addPropTree("xgmml")->addPropTree("graph", graphXgmml.getLink());
@@ -1477,28 +1459,6 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
         return tree.getClear();
     }
 
-    virtual void getStats(StringBuffer &reply, const char *graphName) const override
-    {
-        if (dll)
-        {
-            assertex(dll->queryWorkUnit());
-            Owned<IConstWUGraphIterator> graphs = &dll->queryWorkUnit()->getGraphs(GraphTypeActivities);
-            SCMStringBuffer thisGraphNameStr;
-            ForEach(*graphs)
-            {
-                graphs->query().getName(thisGraphNameStr);
-                if (graphName)
-                {
-                    if (thisGraphNameStr.length() && (stricmp(graphName, thisGraphNameStr.s.str()) != 0))
-                        continue; // not interested in this one
-                }
-                reply.appendf("<Graph id='%s'><xgmml>", thisGraphNameStr.s.str());
-                Owned<IPropertyTree> graphXgmml = graphs->query().getXGMMLTree(false);
-                getGraphStats(reply, *graphXgmml);
-                reply.append("</xgmml></Graph>");
-            }
-        }
-    }
     virtual void getActivityMetrics(StringBuffer &reply) const override
     {
         HashIterator i(allActivities);
@@ -1509,12 +1469,44 @@ static hash64_t getQueryHash(const char *id, const IQueryDll *dll, const IRoxieP
             f->getActivityMetrics(myReply.clear());
             if (myReply.length())
             {
-                reply.appendf("  <activity query='%s' id='%d' channel='%d'\n", queryQueryName(), f->queryId(), queryChannel());
+                reply.appendf("  <activity query='%s' id='%d' channel='%d'>\n", queryQueryName(), f->queryId(), queryChannel());
                 reply.append(myReply);
-                reply.append("  </activity>\n");
+                reply.append("\n  </activity>\n");
             }
         }
     }
+    virtual void gatherStats(IConstWorkUnit *statsWu, int channel, bool reset) const override
+    {
+        if (dll)
+        {
+            assertex(dll->queryWorkUnit());
+            Owned<IConstWUGraphIterator> graphs = &dll->queryWorkUnit()->getGraphs(GraphTypeActivities);
+            SCMStringBuffer thisGraphNameStr;
+            ForEach(*graphs)
+            {
+                IConstWUGraph &graph = graphs->query();
+                graphs->query().getName(thisGraphNameStr);
+                Owned<IWUGraphStats> graphStats = statsWu->updateStats(thisGraphNameStr.str(), SCTroxie, queryStatisticsComponentName(), graph.getWfid(), 0, true);
+                IStatisticGatherer &builder = graphStats->queryStatsBuilder();
+                ActivityArrayPtr *activities = graphMap.getValue(thisGraphNameStr.str());
+                assertex(activities && *activities);
+                (*activities)->gatherStats(builder, channel, reset);
+            }
+            WorkunitUpdate w(&statsWu->lock());
+            Owned<IStatisticGatherer> gbuilder = createGlobalStatisticGatherer(w);
+            if (channel != -1)
+                gbuilder->beginChannelScope(channel);
+            stats.recordStatistics(*gbuilder, reset);
+        }
+    }
+    virtual void mergeStats(const CRuntimeStatisticCollection &from) const override
+    {
+        stats.merge(from);
+    }
+    virtual void mergeStats(const IRoxieContextLogger &from) const override
+    {
+        from.gatherStats(stats);
+    }
     virtual void getQueryInfo(StringBuffer &reply, bool full, IArrayOf<IQueryFactory> *agentQueries, const IRoxieContextLogger &logctx) const override
     {
         Owned<IPropertyTree> xref = createPTree("Query", ipt_fast);
@@ -1795,11 +1787,6 @@ public:
         else
             return NULL;
     }
-
-    virtual IPropertyTree *getQueryStats(time_t from, time_t to)
-    {
-        return queryStats->getStats(from, to);
-    }
 };
 
 unsigned checkWorkunitVersionConsistency(const IConstWorkUnit *wu)

roxie/ccd/ccdquery.hpp

@@ -140,7 +140,6 @@ interface IQueryFactory : extends IInterface
     virtual void suspend(const char *errMsg) = 0;
     virtual bool loadFailed() const = 0;
     virtual bool suspended() const = 0;
-    virtual void getStats(StringBuffer &reply, const char *graphName) const = 0;
     virtual void resetQueryTimings() = 0;
     virtual const QueryOptions &queryOptions() const = 0;
     virtual ActivityArray *lookupGraphActivities(const char *name) const = 0;
@@ -155,6 +154,9 @@ interface IQueryFactory : extends IInterface
 
     virtual const IRoxiePackage &queryPackage() const = 0;
     virtual void getActivityMetrics(StringBuffer &reply) const = 0;
+    virtual void gatherStats(IConstWorkUnit* statsWu, int channel, bool reset) const = 0;
+    virtual void mergeStats(const CRuntimeStatisticCollection &from) const = 0;
+    virtual void mergeStats(const IRoxieContextLogger &from) const = 0;
 
     virtual IPropertyTree *cloneQueryXGMML() const = 0;
     virtual CRoxieWorkflowMachine *createWorkflowMachine(IConstWorkUnit *wu, bool isOnce, const IRoxieContextLogger &logctx, const QueryOptions & options) const = 0;
@@ -199,6 +201,7 @@ public:
     inline IRoxieServerActivityFactory &serverItem(unsigned idx) const { return (IRoxieServerActivityFactory &) activities.item(idx); }
     void append(IActivityFactory &item);
     void setLibraryGraphId(unsigned value) { libraryGraphId = value; }
+    void gatherStats(IStatisticGatherer &builder, int channel, bool reset);
 
     inline unsigned ordinality() const { return activities.ordinality(); }
     inline bool isMultiInstance() const { return multiInstance; }
@@ -231,13 +234,11 @@ protected:
     ActivityArrayArray childQueries;
     UnsignedArray childQueryIndexes;
     CachedOutputMetaData meta;
-    mutable CriticalSection statsCrit;
     mutable CRuntimeStatisticCollection mystats;
-    // MORE: Could be CRuntimeSummaryStatisticCollection to include derived stats, but stats are currently converted
-    // to IPropertyTrees.  Would need to serialize/deserialize and then merge/derived so that they merged properly
+    mutable CRuntimeStatisticCollection myedgestats;
 
 public:
-    CActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode);
+    CActivityFactory(unsigned _id, unsigned _subgraphId, IQueryFactory &_queryFactory, HelperFactory *_helperFactory, ThorActivityKind _kind, IPropertyTree &_graphNode, const StatisticsMapping &_factoryStats);
     ~CActivityFactory() 
     { 
         ForEachItemIn(idx, childQueries)
@@ -254,28 +255,19 @@ public:
 
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const
     {
-        CriticalBlock b(statsCrit);
         mystats.merge(from);
     }
 
-    virtual void getEdgeProgressInfo(unsigned idx, IPropertyTree &edge) const
-    {
-        // No meaningful edge info for remote agent activities...
-    }
-
-    virtual void getNodeProgressInfo(IPropertyTree &node) const
-    {
-        mystats.getNodeProgressInfo(node);
-    }
-
     virtual void resetNodeProgressInfo()
     {
         mystats.reset();
+        myedgestats.reset();
     }
 
     virtual void getActivityMetrics(StringBuffer &reply) const
     {
-        mystats.toXML(reply);
+        mystats.toStr(reply);
+        myedgestats.toStr(reply);
     }
     virtual void getXrefInfo(IPropertyTree &reply, const IRoxieContextLogger &logctx) const
     {

roxie/ccd/ccdstate.cpp

@@ -26,6 +26,7 @@
 #include "udptopo.hpp"
 #include "ccd.hpp"
 #include "ccdquery.hpp"
+#include "ccddali.hpp"
 #include "ccdstate.hpp"
 #include "ccdqueue.ipp"
 #include "ccdlistener.hpp"
@@ -1064,14 +1065,12 @@ public:
             hash = rtlHash64VStr("active", hash);
     }
 
-    virtual void getStats(const char *queryName, const char *graphName, StringBuffer &reply, const IRoxieContextLogger &logctx) const
+    virtual void getStats(const char *queryName, const char *graphName, IConstWorkUnit *statsWu, unsigned channel, bool reset, const IRoxieContextLogger &logctx) const override
     {
         Owned<IQueryFactory> f = getQuery(queryName, NULL, logctx);
         if (f)
         {
-            reply.appendf("<Query id='%s'>\n", queryName);
-            f->getStats(reply, graphName);
-            reply.append("</Query>\n");
+            f->gatherStats(statsWu, channel, reset);
         }
         else
             throw MakeStringException(ROXIE_UNKNOWN_QUERY, "Unknown query %s", queryName);
@@ -1437,7 +1436,7 @@ public:
         return true;
     }
 
-    bool getStats(const char *queryId, const char *action, const char *graphName, StringBuffer &reply, const IRoxieContextLogger &logctx) const
+    bool getStats(const char *queryId, const char *graphName, StringBuffer &reply, const char *wuid, const IRoxieContextLogger &logctx) const
     {
         Owned<IRoxieQuerySetManager> serverManager;
         Owned<CRoxieAgentQuerySetManagerSet> agentManagers;
@@ -1447,18 +1446,45 @@ public:
             Owned<IQueryFactory> query = serverManager->getQuery(queryId, NULL, logctx);
             if (query)
             {
-                StringBuffer freply;
-                serverManager->getStats(queryId, graphName, freply, logctx);
-                Owned<IPropertyTree> stats = createPTreeFromXMLString(freply.str(), ipt_fast);
+                bool reset = false;  // MORE - tidy up around here.
+                Owned<IConstWorkUnit> statsWu;
+                if (wuid)
+                {
+                    Owned<IRoxieDaliHelper> daliHelper = ::connectToDali();
+                    if (!daliHelper->connected())
+                        throw makeStringException(ROXIE_CONTROL_MSG_ERROR, "Can't create stats WU - dali not connected");
+                    statsWu.setown(daliHelper->createStatsWorkUnit(wuid, query->queryDll()->queryName()));
+                }
+                else
+                {
+                    statsWu.setown(createLocalWorkUnitFromPTree(createPTreeFromIPT(queryExtendedWU(query->queryWorkUnit())->queryPTree())));
+                }
+                query->gatherStats(statsWu, -1, reset);
                 for (unsigned channel = 0; channel < numChannels; channel++)
                     if (agentManagers->item(channel))
+                        agentManagers->item(channel)->getStats(queryId, graphName, statsWu, channel+1, reset, logctx);
+                if (!wuid || *wuid=='*')
+                {
+                    WorkunitUpdate wu(&statsWu->lock());
+                    wu->setState(WUStateCompleted);   // We don't set the state when updating existing workunits
+                }
+                reply.appendf("<Query id='%s'>\n", queryId);
+                if (wuid)
+                    reply.appendf(" <wuid>%s</wuid>\n", statsWu->queryWuid());
+                else
+                {
+                    Owned<IConstWUGraphIterator> graphs = &statsWu->getGraphs(GraphTypeActivities);
+                    ForEach(*graphs)
                     {
-                        StringBuffer sreply;
-                        agentManagers->item(channel)->getStats(queryId, graphName, sreply, logctx);
-                        Owned<IPropertyTree> cstats = createPTreeFromXMLString(sreply.str(), ipt_fast);
-                        mergeStats(stats, cstats, 1);
+                        IConstWUGraph &graph = graphs->query();
+                        SCMStringBuffer s;
+                        reply.appendf("<Graph id='%s'>\n <xgmml>\n", graph.getName(s).str());
+                        Owned<IPropertyTree> xgmml = graph.getXGMMLTree(true, false);  // We can't merge between nodes if we format the values
+                        toXML(xgmml, reply, 2);
+                        reply.append(" </xgmml>\n</Graph>\n");
                     }
-                toXML(stats, reply);
+                }
+                reply.append("</Query>\n");
                 return true;
             }
         }
@@ -1727,11 +1753,11 @@ public:
         reply.append("</PackageSets>\n");
     }
 
-    void getStats(StringBuffer &reply, const char *id, const char *action, const char *graphName, const IRoxieContextLogger &logctx) const
+    void getStats(StringBuffer &reply, const char *id, const char *graphName, const char *wuid, const IRoxieContextLogger &logctx) const
     {
         ForEachItemIn(idx, allQueryPackages)
         {
-            if (allQueryPackages.item(idx).getStats(id, action, graphName, reply, logctx))
+            if (allQueryPackages.item(idx).getStats(id, graphName, reply, wuid, logctx))
                return;
         }
     }
@@ -2504,10 +2530,6 @@ private:
                 preabortKeyedJoinsThreshold = control->getPropInt("@val", 100);
                 topology->setPropInt("@preabortKeyedJoinsThreshold", preabortKeyedJoinsThreshold);
             }
-            else if (stricmp(queryName, "control:probeAllRows")==0)
-            {
-                probeAllRows = control->getPropBool("@val", true);
-            }
             else
                 unknown = true;
             break;
@@ -2578,6 +2600,7 @@ private:
                 if (!id)
                     badFormat();
                 const char *action = control->queryProp("Query/@action");
+                const char *wuid = control->queryProp("Query/@wuid");
                 const char *graphName = 0;
                 if (action)
                 {
@@ -2604,7 +2627,7 @@ private:
                         throw MakeStringException(ROXIE_CONTROL_MSG_ERROR, "invalid action in control:queryStats %s", action);
                 }
                 ReadLockBlock readBlock(packageCrit);
-                allQueryPackages->getStats(reply, id, action, graphName, logctx);
+                allQueryPackages->getStats(reply, id, graphName, wuid, logctx);
             }
             else if (stricmp(queryName, "control:queryWuid")==0)
             {
@@ -2960,27 +2983,24 @@ void mergeNodes(IPropertyTree *s1, IPropertyTree *s2)
         {
             StringBuffer xpath;
             xpath.appendf("att[@name='%s']", name);
-            const char *type = e1.queryProp("@type");
-            if (type)
+            if (startsWith(name, "SizeMax"))
             {
                 IPropertyTree *e2 = s2->queryPropTree(xpath.str());
                 if (e2)
                 {
                     unsigned __int64 v2 = e2->getPropInt64("@value", 0);
-                    if (strcmp(name, "max")==0)
-                    {
-                        if (v2 > v1)
-                            e1.setPropInt64("@value", v2);
-                    }
-                    else if (strcmp(type, "min")==0)
-                    {
-                        if (v2 < v1)
-                            e1.setPropInt64("@value", v2);
-                    }
-                    else if (strcmp(type, "sum")==0)
-                        e1.setPropInt64("@value", v1+v2);
-                    else
-                        throw MakeStringException(ROXIE_UNKNOWN_QUERY, "Unknown type %s in graph statistics", type);
+                    if (v2 > v1)
+                        e1.setPropInt64("@value", v2);
+                    s2->removeTree(e2);
+                }
+            }
+            else if (startsWith(name, "Size") || startsWith(name, "Time") || startsWith(name, "Num"))
+            {
+                IPropertyTree *e2 = s2->queryPropTree(xpath.str());
+                if (e2)
+                {
+                    unsigned __int64 v2 = e2->getPropInt64("@value", 0);
+                    e1.setPropInt64("@value", v1+v2);
                     s2->removeTree(e2);
                 }
             }
@@ -3170,7 +3190,7 @@ static const char *g1 =
               " <att name='_kind' value='1'>"   // TAKsubgraph
               "  <graph>"
               "   <node id='7696' label='Nested'>"
-              "    <att name='seeks' value='15' type='sum'/>"
+              "    <att name='NumSeeks' value='15'/>"
               "   </node>"
               "  </graph>"
               " </att>"
@@ -3182,23 +3202,23 @@ static const char *g1 =
               "</node>"
               "<att name='rootGraph' value='1'/>"
               "<edge id='2_0' source='2' target='3'>"
-               "<att name='count' value='15' type='sum'/>"
+               "<att name='NumRows' value='15'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='3_0' source='3' target='5'>"
-               "<att name='count' value='15' type='sum'/>"
+               "<att name='NumRows' value='15'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='5_0' source='5' target='6'>"
-               "<att name='count' value='3' type='sum'/>"
+               "<att name='NumRows' value='3'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='5_1' source='5' target='7'>"
                "<att name='_sourceIndex' value='1'/>"
-               "<att name='count' value='15' type='sum'/>"
+               "<att name='NumRows' value='15'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
@@ -3228,7 +3248,7 @@ static const char *g2 =
               " <att name='_kind' value='1'>"   // TAKsubgraph
               "  <graph>"
               "   <node id='7696' label='Nested'>"
-              "    <att name='seeks' value='25' type='sum'/>"
+              "    <att name='NumSeeks' value='25'/>"
               "   </node>"
               "  </graph>"
               " </att>"
@@ -3240,17 +3260,17 @@ static const char *g2 =
               "</node>"
               "<att name='rootGraph' value='1'/>"
               "<edge id='2_0' source='2' target='3'>"
-               "<att name='count' value='15' type='sum'/>"
+               "<att name='NumRows' value='15'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='3_0' source='3' target='5'>"
-               "<att name='count' value='15' type='sum'/>"
+               "<att name='NumRows' value='15'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='5_0' source='5' target='6'>"
-               "<att name='count' value='3' type='sum'/>"
+               "<att name='NumRows' value='3'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
@@ -3280,7 +3300,7 @@ static const char *expected =
               " <att name='_kind' value='1'>"   // TAKsubgraph
               "  <graph>"
               "   <node id='7696' label='Nested'>"
-              "    <att name='seeks' type='sum' value='40'/>"
+              "    <att name='NumSeeks' value='40'/>"
               "   </node>"
               "  </graph>"
               " </att>"
@@ -3297,23 +3317,23 @@ static const char *expected =
               "</node>"
               "<att name='rootGraph' value='1'/>"
               "<edge id='2_0' source='2' target='3'>"
-               "<att name='count' value='30' type='sum'/>"
+               "<att name='NumRows' value='30'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='3_0' source='3' target='5'>"
-               "<att name='count' value='30' type='sum'/>"
+               "<att name='NumRows' value='30'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='5_0' source='5' target='6'>"
-               "<att name='count' value='6' type='sum'/>"
+               "<att name='NumRows' value='6'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"
               "<edge id='5_1' source='5' target='7'>"
                "<att name='_sourceIndex' value='1'/>"
-               "<att name='count' value='15' type='sum'/>"
+               "<att name='NumRows' value='15'/>"
                "<att name='started' value='1'/>"
                "<att name='stopped' value='1'/>"
               "</edge>"

roxie/ccd/ccdstate.hpp

@@ -124,7 +124,7 @@ interface IRoxieQuerySetManager : extends IInterface
     virtual bool isActive() const = 0;
     virtual IQueryFactory *getQuery(const char *id, StringBuffer *querySet, const IRoxieContextLogger &ctx) const = 0;
     virtual void load(const IPropertyTree *querySet, const IRoxiePackageMap &packages, hash64_t &hash, bool forceRetry) = 0;
-    virtual void getStats(const char *queryName, const char *graphName, StringBuffer &reply, const IRoxieContextLogger &logctx) const = 0;
+    virtual void getStats(const char *queryName, const char *graphName, IConstWorkUnit *statsWu, unsigned channel, bool reset, const IRoxieContextLogger &logctx) const = 0;
     virtual void resetQueryTimings(const char *queryName, const IRoxieContextLogger &logctx) = 0;
     virtual void resetAllQueryTimings() = 0;
     virtual void getActivityMetrics(StringBuffer &reply) const = 0;

roxie/ccd/hpccprotocol.hpp

@@ -71,7 +71,6 @@ interface IHpccProtocolResponse : extends IInterface
 
     virtual void appendContent(TextMarkupFormat mlFmt, const char *content, const char *name=NULL) = 0; //will be transformed
     virtual IXmlWriter *writeAppendContent(const char *name = NULL) = 0;
-    virtual void appendProbeGraph(const char *xml) = 0;
 
     virtual void finalize(unsigned seqNo) = 0;
 

system/jlib/jlog.cpp

@@ -3054,6 +3054,9 @@ public:
     virtual void noteStatistic(StatisticKind kind, unsigned __int64 value) const
     {
     }
+    virtual void setStatistic(StatisticKind kind, unsigned __int64 value) const
+    {
+    }
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const
     {
     }

system/jlib/jlog.hpp

@@ -1286,6 +1286,7 @@ interface jlib_decl IContextLogger : extends IInterface
     void logOperatorException(IException *E, const char *file, unsigned line, const char *format, ...) const  __attribute__((format(printf, 5, 6)));
     virtual void logOperatorExceptionVA(IException *E, const char *file, unsigned line, const char *format, va_list args) const __attribute__((format(printf,5,0))) = 0;
     virtual void noteStatistic(StatisticKind kind, unsigned __int64 value) const = 0;
+    virtual void setStatistic(StatisticKind kind, unsigned __int64 value) const = 0;
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const = 0;
     virtual unsigned queryTraceLevel() const = 0;
 

system/jlib/jstatcodes.h

@@ -27,6 +27,7 @@
 #define WorkflowScopePrefix "w"
 #define ChildGraphScopePrefix "c"
 #define FileScopePrefix "p"
+#define ChannelScopePrefix "x"
 
 #define MATCHES_CONST_PREFIX(search, prefix) (strncmp(search, prefix, strlen(prefix)) == 0)
 
@@ -71,6 +72,7 @@ enum StatisticScopeType
     SSTworkflow,
     SSTchildgraph,
     SSTfile,
+    SSTchannel,                         // a Roxie channel
     SSTunknown,
     SSTmax
 };

system/jlib/jstats.cpp

@@ -72,7 +72,7 @@ void setStatisticsComponentName(StatisticCreatorType processType, const char * p
 // Textual forms of the different enumerations, first items are for none and all.
 static constexpr const char * const measureNames[] = { "", "all", "ns", "ts", "cnt", "sz", "cpu", "skw", "node", "ppm", "ip", "cy", "en", "txt", "bool", "id", "fname", "cost", NULL };
 static constexpr const char * const creatorTypeNames[]= { "", "all", "unknown", "hthor", "roxie", "roxie:s", "thor", "thor:m", "thor:s", "eclcc", "esp", "summary", NULL };
-static constexpr const char * const scopeTypeNames[] = { "", "all", "global", "graph", "subgraph", "activity", "allocator", "section", "compile", "dfu", "edge", "function", "workflow", "child", "file", "unknown", nullptr };
+static constexpr const char * const scopeTypeNames[] = { "", "all", "global", "graph", "subgraph", "activity", "allocator", "section", "compile", "dfu", "edge", "function", "workflow", "child", "file", "channel", "unknown", nullptr };
 
 static unsigned matchString(const char * const * names, const char * search, unsigned dft)
 {
@@ -112,6 +112,7 @@ static const StatisticScopeType scoreOrder[] = {
     SSTworkflow,
     SSTchildgraph,
     SSTfile,
+    SSTchannel,  // MORE - not sure what this means!
     SSTunknown
 };
 static int scopePriority[SSTmax];
@@ -1354,6 +1355,10 @@ StringBuffer & StatsScopeId::getScopeText(StringBuffer & out) const
         return out.append(WorkflowScopePrefix).append(id);
     case SSTchildgraph:
         return out.append(ChildGraphScopePrefix).append(id);
+    case SSTfile:
+        return out.append(FileScopePrefix).append(name);
+    case SSTchannel:
+        return out.append(ChannelScopePrefix).append(id);
     case SSTunknown:
         return out.append(name);
     default:
@@ -1411,6 +1416,7 @@ void StatsScopeId::describe(StringBuffer & description) const
     case SSTactivity:
     case SSTworkflow:
     case SSTchildgraph:
+    case SSTchannel:
         description.append(' ').append(id);
         break;
     case SSTedge:
@@ -1457,12 +1463,14 @@ void StatsScopeId::deserialize(MemoryBuffer & in, unsigned version)
     case SSTactivity:
     case SSTworkflow:
     case SSTchildgraph:
+    case SSTchannel:
         in.read(id);
         break;
     case SSTedge:
         in.read(id);
         in.read(extra);
         break;
+    case SSTfile:
     case SSTfunction:
         in.read(name);
         break;
@@ -1482,6 +1490,7 @@ void StatsScopeId::serialize(MemoryBuffer & out) const
     case SSTactivity:
     case SSTworkflow:
     case SSTchildgraph:
+    case SSTchannel:
         out.append(id);
         break;
     case SSTedge:
@@ -1587,6 +1596,13 @@ bool StatsScopeId::setScopeText(const char * text, const char * * _next)
             return true;
         }
         break;
+    case ChannelScopePrefix[0]:
+        if (MATCHES_CONST_PREFIX(text, ChannelScopePrefix) && isdigit(text[strlen(ChannelScopePrefix)]))
+        {
+            setChannelId(strtoul(text+ strlen(ChannelScopePrefix), next, 10));
+            return true;
+        }
+        break;
     case '\0':
         setId(SSTglobal, 0);
         return true;
@@ -1640,6 +1656,10 @@ void StatsScopeId::setFileId(const char * _name)
     scopeType = SSTfile;
     name.set(_name);
 }
+void StatsScopeId::setChannelId(unsigned _id)
+{
+    setId(SSTchannel, _id);
+}
 void StatsScopeId::setWorkflowId(unsigned _id)
 {
     setId(SSTworkflow, _id);
@@ -1724,9 +1744,9 @@ public:
 
     virtual byte getCollectionType() const { return SCintermediate; }
 
-    StringBuffer &toXML(StringBuffer &out) const;
 
 //interface IStatisticCollection:
+    virtual StringBuffer &toXML(StringBuffer &out) const override;
     virtual StatisticScopeType queryScopeType() const override
     {
         return id.queryScopeType();
@@ -1894,6 +1914,7 @@ public:
 private:
     StatsScopeId id;
     CStatisticCollection * parent;
+protected:
     CollectionHashTable children;
     StatsArray stats;
 };
@@ -1994,6 +2015,15 @@ public:
         out.append(creator);
         out.append(whenCreated);
     }
+    virtual void mergeInto(IStatisticGatherer & target) const override
+    {
+        // Similar to CStatisticCollection::mergeInfo but do not add the root scope.
+        ForEachItemIn(iStat, stats)
+            stats.item(iStat).mergeInto(target);
+
+        for (auto const & cur : children)
+            cur.mergeInto(target);
+    }
 public:
     StatisticCreatorType creatorType;
     StringAttr creator;
@@ -2070,6 +2100,12 @@ public:
         CStatisticCollection & tos = scopes.tos();
         scopes.append(*tos.ensureSubScope(scopeId, true));
     }
+    virtual void beginChannelScope(unsigned id) override
+    {
+        StatsScopeId scopeId(SSTchannel, id);
+        CStatisticCollection & tos = scopes.tos();
+        scopes.append(*tos.ensureSubScope(scopeId, true));
+    }
     virtual void endScope() override
     {
         scopes.pop();
@@ -2138,6 +2174,7 @@ public:
     virtual void beginChildGraphScope(unsigned id) { throwUnexpected(); }
     virtual void beginActivityScope(unsigned id) { throwUnexpected(); }
     virtual void beginEdgeScope(unsigned id, unsigned oid) { throwUnexpected(); }
+    virtual void beginChannelScope(unsigned id) { throwUnexpected(); }
     virtual void endScope()
     {
         node = &stack.popGet();
@@ -2345,12 +2382,12 @@ void CRuntimeStatisticCollection::rollupStatistics(unsigned numTargets, IContext
     reportIgnoredStats();
 }
 
-void CRuntimeStatisticCollection::recordStatistics(IStatisticGatherer & target) const
+void CRuntimeStatisticCollection::recordStatistics(IStatisticGatherer & target, bool clear) const
 {
     ForEachItem(i)
     {
         StatisticKind kind = getKind(i);
-        unsigned __int64 value = values[i].get();
+        unsigned __int64 value = clear ? values[i].getClearAtomic() : values[i].get();
         if (value || includeStatisticIfZero(kind))
         {
             StatisticKind serialKind= querySerializedKind(kind);
@@ -2363,7 +2400,7 @@ void CRuntimeStatisticCollection::recordStatistics(IStatisticGatherer & target)
     reportIgnoredStats();
     CNestedRuntimeStatisticMap *qn = queryNested();
     if (qn)
-        qn->recordStatistics(target);
+        qn->recordStatistics(target, clear);
 }
 
 void CRuntimeStatisticCollection::reportIgnoredStats() const
@@ -2454,12 +2491,6 @@ void CRuntimeStatisticCollection::deserializeMerge(MemoryBuffer& in)
     }
 }
 
-void CRuntimeStatisticCollection::getNodeProgressInfo(IPropertyTree &node) const
-{
-    TreeNodeStatisticGatherer gatherer(node);
-    recordStatistics(gatherer);
-}
-
 bool CRuntimeStatisticCollection::serialize(MemoryBuffer& out) const
 {
     unsigned numValid = 0;
@@ -2616,7 +2647,7 @@ static bool isWorthReportingMergedValue(StatisticKind kind)
 }
 
 
-void CRuntimeSummaryStatisticCollection::recordStatistics(IStatisticGatherer & target) const
+void CRuntimeSummaryStatisticCollection::recordStatistics(IStatisticGatherer & target, bool clear) const
 {
     for (unsigned i = 0; i < ordinality(); i++)
     {
@@ -2628,7 +2659,7 @@ void CRuntimeSummaryStatisticCollection::recordStatistics(IStatisticGatherer & t
             //Thor should always publish the average value for a stat, and the merged value if it makes sense.
             //So that it is easy to analyse graphs independent of the number of slave nodes it is executed on.
 
-            unsigned __int64 mergedValue = convertMeasure(kind, serialKind, values[i].get());
+            unsigned __int64 mergedValue = convertMeasure(kind, serialKind, clear ? values[i].getClearAtomic() : values[i].get());
             if (isWorthReportingMergedValue(serialKind))
             {
                 if (mergedValue || includeStatisticIfZero(serialKind))
@@ -2697,7 +2728,7 @@ void CRuntimeSummaryStatisticCollection::recordStatistics(IStatisticGatherer & t
     reportIgnoredStats();
     CNestedRuntimeStatisticMap *qn = queryNested();
     if (qn)
-        qn->recordStatistics(target);
+        qn->recordStatistics(target, clear);
 }
 
 bool CRuntimeSummaryStatisticCollection::serialize(MemoryBuffer & out) const
@@ -2745,10 +2776,10 @@ bool CNestedRuntimeStatisticCollection::serialize(MemoryBuffer& out) const
     return stats->serialize(out);
 }
 
-void CNestedRuntimeStatisticCollection::recordStatistics(IStatisticGatherer & target) const
+void CNestedRuntimeStatisticCollection::recordStatistics(IStatisticGatherer & target, bool clear) const
 {
     target.beginScope(scope);
-    stats->recordStatistics(target);
+    stats->recordStatistics(target, clear);
     target.endScope();
 }
 
@@ -2871,11 +2902,11 @@ bool CNestedRuntimeStatisticMap::serialize(MemoryBuffer& out) const
     return nonEmpty;
 }
 
-void CNestedRuntimeStatisticMap::recordStatistics(IStatisticGatherer & target) const
+void CNestedRuntimeStatisticMap::recordStatistics(IStatisticGatherer & target, bool clear) const
 {
     ReadLockBlock b(lock);
     ForEachItemIn(i, map)
-        map.item(i).recordStatistics(target);
+        map.item(i).recordStatistics(target, clear);
 }
 
 StringBuffer & CNestedRuntimeStatisticMap::toStr(StringBuffer &str) const

system/jlib/jstats.h

@@ -83,6 +83,7 @@ public:
     void setEdgeId(unsigned _id, unsigned _output);
     void setFunctionId(const char * _name);
     void setFileId(const char * _name);
+    void setChannelId(unsigned id);
     void setSubgraphId(unsigned _id);
     void setWorkflowId(unsigned _id);
     void setChildGraphId(unsigned _id);
@@ -117,6 +118,7 @@ public:
     virtual void serialize(MemoryBuffer & out) const = 0;
     virtual unsigned __int64 queryWhenCreated() const = 0;
     virtual void mergeInto(IStatisticGatherer & target) const = 0;
+    virtual StringBuffer &toXML(StringBuffer &out) const = 0;
 };
 
 interface IStatisticCollectionIterator : public IIteratorOf<IStatisticCollection>
@@ -141,6 +143,7 @@ public:
     virtual void beginActivityScope(unsigned id) = 0;
     virtual void beginEdgeScope(unsigned id, unsigned oid) = 0;
     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 updateStatistic(StatisticKind kind, unsigned __int64 value, StatsMergeAction mergeAction) = 0;
@@ -211,6 +214,15 @@ public:
     }
 };
 
+class ChannelActivityScope : public StatsScopeBlock
+{
+public:
+    inline ChannelActivityScope(IStatisticGatherer & _gatherer, unsigned id) : StatsScopeBlock(_gatherer)
+    {
+        gatherer.beginChannelScope(id);
+    }
+};
+
 class StatsEdgeScope : public StatsScopeBlock
 {
 public:
@@ -583,9 +595,7 @@ public:
     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;
+    virtual void recordStatistics(IStatisticGatherer & target, bool clear) const;
 
     // Print out collected stats to string
     StringBuffer &toStr(StringBuffer &str) const;
@@ -626,7 +636,7 @@ public:
     CRuntimeSummaryStatisticCollection(const StatisticsMapping & _mapping);
     ~CRuntimeSummaryStatisticCollection();
 
-    virtual void recordStatistics(IStatisticGatherer & target) const override;
+    virtual void recordStatistics(IStatisticGatherer & target, bool clear = false) const override;
     virtual bool serialize(MemoryBuffer & out) const override;  // Returns true if any non-zero
     virtual void deserialize(MemoryBuffer & in) override;
     virtual void deserializeMerge(MemoryBuffer& in) override;
@@ -674,7 +684,7 @@ public:
     void deserialize(MemoryBuffer & in);
     void deserializeMerge(MemoryBuffer& in);
     void merge(const CNestedRuntimeStatisticCollection & other, unsigned node);
-    void recordStatistics(IStatisticGatherer & target) const;
+    void recordStatistics(IStatisticGatherer & target, bool clear) const;
     StringBuffer & toStr(StringBuffer &str) const;
     StringBuffer & toXML(StringBuffer &str) const;
     void updateDelta(CNestedRuntimeStatisticCollection & target, const CNestedRuntimeStatisticCollection & source);
@@ -695,7 +705,7 @@ public:
     void deserialize(MemoryBuffer & in);
     void deserializeMerge(MemoryBuffer& in);
     void merge(const CNestedRuntimeStatisticMap & other, unsigned node);
-    void recordStatistics(IStatisticGatherer & target) const;
+    void recordStatistics(IStatisticGatherer & target, bool clear) const;
     StringBuffer & toStr(StringBuffer &str) const;
     StringBuffer & toXML(StringBuffer &str) const;
     void updateDelta(CNestedRuntimeStatisticMap & target, const CNestedRuntimeStatisticMap & source);

thorlcr/graph/thgraph.cpp

@@ -2658,6 +2658,9 @@ public:
     virtual void noteStatistic(StatisticKind kind, unsigned __int64 value) const
     {
     }
+    virtual void setStatistic(StatisticKind kind, unsigned __int64 value) const
+    {
+    }
     virtual void mergeStats(const CRuntimeStatisticCollection &from) const
     {
     }
@@ -2732,7 +2735,7 @@ CJobBase::CJobBase(ILoadedDllEntry *_querySo, const char *_graphName) : querySo(
         throw MakeStringException(0, "Failed to locate workunit info in query : %s", querySo->queryName());
     Owned<ILocalWorkUnit> localWU = createLocalWorkUnit(wuXML);
     Owned<IConstWUGraph> graph = localWU->getGraph(graphName);
-    graphXGMML.setown(graph->getXGMMLTree(false));
+    graphXGMML.setown(graph->getXGMMLTree(false, false));
     if (!graphXGMML)
         throwUnexpected();
 }

thorlcr/master/thdemonserver.cpp

@@ -140,7 +140,7 @@ private:
                 ForEachItemIn (g, activeGraphs)
                 {
                     CGraphBase &graph = activeGraphs.item(g);
-                    Owned<IWUGraphStats> stats = currentWU.updateStats(graphName, SCTthor, queryStatisticsComponentName(), wfid, graph.queryGraphId());
+                    Owned<IWUGraphStats> stats = currentWU.updateStats(graphName, SCTthor, queryStatisticsComponentName(), wfid, graph.queryGraphId(), false);
                     reportGraph(stats->queryStatsBuilder(), &graph);
                 }
                 Owned<IWorkUnit> wu = &currentWU.lock();
@@ -168,7 +168,7 @@ private:
             const char *graphName = ((CJobMaster &)activeGraphs.item(0).queryJob()).queryGraphName();
             unsigned wfid = graph->queryJob().getWfid();
             {
-                Owned<IWUGraphStats> stats = currentWU.updateStats(graphName, SCTthor, queryStatisticsComponentName(), wfid, graph->queryGraphId());
+                Owned<IWUGraphStats> stats = currentWU.updateStats(graphName, SCTthor, queryStatisticsComponentName(), wfid, graph->queryGraphId(), false);
                 reportGraph(stats->queryStatsBuilder(), graph);
             }
 

tools/wutool/wutool.cpp

@@ -306,10 +306,17 @@ bool looksLikeWuid(const char * arg)
     return false;
 }
 
+static constexpr const char * defaultYaml = R"!!(
+version: "1.0"
+wutool:
+    name: wutool
+)!!";
+
 int main(int argc, const char *argv[])
 {
     int ret = 0;
     InitModuleObjects();
+    Owned<IPropertyTree> dummyconfig = loadConfiguration(defaultYaml, argv, "wutool", "WUTOOL", "wutool.xml", nullptr, nullptr, false);
     unsigned count=0;
     globals.setown(createProperties("wutool.ini", true));
     const char *action = NULL;
@@ -1139,7 +1146,7 @@ protected:
         ASSERT(graph->getType() == GraphTypeActivities);
         ASSERT(streq(graph->getName(s).str(),"Graph1"));
         ASSERT(streq(graph->getLabel(s).str(),"graphLabel"));
-        ASSERT(streq(graph->getXGMML(s, false).str(), "<graph/>\n"));
+        ASSERT(streq(graph->getXGMML(s, false, false).str(), "<graph/>\n"));
 
         // Then the lightweight meta....
         wu.setown(factory->openWorkUnit(wuid));
@@ -1174,7 +1181,7 @@ protected:
         {
             ASSERT(it2->query().getType() == GraphTypeActivities);
             ASSERT(streq(it2->query().getLabel(s).str(),"graphLabel"));
-            ASSERT(streq(it2->query().getXGMML(s, false).str(), "<graph/>\n"));
+            ASSERT(streq(it2->query().getXGMML(s, false, false).str(), "<graph/>\n"));
             numIterated++;
         }
         ASSERT(numIterated==2);
@@ -1186,7 +1193,7 @@ protected:
         ForEach (*it2)
         {
             ASSERT(streq(it2->query().getLabel(s).str(),"graphLabel"));
-            ASSERT(streq(it2->query().getXGMML(s, false).str(), "<graph/>\n"));
+            ASSERT(streq(it2->query().getXGMML(s, false, false).str(), "<graph/>\n"));
             numIterated++;
         }
         ASSERT(numIterated==3);
@@ -1210,7 +1217,7 @@ protected:
         {
             ASSERT(it2->query().getType() == GraphTypeActivities);
             ASSERT(streq(it2->query().getLabel(s).str(),"graphLabel"));
-            ASSERT(streq(it2->query().getXGMML(s, false).str(), "<graph/>\n"));
+            ASSERT(streq(it2->query().getXGMML(s, false, false).str(), "<graph/>\n"));
             numIterated++;
         }
         ASSERT(numIterated==2);
@@ -1224,7 +1231,7 @@ protected:
         {
             ASSERT(it2->query().getType() == GraphTypeActivities);
             ASSERT(streq(it2->query().getLabel(s).str(),"graphLabel"));
-            ASSERT(streq(it2->query().getXGMML(s, false).str(), "<graph/>\n"));
+            ASSERT(streq(it2->query().getXGMML(s, false, false).str(), "<graph/>\n"));
             numIterated++;
         }
         ASSERT(numIterated==2);
@@ -1276,7 +1283,7 @@ protected:
         ret = wu->getRunningGraph(s, subid);
         ASSERT(!ret);
 
-        Owned<IWUGraphStats> progress = wu->updateStats("graph1", SCThthor, queryStatisticsComponentName(), 1, 1);
+        Owned<IWUGraphStats> progress = wu->updateStats("graph1", SCThthor, queryStatisticsComponentName(), 1, 1, false);
         IStatisticGatherer & stats = progress->queryStatsBuilder();
         {
             StatsSubgraphScope subgraph(stats, 1);