Merge remote-tracking branch 'origin/candidate-3.10.x'

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

common/workunit/workunit.cpp

@@ -4728,6 +4728,13 @@ void CLocalWorkUnit::copyWorkUnit(IConstWorkUnit *cached, bool all)
     updateProp(p, fromP, "CustomerID");
     updateProp(p, fromP, "SNAPSHOT");
 
+    //MORE: This is very adhoc.  All options that should be cloned should really be in a common branch
+    if (all && (fromP->hasProp("PriorityFlag") || fromP->hasProp("@priorityClass")))
+    {
+        updateProp(p, fromP, "PriorityFlag");
+        updateProp(p, fromP, "@priorityClass");
+    }
+
     //Variables may have been set up as parameters to the query - so need to preserve any values that were supplied.
     pt = fromP->getBranch("Variables");
     if (pt)

dali/base/dadfs.cpp

@@ -4167,6 +4167,9 @@ class CDistributedSuperFile: public CDistributedFileBase<IDistributedSuperFile>
             if (!subfile.isEmpty()) {
                 try {
                     sub.setown(transaction->lookupFile(subfile,SDS_SUB_LOCK_TIMEOUT));
+                    // Must validate before locking for update below, to check sub is not already in parent (and therefore locked already)
+                    CDistributedSuperFile *sf = QUERYINTERFACE(parent.get(),CDistributedSuperFile);
+                    sf->validateAddSubFile(sub);
                 }
                 catch (IDFS_Exception *e) {
                     if (e->errorCode()!=DFSERR_LookupConnectionTimout)
@@ -5258,15 +5261,19 @@ public:
     }
 
 private:
-    void doAddSubFile(IDistributedFile *_sub,bool before,const char *other,IDistributedFileTransaction *transaction) // takes ownership of sub
+    void validateAddSubFile(IDistributedFile *sub)
     {
-        Owned<IDistributedFile> sub = _sub;
         if (strcmp(sub->queryLogicalName(),queryLogicalName())==0)
             throw MakeStringException(-1,"addSubFile: Cannot add file %s to itself", queryLogicalName());
         if (subfiles.ordinality())
             checkFormatAttr(sub,"addSubFile");
         if (findSubFile(sub->queryLogicalName())!=NotFound)
             throw MakeStringException(-1,"addSubFile: File %s is already a subfile of %s", sub->queryLogicalName(),queryLogicalName());
+    }
+    void doAddSubFile(IDistributedFile *_sub,bool before,const char *other,IDistributedFileTransaction *transaction) // takes ownership of sub
+    {
+        Owned<IDistributedFile> sub = _sub;
+        validateAddSubFile(sub);
 
         unsigned pos;
         if (other&&*other) {

ecllibrary/std/File.ecl

@@ -353,8 +353,8 @@ EXPORT DfuPlusExec(varstring cmdline) :=
 EXPORT SprayFixed(varstring sourceIP, varstring sourcePath, integer4 recordSize, varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false) :=
     lib_fileservices.FileServices.SprayFixed(sourceIP, sourcePath, recordSize, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress);
 
-EXPORT SprayVariable(varstring sourceIP, varstring sourcePath, integer4 sourceMaxRecordSize=8192, varstring sourceCsvSeparate='\\,', varstring sourceCsvTerminate='\\n,\\r\\n', varstring sourceCsvQuote='\'', varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false) :=
-    lib_fileservices.FileServices.SprayVariable(sourceIP, sourcePath, sourceMaxRecordSize, sourceCsvSeparate, sourceCsvTerminate, sourceCsvQuote, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress);
+EXPORT SprayVariable(varstring sourceIP, varstring sourcePath, integer4 sourceMaxRecordSize=8192, varstring sourceCsvSeparate='\\,', varstring sourceCsvTerminate='\\n,\\r\\n', varstring sourceCsvQuote='\'', varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false, varstring sourceCsvEscape='') :=
+    lib_fileservices.FileServices.SprayVariable(sourceIP, sourcePath, sourceMaxRecordSize, sourceCsvSeparate, sourceCsvTerminate, sourceCsvQuote, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress, sourceCsvEscape);
 
 EXPORT SprayXml(varstring sourceIP, varstring sourcePath, integer4 sourceMaxRecordSize=8192, varstring sourceRowTag, varstring sourceEncoding='utf8', varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false) :=
     lib_fileservices.FileServices.SprayXml(sourceIP, sourcePath, sourceMaxRecordSize, sourceRowTag, sourceEncoding, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress);
@@ -371,8 +371,8 @@ EXPORT Replicate(varstring logicalName, integer4 timeOut=-1, varstring espServer
 EXPORT varstring fSprayFixed(varstring sourceIP, varstring sourcePath, integer4 recordSize, varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false) :=
     lib_fileservices.FileServices.fSprayFixed(sourceIP, sourcePath, recordSize, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress);
 
-EXPORT varstring fSprayVariable(varstring sourceIP, varstring sourcePath, integer4 sourceMaxRecordSize=8192, varstring sourceCsvSeparate='\\,', varstring sourceCsvTerminate='\\n,\\r\\n', varstring sourceCsvQuote='\'', varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false) :=
-    lib_fileservices.FileServices.fSprayVariable(sourceIP, sourcePath, sourceMaxRecordSize, sourceCsvSeparate, sourceCsvTerminate, sourceCsvQuote, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress);
+EXPORT varstring fSprayVariable(varstring sourceIP, varstring sourcePath, integer4 sourceMaxRecordSize=8192, varstring sourceCsvSeparate='\\,', varstring sourceCsvTerminate='\\n,\\r\\n', varstring sourceCsvQuote='\'', varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false, varstring sourceCsvEscape='') :=
+    lib_fileservices.FileServices.fSprayVariable(sourceIP, sourcePath, sourceMaxRecordSize, sourceCsvSeparate, sourceCsvTerminate, sourceCsvQuote, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress, sourceCsvEscape);
 
 EXPORT varstring fSprayXml(varstring sourceIP, varstring sourcePath, integer4 sourceMaxRecordSize=8192, varstring sourceRowTag, varstring sourceEncoding='utf8', varstring destinationGroup, varstring destinationLogicalName, integer4 timeOut=-1, varstring espServerIpPort=GETENV('ws_fs_server'), integer4 maxConnections=-1, boolean allowoverwrite=false, boolean replicate=false, boolean compress=false) :=
     lib_fileservices.FileServices.fSprayXml(sourceIP, sourcePath, sourceMaxRecordSize, sourceRowTag, sourceEncoding, destinationGroup, destinationLogicalName, timeOut, espServerIpPort, maxConnections, allowoverwrite, replicate, compress);

esp/eclwatch/ws_XSLT/WUQuerysetQueries.xslt

@@ -136,7 +136,7 @@
                   return soapXML;
               }
 
-              function deleteAliases() {
+              function deactivateAliases() {
                 actionAliases('Deactivate');
               }
 
@@ -403,9 +403,9 @@
             <div>
               <div id="aliasdiv">&#160;</div>
               <br/>
-              <span id="DeleteAliasButton" class="yui-button yui-push-button">
+              <span id="DeactivateAliasButton" class="yui-button yui-push-button">
                 <em class="first-child">
-                  <button type="button" name="DeleteAliasButton" onclick="deleteAliases();">Delete</button>
+                  <button type="button" name="DeactivateAliasButton" onclick="deactivateAliases();">Deactivate</button>
                 </em>
               </span>
               <span id="DetailsButton" class="yui-button yui-push-button">

esp/scm/ws_workunits.ecm

@@ -287,6 +287,12 @@ ESPStruct [nil_remove] ThorQueue
     string RunningWU2;
 };
 
+ESPStruct [nil_remove] ResubmittedWU
+{
+    string WUID;
+    string ParentWUID;
+};
+
 ESPrequest [nil_remove] WUCreateRequest
 {
 };
@@ -494,6 +500,7 @@ ESPrequest WUResubmitRequest
 
 ESPresponse [exceptions_inline] WUResubmitResponse
 {
+    [min_ver("1.40")] ESParray<ESPstruct ResubmittedWU, WU> WUs;
 };
 
 ESPrequest WURunRequest
@@ -1328,7 +1335,7 @@ ESPresponse [exceptions_inline] WUQuerySetCopyQueryResponse
 };
 
 ESPservice [
-    version("1.39"), default_client_version("1.39"),
+    version("1.40"), default_client_version("1.40"),
     noforms,exceptions_inline("./smc_xslt/exceptions.xslt"),use_method_name] WsWorkunits
 {
     ESPmethod [resp_xsl_default("/esp/xslt/workunits.xslt")]     WUQuery(WUQueryRequest, WUQueryResponse);

esp/services/ws_workunits/ws_workunitsService.cpp

@@ -1084,14 +1084,16 @@ bool CWsWorkunitsEx::onWUResubmit(IEspContext &context, IEspWUResubmitRequest &r
         SCMStringBuffer wuid;
         StringArray wuids;
 
+        double version = context.getClientVersion();
+        IArrayOf<IEspResubmittedWU> resubmittedWUs;
         for(aindex_t i=0; i<req.getWuids().length();i++)
         {
-            StringBuffer wuidStr = req.getWuids().item(i);
-            checkAndTrimWorkunit("WUResubmit", wuidStr);
+            StringBuffer requestWuid = req.getWuids().item(i);
+            checkAndTrimWorkunit("WUResubmit", requestWuid);
 
-            ensureWsWorkunitAccess(context, wuidStr.str(), SecAccess_Write);
+            ensureWsWorkunitAccess(context, requestWuid.str(), SecAccess_Write);
 
-            wuid.set(wuidStr.str());
+            wuid.set(requestWuid.str());
 
             try
             {
@@ -1114,6 +1116,15 @@ bool CWsWorkunitsEx::onWUResubmit(IEspContext &context, IEspWUResubmitRequest &r
                     throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT,"Cannot open workunit %s.",wuid.str());
 
                 submitWsWorkunit(context, cw, NULL, NULL, 0, req.getRecompile(), req.getResetWorkflow(), false);
+
+                if (version < 1.40)
+                    continue;
+
+                Owned<IEspResubmittedWU> resubmittedWU = createResubmittedWU();
+                resubmittedWU->setWUID(wuid.str());
+                if (!streq(requestWuid.str(), wuid.str()))
+                    resubmittedWU->setParentWUID(requestWuid.str());
+                resubmittedWUs.append(*resubmittedWU.getClear());
             }
             catch (IException *E)
             {
@@ -1135,6 +1146,9 @@ bool CWsWorkunitsEx::onWUResubmit(IEspContext &context, IEspWUResubmitRequest &r
                 waitForWorkUnitToComplete(wuids.item(i), timeToWait);
         }
 
+        if (version >= 1.40)
+            resp.setWUs(resubmittedWUs);
+
         if(wuids.length()==1)
         {
             resp.setRedirectUrl(StringBuffer("/WsWorkunits/WUInfo?Wuid=").append(wuids.item(0)));

plugins/fileservices/fileservices.hpp

@@ -51,6 +51,7 @@ FILESERVICES_API char * FILESERVICES_CALL fsCmdProcess(const char *prog, const c
 FILESERVICES_API void FILESERVICES_CALL fsCmdProcess2(unsigned & tgtLen, char * & tgt, const char *prog, unsigned srcLen, const char * src);
 FILESERVICES_API void FILESERVICES_CALL fsSprayFixed(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int recordSize, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite=false, bool replicate=false, bool compress=false);
 FILESERVICES_API void FILESERVICES_CALL fsSprayVariable(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int sourceMaxRecordSize, const char * sourceCsvSeparate, const char * sourceCsvTerminate, const char * sourceCsvQuote, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite=false, bool replicate=false, bool compress=false);
+FILESERVICES_API void FILESERVICES_CALL fsSprayVariable2(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int sourceMaxRecordSize, const char * sourceCsvSeparate, const char * sourceCsvTerminate, const char * sourceCsvQuote, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite, bool replicate, bool compress, const char * sourceCsvEscape);
 FILESERVICES_API void FILESERVICES_CALL fsSprayXml(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int sourceMaxRecordSize, const char *sourceRowTag, const char *sourceEncoding, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite, bool replicate, bool compress=false);
 FILESERVICES_API void FILESERVICES_CALL fsDespray(ICodeContext *ctx, const char * sourceLogicalName, const char * destinationIP, const char * destinationPath, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite=false);
 FILESERVICES_API void FILESERVICES_CALL fsCopy(ICodeContext *ctx, const char * sourceLogicalName, const char *destinationGroup, const char * destinationLogicalName, const char * sourceDali, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite, bool replicate, bool asSuperfile, bool compress, bool forcePush, int transferBufferSize);
@@ -76,6 +77,7 @@ FILESERVICES_API void  FILESERVICES_CALL fsMonitorLogicalFileName(ICodeContext *
 FILESERVICES_API void  FILESERVICES_CALL fsMonitorFile(ICodeContext *ctx, const char *eventname, const char *ip, const char *filename, bool sub, int shotcount, const char * espServerIpPort);
 FILESERVICES_API char * FILESERVICES_CALL fsfSprayFixed(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int recordSize, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite=false, bool replicate=false, bool compress=false);
 FILESERVICES_API char * FILESERVICES_CALL fsfSprayVariable(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int sourceMaxRecordSize, const char * sourceCsvSeparate, const char * sourceCsvTerminate, const char * sourceCsvQuote, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite=false, bool replicate=false, bool compress=false);
+FILESERVICES_API char * FILESERVICES_CALL fsfSprayVariable2(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int sourceMaxRecordSize, const char * sourceCsvSeparate, const char * sourceCsvTerminate, const char * sourceCsvQuote, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite, bool replicate, bool compress, const char * sourceCsvEscape);
 FILESERVICES_API char * FILESERVICES_CALL fsfSprayXml(ICodeContext *ctx, const char * sourceIP, const char * sourcePath, int sourceMaxRecordSize, const char *sourceRowTag, const char *sourceEncoding, const char * destinationGroup, const char * destinationLogicalName, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite, bool replicate, bool compress=false);
 FILESERVICES_API char * FILESERVICES_CALL fsfDespray(ICodeContext *ctx, const char * sourceLogicalName, const char * destinationIP, const char * destinationPath, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite=false);
 FILESERVICES_API char * FILESERVICES_CALL fsfCopy(ICodeContext *ctx, const char * sourceLogicalName, const char *destinationGroup, const char * destinationLogicalName, const char * sourceDali, int timeOut, const char * espServerIpPort, int maxConnections, bool overwrite, bool replicate, bool asSuperfile, bool compress, bool forcePush, int transferBufferSize);

testing/ecl/key/spray_test2.xml

@@ -0,0 +1,5 @@
+<Dataset name='Result 1'>
+ <Row><name>foo</name><age>10</age><good>true</good></Row>
+ <Row><name>bar</name><age>12</age><good>false</good></Row>
+ <Row><name>baz</name><age>32</age><good>true</good></Row>
+</Dataset>

testing/ecl/spray_test2.ecl

@@ -0,0 +1,41 @@
+/*##############################################################################
+
+    Copyright (C) 2013 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.
+############################################################################## */
+
+import Std.File AS FileServices;
+
+// This is not an engine test, but a DFU.
+// Doesn't matter much which engine does it, so we restrict to only one
+
+//noRoxie
+//noThorLCR
+//noThor
+
+SrcIP := 'localhost';
+File := tempFiles + '/spray_test.txt';
+ClusterName := 'mythor';
+DestFile := '~regress::spray_test2.txt';
+ESPportIP := 'http://127.0.0.1:8010/FileSpray';
+
+FileServices.SprayVariable(SrcIP,
+                        File,
+                        destinationGroup := ClusterName,
+                        destinationLogicalName := DestFile,
+                        espServerIpPort := ESPportIP,
+                        ALLOWOVERWRITE := true);
+
+ds := DATASET(DestFile, { string name, unsigned age, boolean good }, csv);
+output(ds);

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -1141,7 +1141,7 @@ Restart:
 
     void startTX()
     {
-        // not used
+        stopping = false;
     }
 };
 

thorlcr/activities/indexwrite/thindexwrite.cpp

@@ -61,9 +61,12 @@ public:
         OwnedRoxieString fname(helper->getFileName());
         dlfn.set(fname);
         isLocal = 0 != (TIWlocal & helper->getFlags());
-        unsigned maxSize = helper->queryDiskRecordSize()->getMinRecordSize();
+        unsigned minSize = helper->queryDiskRecordSize()->getMinRecordSize();
+        if (minSize > KEYBUILD_MAXLENGTH)
+            throw MakeActivityException(this, 0, "Index minimum record length (%d) exceeds %d internal limit", minSize, KEYBUILD_MAXLENGTH);
+        unsigned maxSize = helper->queryDiskRecordSize()->getRecordSize(NULL);
         if (maxSize > KEYBUILD_MAXLENGTH)
-            throw MakeActivityException(this, 0, "Index minimum record length (%d) exceeds %d internal limit", maxSize, KEYBUILD_MAXLENGTH);
+            throw MakeActivityException(this, 0, "Index maximum record length (%d) exceeds %d internal limit. Minimum size = %d, try setting index MAXLENGTH", maxSize, KEYBUILD_MAXLENGTH, minSize);
 
         singlePartKey = 0 != (helper->getFlags() & TIWsmall) || dlfn.isExternal();
         clusters.kill();