Merge remote-tracking branch 'origin/candidate-9.6.x' into candidate-…

…9.8.x Signed-off-by: Gordon Smith <[email protected]> # Conflicts: # helm/hpcc/Chart.yaml # helm/hpcc/templates/_helpers.tpl # version.cmake
stuartort · Aug 30, 2024 · f9f260a · f9f260a
2 parents b39ce89 + f495df6
commit f9f260a
Show file tree

Hide file tree

Showing 16 changed files with 516 additions and 130 deletions.
diff --git a/.github/workflows/test-helm.yml b/.github/workflows/test-helm.yml
@@ -1,16 +1,6 @@
 name: Run helm chart tests
 
 on:
-  push:
-    branches:
-      - "master"
-      - "candidate-*"
-      - "!candidate-7.8.*"
-      - "!candidate-7.6.*"
-      - "!candidate-7.4.*"
-      - "!candidate-7.2.*"
-      - "!candidate-7.0.*"
-      - "!candidate-6.*"
   pull_request:
     branches:
       - "master"
@@ -57,8 +47,45 @@ jobs:
       - name: Check for changes in helm output for default values
         working-directory: .
         run: |
-          helm template helm/hpcc > ${{ runner.temp }}/newoutput.txt
-          git fetch --no-tags --prune --progress --no-recurse-submodules --depth=1 origin ${{ github.base_ref }}
+          resultcode=0
+          mkdir ${{ runner.temp }}/new
+          mkdir ${{ runner.temp }}/old
+
+          #Generate the output for the default values file, and each of the specialised test files.
+          #Check to see if anything has changed between old and new, and report the differences
+          helm template helm/hpcc > ${{ runner.temp }}/new/output.txt
+          for file in testing/helm/tests/*.yaml
+          do
+            tail=$(basename $file)
+            helm template helm/hpcc --values $file > ${{ runner.temp }}/new/$tail.txt
+          done
+
+          git fetch --no-tags --prune --progress --no-recurse-submodules --quiet --depth=1 origin ${{ github.base_ref }}
+          echo git checkout ${{ github.base_ref }}
           git checkout ${{ github.base_ref }}
-          helm template helm/hpcc > ${{ runner.temp }}/oldoutput.txt
-          diff ${{ runner.temp }}/oldoutput.txt ${{ runner.temp }}/newoutput.txt
+
+          helm template helm/hpcc > ${{ runner.temp }}/old/output.txt
+          for file in testing/helm/tests/*.yaml
+          do
+            tail=$(basename $file)
+            helm template helm/hpcc --values $file > ${{ runner.temp }}/old/$tail.txt
+          done
+
+          diff ${{ runner.temp }}/old/output.txt ${{ runner.temp }}/new/output.txt
+          if [ $? -ne 0 ]
+          then
+            resultcode=1
+          else
+            #Only check for differences in the specialised test files if the default values file is the same
+            for file in testing/helm/tests/*.yaml
+            do
+              tail=$(basename $file)
+              diff ${{ runner.temp }}/old/$tail.txt ${{ runner.temp }}/new/$tail.txt
+              if [ $? -ne 0 ]
+              then
+                resultcode=1
+              fi
+            done
+          fi
+
+          exit $resultcode
diff --git a/common/workunit/workunit.cpp b/common/workunit/workunit.cpp
@@ -6431,7 +6431,7 @@ class CDaliWorkUnitFactory : public CWorkUnitFactory, implements IDaliClientShut
             return WUStateUnknown;
         }
 
-        Owned<WorkUnitWaiter> waiter = new WorkUnitWaiter(wuid, SubscribeOptionState);
+        Owned<WorkUnitWaiter> waiter = new WorkUnitWaiter(wuid, (WUSubscribeOptions)(SubscribeOptionState|SubscribeOptionAbort));
         LocalIAbortHandler abortHandler(*waiter);
         if (conn)
         {
@@ -6505,7 +6505,7 @@ class CDaliWorkUnitFactory : public CWorkUnitFactory, implements IDaliClientShut
                     waiter->wait(20000);  // recheck state every 20 seconds, in case eclagent has crashed.
                     if (waiter->isAborted())
                     {
-                        ret = WUStateUnknown;  // MORE - throw an exception?
+                        ret = WUStateAborting;
                         break;
                     }
                 }
@@ -14483,13 +14483,13 @@ void executeThorGraph(const char * graphName, IConstWorkUnit &workunit, const IP
         unsigned runningTimeLimit = workunit.getDebugValueInt("maxRunTime", 0);
         runningTimeLimit = runningTimeLimit ? runningTimeLimit : INFINITE;
 
-        std::list<WUState> expectedStates = { WUStateRunning, WUStateWait, WUStateFailed };
+        std::list<WUState> expectedStates = { WUStateRunning, WUStateWait, WUStateAborting, WUStateFailed };
         unsigned __int64 blockedTime = 0;
         for (unsigned i=0; i<2; i++)
         {
             WUState state = waitForWorkUnitToComplete(wuid, timelimit*1000, expectedStates);
             DBGLOG("Got state: %s", getWorkunitStateStr(state));
-            if ((WUStateWait == state) || (WUStateFailed == state)) // already finished or failed
+            if ((WUStateWait == state) || (WUStateFailed == state) || (WUStateAborting == state)) // already finished or failed or aborting
             {
                 // workunit may have spent time in blocked state, but then transitioned to
                 // wait or failed state quickly such that this code did not see its running state.

diff --git a/common/workunit/workunit.ipp b/common/workunit/workunit.ipp
@@ -699,20 +699,33 @@ class WorkUnitWaiter : public CInterface, implements IAbortHandler, implements I
 {
     Semaphore changed;
     Owned<IWorkUnitWatcher> watcher;
-    bool aborted;
+    mutable bool aborted = false;
+    mutable bool abortDirty = false;
+    StringAttr wuid;
 public:
     IMPLEMENT_IINTERFACE;
-    WorkUnitWaiter(const char *wuid, WUSubscribeOptions watchFor)
+    WorkUnitWaiter(const char *_wuid, WUSubscribeOptions watchFor) : wuid(_wuid)
     {
         Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
         watcher.setown(factory->getWatcher(this, watchFor, wuid));
-        aborted = false;
+        if (watchFor & SubscribeOptionAbort)
+            abortDirty = true;
     }
     ~WorkUnitWaiter()
     {
         unsubscribe();
     }
-    bool isAborted() const { return aborted; }
+    bool isAborted() const
+    {
+        if (!aborted && abortDirty)
+        {
+            Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
+            if (factory->isAborting(wuid))
+                aborted = true;
+            abortDirty = false;
+        }
+        return aborted;
+    }
     bool wait(unsigned timeout)
     {
         return changed.wait(timeout) && !aborted;
@@ -728,6 +741,8 @@ public:
 // IWorkUnitSubscriber
     virtual void notify(WUSubscribeOptions flags, unsigned valueLen, const void *valueData) override
     {
+        if (SubscribeOptionAbort == flags)
+             abortDirty = true;
         changed.signal();
     }
 // IAbortHandler

diff --git a/dali/base/dacsds.cpp b/dali/base/dacsds.cpp
@@ -2239,6 +2239,7 @@ bool CClientSDSManager::updateEnvironment(IPropertyTree *newEnv, bool forceGroup
 
 //////////////
 
+static bool releaseActiveManager = false;
 static ISDSManager * activeSDSManager=NULL;
 static ISDSManager * savedSDSManager=NULL;
 
@@ -2248,10 +2249,13 @@ MODULE_INIT(INIT_PRIORITY_STANDARD)
 }
 MODULE_EXIT()
 {
-    delete activeSDSManager;
-    activeSDSManager = nullptr;
-    delete savedSDSManager;
-    savedSDSManager = nullptr;
+    if (releaseActiveManager)
+    {
+        delete activeSDSManager;
+        activeSDSManager = nullptr;
+        delete savedSDSManager;
+        savedSDSManager = nullptr;
+    }
 }
 
 ISDSManager &querySDS()
@@ -2261,9 +2265,8 @@ ISDSManager &querySDS()
         return *activeSDSManager;
     else if (!queryCoven().inCoven())
     {
-        if (!activeSDSManager)
-            activeSDSManager = new CClientSDSManager();
-
+        releaseActiveManager = true;
+        activeSDSManager = new CClientSDSManager();
         return *activeSDSManager;
     }
     else

diff --git a/docs/EN_US/HPCCClientTools/CT_Mods/CT_ECL_CLI.xml b/docs/EN_US/HPCCClientTools/CT_Mods/CT_ECL_CLI.xml
@@ -1006,10 +1006,11 @@ ecl publish roxie ArchiveQuery.xml --query-name=FindPersonService --no-activate
                 </row>
 
                 <row>
-                  <entry>--wait=&lt;sec&gt;</entry>
+                  <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>Maximum time to wait for cluster finish
-                  updating</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -1594,8 +1595,9 @@ ecl run thor findperson.ecl -I C:\MyECL\
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>Maximum time to wait for cluster finish updating (in
-                  ms)</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -2803,8 +2805,9 @@ ecl queries copy //192.168.1.10:8010/thor/findperson thor
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>Maximum time to wait for cluster finish updating (in
-                  ms)</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -3363,10 +3366,11 @@ ecl queries copy-set roxie1 roxie2 --clone-active-state</programlisting>
                 </row>
 
                 <row>
-                  <entry>--wait=&lt;sec&gt;</entry>
+                  <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>Maximum time to wait for cluster finish updating (in
-                  ms)</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -3704,10 +3708,11 @@ ecl queries recreate roxie findpeople roxie2</programlisting>
                 </row>
 
                 <row>
-                  <entry>--wait=&lt;sec&gt;</entry>
+                  <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>Maximum time to wait for cluster finish
-                  updating</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -6549,7 +6554,9 @@ ecl packagemap copy //192.168.0.100:8010/roxie/MyPkg roxie2
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>The maximum time to wait in milliseconds</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -6680,7 +6687,9 @@ ecl packagemap copy //192.168.0.100:8010/roxie/MyPkg roxie2
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>The maximum time to wait in milliseconds</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -6811,7 +6820,9 @@ ecl packagemap copy //192.168.0.100:8010/roxie/MyPkg roxie2
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>The maximum time to wait in milliseconds</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -6942,7 +6953,9 @@ ecl packagemap copy //192.168.0.100:8010/roxie/MyPkg roxie2
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>The maximum time to wait in milliseconds</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -7116,7 +7129,9 @@ ecl roxie xref myroxie --queryids=myquery.1,myotherquery.1</programlisting>
                 <row>
                   <entry>--wait=&lt;ms&gt;</entry>
 
-                  <entry>The maximum time to wait in milliseconds</entry>
+                  <entry>Maximum time (in milliseconds) to wait for cluster
+                  finish updating. Any value lower than the minimum of 1000 is
+                  ignored. </entry>
                 </row>
 
                 <row>
@@ -8652,15 +8667,15 @@ ecl zapgen W20171018-091399 --path ~ --inc-thor-slave-logs --description "Unexpe
             </tgroup>
           </informaltable>The <emphasis>ecl url-secret-name </emphasis>command
         generates a secret name from a URL that can be used to support ECL
-        SOAPCALL/HTTPCALL automated URL to Secret mapping. </para>
+        SOAPCALL/HTTPCALL automated URL to Secret mapping.</para>
 
         <para>A username can either be embedded in the URL, such as
         https://[email protected], or passed in as a parameter using the
         --username=username option. If a username is passed in as a parameter
-        it overrides a username in the URL. </para>
+        it overrides a username in the URL.</para>
 
         <para>Passwords embedded in the URL are not needed and will be
-        ignored. </para>
+        ignored.</para>
 
         <para>When ECL SOAPCALL URL secret mapping is enabled SOAPCALL will
         convert the URL provided into a name of this format. ECL will then

diff --git a/ecl/agentexec/agentexec.cpp b/ecl/agentexec/agentexec.cpp
@@ -295,9 +295,24 @@ class WaitThread : public CInterfaceOf<IPooledThread>
 
                     {
                         Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
-                        Owned<IWorkUnit> workunit = factory->updateWorkUnit(wuid);
-                        if (isContainerized())
-                            workunit->setContainerizedProcessInfo("AgentExec", compConfig->queryProp("@name"), k8s::queryMyPodName(), k8s::queryMyContainerName(), graphName, nullptr);
+                        Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid);
+                        if (!cw)
+                        {
+                            WARNLOG("Queued wuid does not exist: %s", wuid.str());
+                            return; // exit pooled thread
+                        }
+                        if (isThorAgent)
+                        {
+                            SessionId agentSessionID = cw->getAgentSession();
+                            if ((agentSessionID <= 0) || querySessionManager().sessionStopped(agentSessionID, 0))
+                            {
+                                WARNLOG("Discarding agentless queued Thor job: %s", wuid.str());
+                                return; // exit pooled thread
+                            }
+                        }
+
+                        Owned<IWorkUnit> workunit = &cw->lock();
+                        workunit->setContainerizedProcessInfo("AgentExec", compConfig->queryProp("@name"), k8s::queryMyPodName(), k8s::queryMyContainerName(), graphName, nullptr);
                         addTimeStamp(workunit, wfid, graphName, StWhenK8sLaunched);
                     }
                     k8s::runJob(jobSpecName, wuid, jobName, params);

diff --git a/ecl/hql/hqlopt.cpp b/ecl/hql/hqlopt.cpp
@@ -4269,9 +4269,24 @@ IHqlExpression * CTreeOptimizer::doCreateTransformed(IHqlExpression * transforme
 
                         if (ok)
                         {
-                            //If expanding the project removed all references to left (very silly join....) make it an all join
-                            if (transformed->hasAttribute(lookupAtom) && !exprReferencesDataset(&args.item(2), newLeft))
+                            //If expanding the project removed all references to left (very silly join....)
+                            //then it can be converted to an ALL join against a filtered right dataset
+                            IHqlExpression * transformedCond = &args.item(2);
+                            if (transformed->hasAttribute(lookupAtom) && !exprReferencesDataset(transformedCond, newLeft))
+                            {
+                                IHqlExpression * rightDs = &args.item(1);
+                                OwnedHqlExpr right = createSelector(no_right, rightDs, transformedSeq);
+                                if (exprReferencesDataset(transformedCond, right))
+                                {
+                                    OwnedHqlExpr newCond = replaceSelector(transformedCond, right, rightDs);
+                                    OwnedHqlExpr filteredRightDs = createDataset(no_filter, { LINK(rightDs), LINK(newCond) });
+                                    //Replace the right dataset with the filtered dataset, and the join condition is now unconditional
+                                    args.replace(*filteredRightDs.getClear(), 1);
+                                    args.replace(*createConstant(true), 2);
+                                }
+
                                 args.append(*createAttribute(allAtom));
+                            }
                             if (doTrace(traceOptimizations))
                                 DBGLOG("Optimizer: Merge %s and %s", queryNode0Text(transformed), queryNode1Text(child));
                             noteUnused(child);
@@ -4287,6 +4302,10 @@ IHqlExpression * CTreeOptimizer::doCreateTransformed(IHqlExpression * transforme
 
                             if (merged)
                                 return merged.getClear();
+
+                            //The combined join is not going to replace the join(project), so the noteUnused() above
+                            //needs to be reversed so that the usage counts stay correct
+                            incUsage(child);
                         }
                         break;
                     }