From 8fe8180c20f1b38f75c1afb33d8dcb24e986b867 Mon Sep 17 00:00:00 2001
From: Shamser Ahmed <shamser.ahmed@lexisnexis.com>
Date: Wed, 6 Dec 2023 16:35:30 +0000
Subject: [PATCH] HPCC-30590 Report warning on any activity that has a large
 skew in execution worktime

Signed-off-by: Shamser Ahmed <shamser.ahmed@lexisnexis.com>
---
 common/wuanalysis/anaerrorcodes.hpp |  2 +-
 common/wuanalysis/anarule.cpp       | 50 +++++++++++++++++++++++++++++
 2 files changed, 51 insertions(+), 1 deletion(-)

diff --git a/common/wuanalysis/anaerrorcodes.hpp b/common/wuanalysis/anaerrorcodes.hpp
index e2503c27759..37198fb970a 100644
--- a/common/wuanalysis/anaerrorcodes.hpp
+++ b/common/wuanalysis/anaerrorcodes.hpp
@@ -9,7 +9,7 @@ typedef enum
     ANA_DISTRIB_SKEW_INPUT_ID,
     ANA_DISTRIB_SKEW_OUTPUT_ID,
     ANA_IOSKEW_RECORDS_ID,
-    ANA_UNUSED_ID,                                 /* May re-use but don't remove to avoid changing later id's */
+    ANA_EXECUTE_SKEW_ID,
     ANA_KJ_EXCESS_PREFILTER_ID
 } AnalyzerErrorCode;
 
diff --git a/common/wuanalysis/anarule.cpp b/common/wuanalysis/anarule.cpp
index 0191d17271c..b33d572da3e 100644
--- a/common/wuanalysis/anarule.cpp
+++ b/common/wuanalysis/anarule.cpp
@@ -183,6 +183,55 @@ class IoSkewRule : public AActivityRule
     const char * category;
 };
 
+class IoSkewLocalExecuteRule : public AActivityRule
+{
+public:
+    virtual bool isCandidate(IWuActivity & activity) const override
+    {
+        switch (activity.getAttr(WaKind))
+        {
+            case TAKfirstn:
+            case TAKtopn:
+            case TAKsort:
+                return false;
+        }
+        return true;
+    }
+
+    virtual bool check(PerformanceIssue & result, IWuActivity & activity, const IAnalyserOptions & options) override
+    {
+        stat_type localExecuteMaxSkew = activity.getStatRaw(StTimeLocalExecute, StSkewMax);
+        if (localExecuteMaxSkew<options.queryOption(watOptSkewThreshold))
+            return false;
+
+        stat_type timeMaxLocalExecute = activity.getStatRaw(StTimeLocalExecute, StMaxX);
+        stat_type timeAvgLocalExecute = activity.getStatRaw(StTimeLocalExecute, StAvgX);
+        stat_type cost = (timeMaxLocalExecute - timeAvgLocalExecute);;
+
+        bool inputSkewed = false;
+        for(unsigned edgeNo = 0; IWuEdge *wuInputEdge = activity.queryInput(edgeNo); edgeNo++)
+        {
+            if (wuInputEdge->getStatRaw(StNumRowsProcessed, StSkewMax)>options.queryOption(watOptSkewThreshold))
+            {
+                inputSkewed = true;
+                break;
+            }
+        }
+        bool outputSkewed = false;
+        IWuEdge *wuOutputEdge = activity.queryOutput(0);
+        if (wuOutputEdge && (wuOutputEdge->getStatRaw(StNumRowsProcessed, StSkewMax)>options.queryOption(watOptSkewThreshold)))
+            outputSkewed = true;
+
+        if (inputSkewed)
+            result.set(ANA_EXECUTE_SKEW_ID, cost, "Significant skew in local execute time caused by uneven input");
+        else if (outputSkewed)
+            result.set(ANA_EXECUTE_SKEW_ID, cost, "Significant skew in local execute time caused by uneven output");
+        else
+            result.set(ANA_EXECUTE_SKEW_ID, cost, "Significant skew in local execute time");
+        return true;
+    }
+};
+
 class KeyedJoinExcessRejectedRowsRule : public ActivityKindRule
 {
 public:
@@ -221,4 +270,5 @@ void gatherRules(CIArrayOf<AActivityRule> & rules)
     rules.append(*new IoSkewRule(StTimeDiskWriteIO, "disk write"));
     rules.append(*new IoSkewRule(StTimeSpillElapsed, "spill"));
     rules.append(*new KeyedJoinExcessRejectedRowsRule);
+    rules.append(*new IoSkewLocalExecuteRule);
 }