Merge branch 'candidate-7.2.x' into candidate-7.4.0

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

common/environment/environment.cpp

@@ -347,19 +347,31 @@ public:
     {
         synchronized procedure(safeCache);
         ensureClusterGroupKeyMap();
-        return keyGroupMap[group].c_str();
+        auto it = keyGroupMap.find(group);
+        if (it == keyGroupMap.end())
+            return nullptr;
+        else
+            return it->second.c_str();
     }
     virtual const char *getPublicKeyPath(const char *keyPairName) const override
     {
         synchronized procedure(safeCache);
         ensureClusterGroupKeyMap();
-        return keyPairMap[keyPairName].publicKey.c_str();
+        auto it = keyPairMap.find(keyPairName);
+        if (it == keyPairMap.end())
+            return nullptr;
+        else
+            return it->second.publicKey.c_str();
     }
     virtual const char *getPrivateKeyPath(const char *keyPairName) const override
     {
         synchronized procedure(safeCache);
         ensureClusterGroupKeyMap();
-        return keyPairMap[keyPairName].privateKey.c_str();
+        auto it = keyPairMap.find(keyPairName);
+        if (it == keyPairMap.end())
+            return nullptr;
+        else
+            return it->second.privateKey.c_str();
     }
     virtual const char *getFileAccessUrl() const
     {

esp/services/ws_dfu/ws_dfuService.cpp

@@ -6084,14 +6084,17 @@ void CWsDfuEx::dFUFileAccessCommon(IEspContext &context, const CDfsLogicalFileNa
         else
         {
             const char *kindStr = queryFileKind(fileDesc);
-            if (streq("flat", kindStr))
-                kind = CDFUFileType_Flat;
-            else if (streq("csv", kindStr))
-                kind = CDFUFileType_Csv;
-            else if (streq("xml", kindStr))
-                kind = CDFUFileType_Xml;
-            else if (streq("json", kindStr))
-                kind = CDFUFileType_Json;
+            if (!isEmptyString(kindStr))
+            {
+                if (streq("flat", kindStr))
+                    kind = CDFUFileType_Flat;
+                else if (streq("csv", kindStr))
+                    kind = CDFUFileType_Csv;
+                else if (streq("xml", kindStr))
+                    kind = CDFUFileType_Xml;
+                else if (streq("json", kindStr))
+                    kind = CDFUFileType_Json;
+            }
         }
     }
     resp.setType(kind);
@@ -6401,7 +6404,7 @@ bool CWsDfuEx::onDFUFileCreateV2(IEspContext &context, IEspDFUFileCreateV2Reques
             case CDFUFileType_Index:
                 fileDesc->queryProperties().setProp("@kind", "key");
             default:
-                break; // unknown
+                throw makeStringExceptionV(ECLWATCH_MISSING_FILETYPE, "DFUFileCreateV2: File type not provided");
         }
 
         MemoryBuffer layoutBin;

esp/smc/SMCLib/eclwatch_errorlist.hpp

@@ -126,6 +126,7 @@
 #define ECLWATCH_VIEW_ACCESS_DENIED         ECLWATCH_ERROR_START+105
 #define ECLWATCH_CANNOT_COPY_DLL            ECLWATCH_ERROR_START+106
 #define ECLWATCH_INVALID_ECLRECDEF          ECLWATCH_ERROR_START+107
+#define ECLWATCH_MISSING_FILETYPE           ECLWATCH_ERROR_START+108
 
 
 #endif //_ECLWATCH_ERRORLIST_HPP__

plugins/workunitservices/workunitservices.cpp

@@ -534,8 +534,10 @@ public:
     {
         const char * curScope = cur.queryScope();
         const char * kindName = queryStatisticName(kind);
-        assertex(kindName && memicmp(kindName, "when", 4) == 0);
-        kindName += 4;
+        assertex(kindName);
+        ///The following will be true on workunits >= 7.0, but may not be for 6.4 and earlier
+        if (memicmp(kindName, "when", 4) == 0)
+            kindName += 4;
 
         StringBuffer formattedTime;
         convertTimestampToStr(value, formattedTime, true);

testing/regress/ecl/apply3.ecl

@@ -32,8 +32,13 @@ namesTable := dataset([
 
 f := namesTable.ids(id & 1 = 1);
 
-apply(namesTable,
-    output(PROJECT(f, TRANSFORM(o, SELF.id := LEFT.id; SELF.forename := namesTable.forename)),named('Result'),EXTEND),
-    output(PROJECT(f, TRANSFORM(o, SELF.id := LEFT.id * 3; SELF.forename := namesTable.forename)),named('Result'),EXTEND)
-    );
+ORDERED(
+ apply(namesTable,
+       output(PROJECT(f, TRANSFORM(o, SELF.id := LEFT.id; SELF.forename := namesTable.forename)),named('Result'),EXTEND)
+      );
+
+ apply(namesTable,
+       output(PROJECT(f, TRANSFORM(o, SELF.id := LEFT.id * 3; SELF.forename := namesTable.forename)),named('Result'),EXTEND)
+      );
+);
 

testing/regress/ecl/key/apply3.xml

@@ -1,10 +1,10 @@
 <Dataset name='Result'>
  <Row><forename>Gavin     </forename><id>1</id></Row>
  <Row><forename>Gavin     </forename><id>3</id></Row>
+ <Row><forename>Mia       </forename><id>5</id></Row>
+ <Row><forename>Pru       </forename><id>11</id></Row>
  <Row><forename>Gavin     </forename><id>3</id></Row>
  <Row><forename>Gavin     </forename><id>9</id></Row>
- <Row><forename>Mia       </forename><id>5</id></Row>
  <Row><forename>Mia       </forename><id>15</id></Row>
- <Row><forename>Pru       </forename><id>11</id></Row>
  <Row><forename>Pru       </forename><id>33</id></Row>
 </Dataset>

testing/regress/ecl/key/wuidresultincq.xml

@@ -0,0 +1,12 @@
+<Dataset name='cqresult'>
+ <Row><num>0</num><total>25</total></Row>
+ <Row><num>1</num><total>26</total></Row>
+ <Row><num>2</num><total>27</total></Row>
+ <Row><num>3</num><total>28</total></Row>
+</Dataset>
+<Dataset name='Result 2'>
+ <Row><num>0</num><firstnum>0</firstnum><groupcount>25</groupcount><groupaggregate>0</groupaggregate></Row>
+ <Row><num>1</num><firstnum>1</firstnum><groupcount>25</groupcount><groupaggregate>250</groupaggregate></Row>
+ <Row><num>2</num><firstnum>2</firstnum><groupcount>25</groupcount><groupaggregate>500</groupaggregate></Row>
+ <Row><num>3</num><firstnum>3</firstnum><groupcount>25</groupcount><groupaggregate>750</groupaggregate></Row>
+</Dataset>

testing/regress/ecl/wuidresultincq.ecl

@@ -0,0 +1,52 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2019 HPCC Systems®.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+############################################################################## */
+
+// Tests a workunit result inside a child query,
+// generated as side-effect that extends a result
+
+numRecs := 100;
+
+rec := RECORD
+ unsigned num;
+ unsigned firstNum := 0;
+ unsigned groupCount := 0;
+ unsigned groupAggregate := 0;
+END;
+
+ds := DATASET(numRecs, TRANSFORM(rec, SELF.num := COUNTER%4));
+
+sds := SORT(ds, num, LOCAL);
+gds := GROUP(sds, num, LOCAL);
+
+someFunc(DATASET(rec) ss, unsigned groupCount) := FUNCTION
+  pr := PROJECT(ss, TRANSFORM({LEFT.num, unsigned total}, SELF.total := LEFT.num + groupCount; SELF := LEFT));
+  gpr := GROUP(pr, total);
+  o := OUTPUT(gpr, NAMED('cqresult'), EXTEND);
+  RETURN WHEN(true, o, SUCCESS);
+END;
+
+rec rollupTrans(rec l, DATASET(rec) s) := TRANSFORM
+ pr := PROJECT(s, TRANSFORM(rec, SELF.num := LEFT.num*10));
+ SELF.firstNum := l.num;
+ SELF.groupCount := COUNT(pr);
+ SELF.groupAggregate := SUM(pr, num);
+ SELF.num := IF(someFunc(CHOOSEN(s, 1), SELF.groupCount), l.num, 0);
+END;
+
+r := ROLLUP(gds, GROUP, rollupTrans(LEFT, ROWS(LEFT)));
+
+OUTPUT(r);

thorlcr/activities/diskread/thdiskreadslave.cpp

@@ -281,10 +281,11 @@ void CDiskRecordPartHandler::open()
         {
             RemoteFilename rfn;
             partDesc->getFilename(copy, rfn);
-            StringBuffer localPath;
-            if (!isRemoteReadCandidate(activity, rfn, localPath))
+            if (!isRemoteReadCandidate(activity, rfn))
             {
-                Owned<IFile> iFile = createIFile(localPath.str());
+                StringBuffer path;
+                rfn.getPath(path);
+                Owned<IFile> iFile = createIFile(path);
                 try
                 {
                     if (iFile->exists())
@@ -330,7 +331,7 @@ void CDiskRecordPartHandler::open()
                 iRemoteFileIO->addVirtualFieldMapping("logicalFilename", logicalFilename.get());
                 iRemoteFileIO->addVirtualFieldMapping("baseFpos", tmp.clear().append(fileBaseOffset).str());
                 iRemoteFileIO->addVirtualFieldMapping("partNum", tmp.clear().append(partDesc->queryPartIndex()).str());
-                rfn.getPath(path);
+                rfn.getPath(path.clear());
                 filename.set(path);
                 checkFileCrc = false;
 
@@ -348,13 +349,13 @@ void CDiskRecordPartHandler::open()
                     else
                     {
                         remoteReadException.setown(e);
-                        remoteReadExceptionPath.set(path);
+                        remoteReadExceptionPath.set(filename);
                     }
                     e->Release();
                     continue; // try next copy and ultimately failover to local when no more copies
                 }
                 partStream.setown(createRowStreamEx(iRemoteFileIO, activity.queryProjectedDiskRowInterfaces(), 0, (offset_t)-1, (unsigned __int64)-1, rwFlags, nullptr, this));
-                ActPrintLog(&activity, "%s[part=%d]: reading remote dafilesrv file '%s' (logical file = %s)", kindStr, which, path.str(), activity.logicalFilename.get());
+                ActPrintLog(&activity, "%s[part=%d]: reading remote dafilesrv file '%s' (logical file = %s)", kindStr, which, filename.get(), activity.logicalFilename.get());
                 break;
             }
         }

thorlcr/activities/hashdistrib/thhashdistrib.cpp

@@ -139,7 +139,7 @@ public:
         unsigned location;
         OwnedIFile iFile;
         StringBuffer filePath;
-        if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath))
+        if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath, this))
             throw MakeThorException(TE_FileNotFound, "Top level key part does not exist, for key: %s", file->queryLogicalName());
         OwnedIFileIO iFileIO = iFile->open(IFOread);
         assertex(iFileIO);

thorlcr/activities/indexread/thindexreadslave.cpp

@@ -183,10 +183,11 @@ public:
                     {
                         RemoteFilename rfn;
                         part.getFilename(copy, rfn);
-                        StringBuffer localPath;
-                        if (!isRemoteReadCandidate(*this, rfn, localPath))
+                        if (!isRemoteReadCandidate(*this, rfn))
                         {
-                            Owned<IFile> iFile = createIFile(localPath.str());
+                            StringBuffer path;
+                            rfn.getPath(path);
+                            Owned<IFile> iFile = createIFile(path);
                             try
                             {
                                 if (iFile->exists())

thorlcr/activities/keyedjoin/thkeyedjoin-legacy.cpp

@@ -206,7 +206,7 @@ public:
                         StringBuffer filePath;
                         Owned<IFileDescriptor> fileDesc = f->getFileDescriptor();
                         Owned<IPartDescriptor> tlkDesc = fileDesc->getPart(fileDesc->numParts()-1);
-                        if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath))
+                        if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath, this))
                             throw MakeThorException(TE_FileNotFound, "Top level key part does not exist, for key: %s", f->queryLogicalName());
                         OwnedIFileIO iFileIO = iFile->open(IFOread);
                         assertex(iFileIO);

thorlcr/activities/keyedjoin/thkeyedjoin.cpp

@@ -429,7 +429,7 @@ public:
                             StringBuffer filePath;
                             Owned<IFileDescriptor> fileDesc = f->getFileDescriptor();
                             Owned<IPartDescriptor> tlkDesc = fileDesc->getPart(fileDesc->numParts()-1);
-                            if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath))
+                            if (!getBestFilePart(this, *tlkDesc, iFile, location, filePath, this))
                                 throw MakeThorException(TE_FileNotFound, "Top level key part does not exist, for key: %s", f->queryLogicalName());
                             OwnedIFileIO iFileIO = iFile->open(IFOread);
                             assertex(iFileIO);

thorlcr/activities/soapcall/thsoapcallslave.cpp

@@ -51,7 +51,7 @@ public:
     }
 
     // IThorDataLink methods
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
         PARENT::start();
@@ -72,7 +72,7 @@ public:
             wscHelper->start();
         }
     }
-    virtual void stop()
+    virtual void stop() override
     {
         if (wscHelper)
         {
@@ -97,7 +97,7 @@ public:
         return NULL;
     }
     virtual bool isGrouped() const override { return false; }
-    virtual void abort()
+    virtual void abort() override
     {
         CSlaveActivity::abort();
         if (wscHelper)
@@ -110,16 +110,16 @@ public:
     }
 
     // IWSCRowProvider
-    virtual IHThorWebServiceCallActionArg * queryActionHelper()
+    virtual IHThorWebServiceCallActionArg * queryActionHelper() override
     {
         return (static_cast <IHThorWebServiceCallActionArg *> (queryHelper()));
     }
-    virtual IHThorWebServiceCallArg * queryCallHelper()
+    virtual IHThorWebServiceCallArg * queryCallHelper() override
     {
         return (static_cast <IHThorWebServiceCallArg *> (queryHelper()));
     }
-    virtual const void * getNextRow() { return NULL; }
-    virtual void releaseRow(const void *r) { }
+    virtual const void * getNextRow() override { return NULL; }
+    virtual void releaseRow(const void *r) override { }
 };
 
 //---------------------------------------------------------------------------
@@ -143,7 +143,7 @@ public:
     }
 
     // IThorSlaveActivity overloaded methods
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         buildAuthToken(queryJob().queryUserDescriptor(), authToken);
     }
@@ -185,7 +185,8 @@ public:
     virtual void abort() override
     {
         CSlaveActivity::abort();
-        wscHelper->abort();
+        if (wscHelper)
+            wscHelper->abort();
     }
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info) const override
     {
@@ -194,22 +195,22 @@ public:
     }
 
     // IWSCRowProvider
-    virtual IHThorWebServiceCallActionArg * queryActionHelper()
+    virtual IHThorWebServiceCallActionArg * queryActionHelper() override
     {
         return (static_cast <IHThorWebServiceCallActionArg *> (queryHelper()));
     }
-    virtual IHThorWebServiceCallArg * queryCallHelper()
+    virtual IHThorWebServiceCallArg * queryCallHelper() override
     {
         return (static_cast <IHThorWebServiceCallArg *> (queryHelper()));
     }
-    virtual const void * getNextRow()
+    virtual const void * getNextRow() override
     {
         CriticalBlock b(crit);
         if (eof)
             return NULL;
         return inputStream->nextRow();
     }
-    virtual void releaseRow(const void *r)
+    virtual void releaseRow(const void *r) override
     {
         ReleaseThorRow(r);
     }
@@ -232,13 +233,13 @@ public:
     }
 
     // IThorSlaveActivity overloaded methods
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         buildAuthToken(queryJob().queryUserDescriptor(), authToken);
     }
 
     // IThorSlaveProcess overloaded methods
-    virtual void process()
+    virtual void process() override
     {
         if (container.queryLocalOrGrouped() || firstNode())
         {
@@ -250,8 +251,8 @@ public:
                 throw e;
         }
     }
-    virtual void endProcess() { }
-    virtual void abort()
+    virtual void endProcess() override { }
+    virtual void abort() override
     {
         ProcessSlaveActivity::abort();
         if (wscHelper)
@@ -259,13 +260,13 @@ public:
     }
 
     // IWSCRowProvider
-    virtual IHThorWebServiceCallActionArg * queryActionHelper()
+    virtual IHThorWebServiceCallActionArg * queryActionHelper() override
     {
         return (static_cast <IHThorWebServiceCallActionArg *> (queryHelper()));
     }
-    virtual IHThorWebServiceCallArg * queryCallHelper() { return NULL; }
-    virtual const void * getNextRow() { return NULL; }
-    virtual void releaseRow(const void *r) { }
+    virtual IHThorWebServiceCallArg * queryCallHelper() override { return NULL; }
+    virtual const void * getNextRow() override { return NULL; }
+    virtual void releaseRow(const void *r) override { }
 };
 
 //---------------------------------------------------------------------------
@@ -287,13 +288,13 @@ public:
     }
 
     // IThorSlaveActivity overloaded methods
-    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         buildAuthToken(queryJob().queryUserDescriptor(), authToken);
     }
 
     // IThorSlaveProcess overloaded methods
-    virtual void process()
+    virtual void process() override
     {
         start();
         processed = 0;
@@ -307,7 +308,7 @@ public:
         if (e)
             throw e;
     }
-    virtual void endProcess()
+    virtual void endProcess() override
     {
         if (processed & THORDATALINK_STARTED)
         {
@@ -315,7 +316,7 @@ public:
             processed |= THORDATALINK_STOPPED;
         }
     }
-    virtual void abort()
+    virtual void abort() override
     {
         ProcessSlaveActivity::abort();
         if (wscHelper)
@@ -323,12 +324,12 @@ public:
     }
 
     // IWSCRowProvider
-    virtual IHThorWebServiceCallActionArg * queryActionHelper()
+    virtual IHThorWebServiceCallActionArg * queryActionHelper() override
     {
         return (static_cast <IHThorWebServiceCallActionArg *> (queryHelper()));
     }
-    virtual IHThorWebServiceCallArg * queryCallHelper() { return NULL; }
-    virtual const void * getNextRow()
+    virtual IHThorWebServiceCallArg * queryCallHelper() override { return NULL; }
+    virtual const void * getNextRow() override
     {
         CriticalBlock b(crit);
 
@@ -340,7 +341,7 @@ public:
         processed++;
         return row;
     }
-    virtual void releaseRow(const void *r) 
+    virtual void releaseRow(const void *r) override
     {
         ReleaseThorRow(r);
     }

thorlcr/activities/wuidwrite/thwuidwrite.cpp

@@ -38,32 +38,6 @@ protected:
     StringAttr resultName;
     unsigned resultSeq;
 
-public:
-    CWorkUnitWriteMasterBase(CMasterGraphElement * info) : CMasterActivity(info)
-    {
-        numResults = 0;
-        totalSize = 0;
-        resultSeq = 0;
-        appendOutput = false;
-        flushThreshold = -1;
-        workunitWriteLimit = 0;
-        mpTag = container.queryJob().allocateMPTag(); // used by local too
-        activityMaxSize = 0;
-    }
-    virtual void init()
-    {
-        CMasterActivity::init();
-        // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
-        workunitWriteLimit = activityMaxSize ? activityMaxSize : getOptInt(OPT_OUTPUTLIMIT, getOptInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit));
-        if (workunitWriteLimit>defaultDaliResultOutputMax)
-            throw MakeActivityException(this, 0, "Configured max result size, %d MB, exceeds absolute max limit of %d MB. A huge Dali result usually indicates the ECL needs altering.", workunitWriteLimit, defaultDaliResultOutputMax);
-        assertex(workunitWriteLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
-        workunitWriteLimit *= 0x100000;
-    }
-    virtual void serializeSlaveData(MemoryBuffer &dst, unsigned slave)
-    {
-        dst.append((int)mpTag);
-    }
     void addResult(rowcount_t resultCount, MemoryBuffer &resultData, bool complete)
     {
         Owned<IWorkUnit> wu = &container.queryJob().queryWorkUnit().lock();
@@ -89,7 +63,33 @@ public:
             ActPrintLog("result flushed");
         }
     }
-    virtual void abort()
+public:
+    CWorkUnitWriteMasterBase(CMasterGraphElement * info) : CMasterActivity(info)
+    {
+        numResults = 0;
+        totalSize = 0;
+        resultSeq = 0;
+        appendOutput = false;
+        flushThreshold = getOptInt(THOROPT_OUTPUT_FLUSH_THRESHOLD, -1);
+        workunitWriteLimit = 0;
+        mpTag = container.queryJob().allocateMPTag(); // used by local too
+        activityMaxSize = 0;
+    }
+    virtual void init() override
+    {
+        CMasterActivity::init();
+        // In absense of OPT_OUTPUTLIMIT check pre 5.2 legacy name OPT_OUTPUTLIMIT_LEGACY
+        workunitWriteLimit = activityMaxSize ? activityMaxSize : getOptInt(OPT_OUTPUTLIMIT, getOptInt(OPT_OUTPUTLIMIT_LEGACY, defaultDaliResultLimit));
+        if (workunitWriteLimit>defaultDaliResultOutputMax)
+            throw MakeActivityException(this, 0, "Configured max result size, %d MB, exceeds absolute max limit of %d MB. A huge Dali result usually indicates the ECL needs altering.", workunitWriteLimit, defaultDaliResultOutputMax);
+        assertex(workunitWriteLimit<=0x1000); // 32bit limit because MemoryBuffer/CMessageBuffers involved etc.
+        workunitWriteLimit *= 0x100000;
+    }
+    virtual void serializeSlaveData(MemoryBuffer &dst, unsigned slave) override
+    {
+        dst.append((int)mpTag);
+    }
+    virtual void abort() override
     {
         CMasterActivity::abort();
         cancelReceiveMsg(RANK_ALL, mpTag);
@@ -102,7 +102,7 @@ public:
     CWorkUnitWriteGlobalMasterBase(CMasterGraphElement * info) : CWorkUnitWriteMasterBase(info)
     {
     }
-    void process()
+    virtual void process() override
     {
         CWorkUnitWriteMasterBase::process();
 
@@ -152,13 +152,12 @@ public:
     {
         helper = (IHThorWorkUnitWriteArg *)queryHelper();
         appendOutput = 0 != (POFextend & helper->getFlags());
-        flushThreshold = getOptInt(THOROPT_OUTPUT_FLUSH_THRESHOLD, -1);
         resultName.set(helper->queryName());
         resultSeq = helper->getSequence();
         if (POFmaxsize & helper->getFlags())
             activityMaxSize = helper->getMaxSize();
     }
-    virtual void init()
+    virtual void init() override
     {
         CWorkUnitWriteGlobalMasterBase::init();
         if (appendOutput)
@@ -237,7 +236,6 @@ public:
     {
         helper = (IHThorWorkUnitWriteArg *)queryHelper();
         appendOutput = 0 != (POFextend & helper->getFlags());
-        flushThreshold = globals->getPropInt("@output_flush_threshold", -1);
         resultName.set(helper->queryName());
         resultSeq = helper->getSequence();
         sent = 0;
@@ -252,21 +250,21 @@ public:
         replyMsg.swapWith(msg);
         unsigned numGot;
         msg.read(numGot);
+        dbgassertex(numGot); // slave never sends 0
         unsigned l=msg.remaining();
         if (workunitWriteLimit && totalSize+resultData.length()+l > workunitWriteLimit)
             throw MakeThorException(TE_WorkUnitWriteLimitExceeded, "Dataset too large to output to workunit (limit %d megabytes)", workunitWriteLimit/0x100000);
         resultData.append(l, msg.readDirect(l));
         numResults += numGot;
 
-        // NB if 0 == numGot - then final packet from sender
-        if (0 == numGot || (-1 != flushThreshold && resultData.length() >= (unsigned)flushThreshold))
+        if (-1 != flushThreshold && resultData.length() >= (unsigned)flushThreshold)
         {
             totalSize += resultData.length();
             flushResults(0 == numGot);
         }
         queryJobChannel().queryJobComm().reply(replyMsg); // ack
     }
-    virtual void handleSlaveMessage(CMessageBuffer &msg)
+    virtual void handleSlaveMessage(CMessageBuffer &msg) override
     {
         ++sent;
         rank_t sender = container.queryJob().queryJobGroup().rank(msg.getSender());
@@ -276,11 +274,20 @@ public:
         msg.clear();
         queryJobChannel().queryJobComm().reply(msg);
     }
+    virtual void done() override
+    {
+        // NB: This is called when the parent graph is complete
+
+        CWorkUnitWriteMasterBase::done();
+
+        // published any unpublished result or force if 0 results
+        flushResults(0 == numResults);
+    }
 };
 
 CActivityBase *createWorkUnitWriteActivityMaster(CMasterGraphElement *container)
 {
-    if (container->queryLocalOrGrouped())
+    if (container->queryOwner().isLocalChild())
         return new CWorkUnitWriteLocalActivityMaster(container);
     else
         return new CWorkUnitWriteActivityMaster(container);
@@ -298,7 +305,7 @@ public:
         resultName.set(helper->queryName());
         resultSeq = helper->getSequence();
     }
-    virtual void flushResults(bool complete=false)
+    virtual void flushResults(bool complete=false) override
     {
         assertex(complete);
         ActPrintLog("dictionary result");

thorlcr/activities/wuidwrite/thwuidwriteslave.cpp

@@ -159,7 +159,7 @@ public:
         CMessageBuffer msg;
         msg.append((unsigned)0);
         unsigned numGot;
-        do
+        while (!abortSoon)
         {
             numGot = 0;
             try
@@ -174,29 +174,28 @@ public:
                 ActPrintLog("WORKUNITWRITE: exception");
                 throw;
             }
-            if (numGot || totalNum)
+            if (!numGot)
+                break;
+            if (!queryJobChannel().queryJobComm().send(reqMsg, 0, container.queryJob().querySlaveMpTag(), MEDIUMTIMEOUT))
+                throwUnexpected();
+            bool got = false;
+            for (;;)
             {
-                if (!queryJobChannel().queryJobComm().send(reqMsg, 0, container.queryJob().querySlaveMpTag(), MEDIUMTIMEOUT))
-                    throwUnexpected();
-                bool got = false;
-                for (;;)
+                CMessageBuffer replyMsg;
+                if (receiveMsg(replyMsg, 0, replyTag, NULL, MEDIUMTIMEOUT))
                 {
-                    CMessageBuffer replyMsg;
-                    if (receiveMsg(replyMsg, 0, replyTag, NULL, MEDIUMTIMEOUT))
-                    {
-                        msg.setReplyTag(replyTag);
-                        if (!queryJobChannel().queryJobComm().send(msg, 0, mpTag, LONGTIMEOUT))
-                            throwUnexpected();
-                        if (!receiveMsg(replyMsg, 0, replyTag, NULL, LONGTIMEOUT))
-                            throwUnexpected();
-                        got = true;
-                    }
-                    if (got || abortSoon)
-                        break;
-                    ActPrintLog("Blocked (child workunitwrite) sending request to master");
+                    msg.setReplyTag(replyTag);
+                    if (!queryJobChannel().queryJobComm().send(msg, 0, mpTag, LONGTIMEOUT))
+                        throwUnexpected();
+                    if (!receiveMsg(replyMsg, 0, replyTag, NULL, LONGTIMEOUT))
+                        throwUnexpected();
+                    got = true;
                 }
+                if (got || abortSoon)
+                    break;
+                ActPrintLog("Blocked (child workunitwrite) sending request to master");
             }
-        } while (!abortSoon && numGot);
+        }
     }
     virtual void abort() override
     {
@@ -219,7 +218,7 @@ public:
 
 CActivityBase *createWorkUnitWriteSlave(CGraphElementBase *container)
 {
-    if (container->queryLocalOrGrouped())
+    if (container->queryOwner().isLocalChild())
         return new CWorkUnitWriteLocalSlaveActivity(container);
     else
         return new CWorkUnitWriteGlobalSlaveActivity(container);

thorlcr/graph/thgraph.cpp

@@ -1462,12 +1462,29 @@ bool CGraphBase::prepare(size32_t parentExtractSz, const byte *parentExtract, bo
 void CGraphBase::done()
 {
     if (aborted) return; // activity done methods only called on success
-    Owned<IThorActivityIterator> iter = getConnectedIterator();
-    ForEach (*iter)
+
+    if (isLocalChild()) // CQ master activities are created on demand, call done() on any created
     {
-        CGraphElementBase &element = iter->query();
-        element.queryActivity()->done();
+        Owned<IThorActivityIterator> iter = getIterator();
+        ForEach(*iter)
+        {
+            CGraphElementBase &element = iter->query();
+            if (element.queryActivity())
+                element.queryActivity()->done();
+        }
     }
+    else
+    {
+        Owned<IThorActivityIterator> iter = getConnectedIterator();
+        ForEach (*iter)
+        {
+            CGraphElementBase &element = iter->query();
+            element.queryActivity()->done();
+        }
+    }
+    Owned<IThorGraphIterator> childIter = getChildGraphIterator();
+    ForEach(*childIter)
+        childIter->query().done();
 }
 
 unsigned CGraphBase::queryJobChannelNumber() const

thorlcr/thorutil/thormisc.cpp

@@ -1471,15 +1471,16 @@ const ITranslator *getLayoutTranslation(const char *fname, IPartDescriptor &part
     return getTranslators(fname, expectedFormatCrc, expectedFormat, publishedFormatCrc, actualFormat, projectedFormatCrc, projectedFormat, translationMode);
 }
 
-// NB: If returns true, localPath contains the local path of rfn
-bool isRemoteReadCandidate(const CActivityBase &activity, const RemoteFilename &rfn, StringBuffer &localPath)
+bool isRemoteReadCandidate(const CActivityBase &activity, const RemoteFilename &rfn)
 {
     if (!activity.getOptBool(THOROPT_FORCE_REMOTE_DISABLED))
     {
+        if (!rfn.isLocal())
+            return true;
+        StringBuffer localPath;
         rfn.getLocalPath(localPath);
-        if (!rfn.isLocal() || activity.getOptBool(THOROPT_FORCE_REMOTE_READ, testForceRemote(localPath)))
+        if (activity.getOptBool(THOROPT_FORCE_REMOTE_READ, testForceRemote(localPath)))
             return true;
-        localPath.clear();
     }
     return false;
 }

thorlcr/thorutil/thormisc.hpp

@@ -527,7 +527,7 @@ extern graph_decl IThorException *checkAndCreateOOMContextException(CActivityBas
 extern graph_decl RecordTranslationMode getTranslationMode(CActivityBase &activity);
 extern graph_decl void getLayoutTranslations(IConstPointerArrayOf<ITranslator> &translators, const char *fname, IArrayOf<IPartDescriptor> &partDescriptors, RecordTranslationMode translationMode, unsigned expectedFormatCrc, IOutputMetaData *expectedFormat, unsigned projectedFormatCrc, IOutputMetaData *projectedFormat);
 extern graph_decl const ITranslator *getLayoutTranslation(const char *fname, IPartDescriptor &partDesc, RecordTranslationMode translationMode, unsigned expectedFormatCrc, IOutputMetaData *expectedFormat, unsigned projectedFormatCrc, IOutputMetaData *projectedFormat);
-extern graph_decl bool isRemoteReadCandidate(const CActivityBase &activity, const RemoteFilename &rfn, StringBuffer &localPath);
+extern graph_decl bool isRemoteReadCandidate(const CActivityBase &activity, const RemoteFilename &rfn);
 
 extern graph_decl void checkAndDumpAbortInfo(const char *cmd);