Merge pull request #1897 from hpcc-systems/candidate-3.6.x

Merge latest 3.6 changes back to master

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>

common/thorhelper/thorcommon.cpp

@@ -580,24 +580,6 @@ public:
     }
 };
 
-
-//Deprecated - should use the second definition below
-void * cloneRow(IEngineRowAllocator * allocator, const void * row, size32_t &sizeout)
-{
-    IOutputMetaData * meta = allocator->queryOutputMeta();
-    void * ret = allocator->createRow();        
-    sizeout = meta->getRecordSize(row);     // TBD could be better?
-    //GH this may no longer be big enough
-    memcpy(ret, row, sizeout);
-    if (meta->getMetaFlags() & MDFneedserialize)
-    {
-        ChildRowLinkerWalker walker;
-        meta->walkIndirectMembers(static_cast<const byte *>(ret), walker);
-    }
-    //NB: Does not call finalizeRow()...
-    return ret;
-}
-
 //the visitor callback is used to ensure link counts for children are updated.
 size32_t cloneRow(ARowBuilder & rowBuilder, const void * row, IOutputMetaData * meta)
 {

common/thorhelper/thorcommon.ipp

@@ -709,17 +709,8 @@ class NullDiskCallback : public IThorDiskCallback, extends CInterface
     virtual const char * queryLogicalFilename(const void * row) { return NULL; }
 };
 
-extern THORHELPER_API void * cloneRow(IEngineRowAllocator * allocator, const void * row, size32_t &sizeout);
 extern THORHELPER_API size32_t cloneRow(ARowBuilder & rowBuilder, const void * row, IOutputMetaData * meta);
 
-inline void *cloneRow(ICodeContext * ctx, IEngineRowAllocator * allocator, const void * row)
-{
-    // legacy version that doesn't return size
-    size32_t sztmp;
-    return cloneRow(allocator, row, sztmp);
-}
-
-
 //The CThorContiguousRowBuffer is the source for a readAhead call to ensure the entire row
 //is in a contiguous block of memory.  The read() and skip() functions must be implemented
 class THORHELPER_API CThorContiguousRowBuffer : implements IRowDeserializerSource

common/thorhelper/thorsoapcall.cpp

@@ -715,7 +715,8 @@ public:
 
         rowProvider = _rowProvider;
         helper = rowProvider->queryActionHelper();
-        callHelper = rowProvider->queryCallHelper();
+        callHelper = rowProvider->queryCallHelper();  //MORE: This should not be done this way!! Should use extra as below.
+        helperExtra = static_cast<IHThorSoapCallExtra*>(helper->selectInterface(TAIsoapcallextra_1));
         flags = helper->getFlags();
 
         authToken.append(_authToken);
@@ -725,7 +726,10 @@ public:
         else
             maxRetries = helper->numRetries();
 
+        //Allow all of these options to be specified separately.  Possibly useful, and the code is cleaner.
+        logMin = (flags & SOAPFlogmin) != 0;
         logXML = (flags & SOAPFlog) != 0;
+        logUserMsg = (flags & SOAPFlogusermsg) != 0;
 
         timeout = helper->getTimeout();
         if (timeout == (unsigned)-1)
@@ -975,7 +979,19 @@ public:
         return false;
     }
 
+    void addUserLogMsg(const byte * row)
+    {
+        if (logUserMsg)
+        {
+            size32_t lenText;
+            rtlDataAttr text;
+            helperExtra->getLogText(lenText, text.refstr(), row);
+            logctx.CTXLOG("%s: %.*s", wscCallTypeText(), lenText, text.getstr());
+        }
+    }
+
     inline IXmlToRowTransformer * getRowTransformer() { return rowTransformer; }
+    inline const char * wscCallTypeText() const { return wscType == STsoap ? "SOAPCALL" : "HTTPCALL"; }
 
 protected:
     friend class CWSCHelperThread;
@@ -1019,6 +1035,7 @@ protected:
     IWSCRowProvider * rowProvider;
     IHThorWebServiceCallActionArg * helper;
     IHThorWebServiceCallArg *   callHelper;
+    IHThorWebServiceCallExtra * helperExtra;
     Linked<IEngineRowAllocator> outputAllocator;
     Owned<IException> error;
     UrlArray urlArray;
@@ -1030,7 +1047,10 @@ protected:
     unsigned maxRetries;
     unsigned timeout; //seconds
     unsigned timeLimit; //seconds
-    bool logXML, aborted;
+    bool logXML;
+    bool logMin;
+    bool logUserMsg;
+    bool aborted;
     const IContextLogger &logctx;
     unsigned flags;
     StringAttr soapaction;
@@ -1085,6 +1105,8 @@ void CWSCHelperThread::outputXmlRows(CommonXmlWriter &xmlWriter, ConstPointerArr
             xmlWriter.outputQuoted(itemtag);
             xmlWriter.outputQuoted(">");
         }
+
+        master->addUserLogMsg((const byte *)inputRows.item(idx));
     }
 }
 
@@ -1393,7 +1415,10 @@ private:
             request.append("\r\n");//httpcall
 
         if (soapTraceLevel > 6 || master->logXML)
-            master->logctx.CTXLOG("%sCALL: request(%s)", master->wscType == STsoap ? "SOAP" : "HTTP", request.str());
+            master->logctx.CTXLOG("%s: request(%s)", master->wscCallTypeText(), request.str());
+
+        if (master->logMin)
+            master->logctx.CTXLOG("%s: request(%s:%u)", master->wscCallTypeText(), url.host.str(), url.port);
     }
 
     int readHttpResponse(StringBuffer &response, ISocket *socket)

ecl/hql/hqlfold.cpp

@@ -5316,6 +5316,7 @@ HqlConstantPercolator * CExprFolderTransformer::gatherConstants(IHqlExpression *
     case no_serialize:
     case no_typetransfer:
     case no_fromxml:
+    case no_httpcall:
         break;
 
     case no_null:

ecl/hql/hqlgram.y

@@ -3197,6 +3197,15 @@ soapFlag
                             $$.setExpr(createExprAttribute(responseAtom, createAttribute(noTrimAtom)), $1);
                         }
     | commonAttribute
+    | TOK_LOG '(' MIN ')'
+                        {
+                            $$.setExpr(createExprAttribute(logAtom, createAttribute(minAtom)), $1);
+                        }
+    | TOK_LOG '(' expression ')'
+                        {
+                            parser->normalizeExpression($3, type_string, false);
+                            $$.setExpr(createExprAttribute(logAtom, $3.getExpr()), $1);
+                        }
     ;
 
 onFailAction

ecl/hqlcpp/hqlcpp.ipp

@@ -414,7 +414,7 @@ class GlobalFileTracker : public CInterface, public IHqlDelayedCodeGenerator
 public:
     GlobalFileTracker(IHqlExpression * _filename, IPropertyTree * _graphNode)
     {
-        filename.set(_filename);
+        filename.set(_filename->queryBody());
         graphNode.set(_graphNode);
         usageCount = 0;
     }

ecl/hqlcpp/hqlhtcpp.cpp

@@ -1721,7 +1721,7 @@ void HqlCppTranslator::finalizeTempRow(BuildCtx & ctx, BoundRow * row, BoundRow
 
 bool GlobalFileTracker::checkMatch(IHqlExpression * searchFilename)
 {
-    if (searchFilename == filename.get())
+    if (searchFilename->queryBody() == filename.get())
     {
         usageCount++;
         return true;
@@ -9754,11 +9754,12 @@ void HqlCppTranslator::buildXmlWriteMembers(ActivityInstance * instance, IHqlExp
 ABoundActivity * HqlCppTranslator::doBuildActivityOutput(BuildCtx & ctx, IHqlExpression * expr, bool isRoot)
 {
     IHqlExpression * dataset  = expr->queryChild(0);
-    IHqlExpression * filename = queryRealChild(expr, 1);
+    IHqlExpression * rawFilename = queryRealChild(expr, 1);
 
-    if (!filename)
+    if (!rawFilename)
         return doBuildActivityOutputWorkunit(ctx, expr, isRoot);
 
+    OwnedHqlExpr filename = foldHqlExpression(rawFilename);
     IHqlExpression * program  = queryRealChild(expr, 2);
     IHqlExpression * csvAttr = expr->queryProperty(csvAtom);
     IHqlExpression * xmlAttr = expr->queryProperty(xmlAtom);
@@ -16157,6 +16158,24 @@ ABoundActivity * HqlCppTranslator::doBuildActivitySOAP(BuildCtx & ctx, IHqlExpre
 
     IHqlExpression * namespaceAttr = expr->queryProperty(namespaceAtom);
     IHqlExpression * responseAttr = expr->queryProperty(responseAtom);
+    IHqlExpression * logText = NULL;
+    bool logMin = false;
+    bool logXml = false;
+    ForEachChildFrom(i, expr, 1)
+    {
+        IHqlExpression * cur = expr->queryChild(i);
+        if (cur->isAttribute() && cur->queryName()==logAtom)
+        {
+            IHqlExpression * opt = cur->queryChild(0);
+            if (!opt)
+                logXml = true;
+            else if (!opt->isAttribute())
+                logText = opt;
+            else if (opt->queryName() == minAtom)
+                logMin = true;
+        }
+    }
+
     //virtual unsigned getFlags()
     {
         StringBuffer flags;
@@ -16164,7 +16183,7 @@ ABoundActivity * HqlCppTranslator::doBuildActivitySOAP(BuildCtx & ctx, IHqlExpre
             flags.append("|SOAPFgroup");
         if (expr->hasProperty(onFailAtom))
             flags.append("|SOAPFonfail");
-        if (expr->hasProperty(logAtom))
+        if (logXml)
             flags.append("|SOAPFlog");
         if (expr->hasProperty(trimAtom))
             flags.append("|SOAPFtrim");
@@ -16176,6 +16195,10 @@ ABoundActivity * HqlCppTranslator::doBuildActivitySOAP(BuildCtx & ctx, IHqlExpre
             flags.append("|SOAPFencoding");
         if (responseAttr && responseAttr->hasProperty(noTrimAtom))
             flags.append("|SOAPFpreserveSpace");
+        if (logMin)
+            flags.append("|SOAPFlogmin");
+        if (logText)
+            flags.append("|SOAPFlogusermsg");
 
         if (flags.length())
             doBuildUnsignedFunction(instance->classctx, "getFlags", flags.str()+1);
@@ -16203,6 +16226,18 @@ ABoundActivity * HqlCppTranslator::doBuildActivitySOAP(BuildCtx & ctx, IHqlExpre
             doBuildVarStringFunction(instance->startctx, "queryNamespaceVar", namespaceAttr->queryChild(1));
     }
 
+    if (logText)
+    {
+        BuildCtx funcctx(instance->startctx);
+        funcctx.addQuotedCompound("virtual void getLogText(size32_t & __lenResult, char * & __result, const void * _left)");
+        if (dataset)
+        {
+            funcctx.addQuoted("const unsigned char * left = (const unsigned char *) _left;");
+            bindTableCursor(funcctx, dataset, "left");
+            bindTableCursor(funcctx, dataset, "left", no_left, selSeq);
+        }
+        doBuildFunctionReturn(funcctx, unknownStringType, logText);
+    }
     if (!isSink)
     {
         //virtual IXmlToRowTransformer * queryTransformer()
@@ -16279,6 +16314,24 @@ ABoundActivity * HqlCppTranslator::doBuildActivityHTTP(BuildCtx & ctx, IHqlExpre
         doBuildVarStringFunction(instance->startctx, "queryOutputIteratorPath", separator->queryChild(0));
 
     IHqlExpression * namespaceAttr = expr->queryProperty(namespaceAtom);
+    IHqlExpression * logText = NULL;
+    bool logMin = false;
+    bool logXml = false;
+    ForEachChildFrom(i, expr, 1)
+    {
+        IHqlExpression * cur = expr->queryChild(i);
+        if (cur->isAttribute() && cur->queryName()==logAtom)
+        {
+            IHqlExpression * opt = cur->queryChild(0);
+            if (!opt)
+                logXml = true;
+            else if (!opt->isAttribute())
+                logText = opt;
+            else if (opt->queryName() == minAtom)
+                logMin = true;
+        }
+    }
+
     //virtual unsigned getFlags()
     {
         StringBuffer flags;
@@ -16286,7 +16339,7 @@ ABoundActivity * HqlCppTranslator::doBuildActivityHTTP(BuildCtx & ctx, IHqlExpre
             flags.append("|SOAPFgroup");
         if (expr->hasProperty(onFailAtom))
             flags.append("|SOAPFonfail");
-        if (expr->hasProperty(logAtom))
+        if (logXml)
             flags.append("|SOAPFlog");
         if (expr->hasProperty(trimAtom))
             flags.append("|SOAPFtrim");
@@ -16294,6 +16347,10 @@ ABoundActivity * HqlCppTranslator::doBuildActivityHTTP(BuildCtx & ctx, IHqlExpre
             flags.append("|SOAPFliteral");
         if (namespaceAttr)
             flags.append("|SOAPFnamespace");
+        if (logMin)
+            flags.append("|SOAPFlogmin");
+        if (logText)
+            flags.append("|SOAPFlogusermsg");
 
         if (flags.length())
             doBuildUnsignedFunction(instance->classctx, "getFlags", flags.str()+1);
@@ -16321,6 +16378,13 @@ ABoundActivity * HqlCppTranslator::doBuildActivityHTTP(BuildCtx & ctx, IHqlExpre
             doBuildVarStringFunction(instance->startctx, "queryNamespaceVar", namespaceAttr->queryChild(1));
     }
 
+    if (logText)
+    {
+        BuildCtx funcctx(instance->startctx);
+        funcctx.addQuotedCompound("virtual void getLogText(size32_t & __lenResult, char * & __result, const void * _left)");
+        doBuildFunctionReturn(funcctx, unknownStringType, logText);
+    }
+
     if (!isSink)
     {
         //virtual IXmlToRowTransformer * queryTransformer()

ecl/hqlcpp/hqlsource.cpp

@@ -664,8 +664,9 @@ class SourceBuilder
 {
 public:
     SourceBuilder(HqlCppTranslator & _translator, IHqlExpression *_tableExpr, IHqlExpression *_nameExpr)
-        : tableExpr(_tableExpr), nameExpr(_nameExpr), translator(_translator)
+        : tableExpr(_tableExpr), translator(_translator)
     { 
+        nameExpr.setown(foldHqlExpression(_nameExpr));
         needDefaultTransform = true; 
         needToCallTransform = false; 
         isPreloaded = false;

ecl/hqlcpp/hqlttcpp.cpp

@@ -4611,7 +4611,8 @@ IHqlExpression * GlobalAttributeInfo::queryFilename(IHqlExpression * value, ICon
             {
                 ITypeInfo * type = makeStringType(UNKNOWN_LENGTH, NULL, NULL);
 
-                cachedFilename.setown(createValue(no_concat, type, createConstant(prefix), cachedFilename.getClear()));
+                OwnedHqlExpr filename = createValue(no_concat, type, createConstant(prefix), cachedFilename.getClear());
+                cachedFilename.setown(foldHqlExpression(filename));
             }
         }
     }

ecl/hthor/hthorkey.cpp

@@ -954,7 +954,9 @@ const void *CHThorIndexReadActivity::nextInGroup()
                 }
                 try
                 {
-                    return cloneRow(agent.queryCodeContext(), rowAllocator, keyRow);
+                    RtlDynamicRowBuilder rowBuilder(rowAllocator);
+                    size32_t finalSize = cloneRow(rowBuilder, keyRow, outputMeta);
+                    return rowBuilder.finalizeRowClear(finalSize);
                 }
                 catch(IException * e)
                 {
@@ -1040,7 +1042,9 @@ const void *CHThorIndexReadActivity::nextGE(const void * seek, unsigned numField
                 }
                 try
                 {
-                    return cloneRow(agent.queryCodeContext(), rowAllocator, row);
+                    RtlDynamicRowBuilder rowBuilder(rowAllocator);
+                    size32_t finalSize = cloneRow(rowBuilder, row, outputMeta);
+                    return rowBuilder.finalizeRowClear(finalSize);
                 }
                 catch(IException * e)
                 {

ecl/regress/soapcall12.ecl

@@ -0,0 +1,73 @@
+/*##############################################################################
+
+    Copyright (C) 2011 HPCC Systems.
+
+    All rights reserved. This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as
+    published by the Free Software Foundation, either version 3 of the
+    License, or (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+############################################################################## */
+
+//Test all the different variants of SOAPCALL to ensure they get processed correctly
+
+GetAttributeInRecord := RECORD
+string                ModuleName{xpath('ModuleName')};
+string                AttributeName{xpath('AttributeName')};
+unsigned              Version{xpath('Version')};
+boolean               GetSandbox{xpath('GetSandbox')};
+boolean               GetText{xpath('GetText')};
+                 END;
+
+
+GetAttributeOutRecord := RECORD
+string                ModuleName{xpath('ModuleName')};
+string                text{xpath('Text')};
+                END;
+
+
+GetAttributeInRecord createInRecord := TRANSFORM
+            SELF.ModuleName := 'SearchModule';
+            SELF.AttributeName := 'SearchAttribute';
+            SELF.Version := 0;
+            SELF.GetSandbox := true;
+            SELF.GetText := true;
+        END;
+
+GetAttributeInRecord createInRecord2(string moduleName, string attr) := TRANSFORM
+            SELF.ModuleName := moduleName;
+            SELF.AttributeName := attr;
+            SELF.Version := 0;
+            SELF.GetSandbox := true;
+            SELF.GetText := true;
+        END;
+
+GetAttributeInRecord2 := RECORD
+string                ModuleName{xpath('ModuleName')} := 'MyModule';
+string                AttributeName{xpath('AttributeName')} := 'MyAttribute';
+unsigned              Version{xpath('Version')} := 1;
+boolean               GetSandbox{xpath('GetSandbox')} := true;
+boolean               GetText{xpath('GetText')} := true;
+                 END;
+
+ds := dataset([{'doxie','One'},{'jimbo','Two'},{'yankie','Three'}],{string moduleName, string attr});
+
+//--No argument, record provides values--
+//actions
+SOAPCALL('http://webservices.megacorp.com', 'WsAttributes', GetAttributeInRecord2, LOG(MIN));
+SOAPCALL('http://webservices.megacorp.com', 'WsAttributes', GetAttributeInRecord2, RETRY(100), TIMEOUT(99), LOG('Call abc'));
+
+
+output(SOAPCALL(ds, 'http://webservices.megacorp.com', 'WsAttributesDs', GetAttributeInRecord, createInRecord2(LEFT.moduleName, LEFT.attr), dataset(GetAttributeOutRecord), LOG('megacorp:'+LEFT.moduleName+'.'+LEFT.attr)));
+
+output(SOAPCALL(ds, 'http://webservices.megacorp.com', 'WsAttributesDs', GetAttributeInRecord, createInRecord2(LEFT.moduleName, LEFT.attr), dataset(GetAttributeOutRecord), LOG,LOG(MIN),LOG('megacorp:'+moduleName+'.'+attr)));
+
+
+output(HTTPCALL('http://webservices.megacorp.com', 'WsAttributesDs', 'blah', GetAttributeOutRecord, LOG,LOG(MIN),LOG('megacorp:xxx')));

initfiles/sbin/configmgr.in

@@ -91,7 +91,6 @@ if [ ${DEBUG:-NO_DEBUG} != "NO_DEBUG" ]; then
 fi
 
 exec_script_path=${path}/bin
-configgenPath="${path}/sbin/configgen"
 reg_path=${path}/sbin
 
 compName=configmgr
@@ -170,19 +169,6 @@ fi
 createConf "${filename}" "${portnum}" "${conffile}"
 chown ${user}:${group} ${conffile}
 
-# Configgen checking step
-echo -n "Validating environment file ${filename} using configgen ..."
-sleep 1
-cmd="${configgenPath} -env ${filename} -validateonly"
-eval $cmd
-if [ "$?" -ne 0 ]; then
-    echo " Failure"
-    exit
-else
-    echo " Success"
-fi
-
-
 # Sanity Check for previous instances of configmgr. This code also takes care if configmgr script/configesp is not running and it killed by kill command
 
 check_status ${initPidFile} ${lockFile} ${compPidFile} 0

roxie/ccd/ccdactivities.cpp

@@ -2408,18 +2408,14 @@ public:
                 }
                 row = m.readDirect(len);
                 if (!finalBuilder.exists())
-                {
-                    size32_t clonedSize;
-                    void * cloned = cloneRow(rowAllocator, row, clonedSize);
-                    finalBuilder.setown(clonedSize, cloned);
-                }
+                    cloneRow(finalBuilder, row, meta);
                 else
                     helper->mergeAggregate(finalBuilder, row);
             }
         }
         if (finalBuilder.exists())
         {
-            size32_t finalSize = meta.getRecordSize(finalBuilder.getSelf());
+            size32_t finalSize = meta.getRecordSize(finalBuilder.getSelf());  // MORE - can probably track it above...
             finalRow.setown(finalBuilder.finalizeRowClear(finalSize));
         }
     }

roxie/ccd/ccdserver.cpp

@@ -16422,7 +16422,7 @@ class CRoxieServerSelfJoinActivity : public CRoxieServerActivity
     Owned<IEngineRowAllocator> defaultAllocator;
     Owned<IRHLimitedCompareHelper> limitedhelper;
     Owned<CRHDualCache> dualcache;
-    IInputBase *input; // Hiding the one in the base class - note this is an IInputBase not an IRoxieInput
+    IInputBase *dualCacheInput;
 
     bool fillGroup()
     {
@@ -16590,6 +16590,7 @@ public:
         leftIndex = 0;
         rightIndex = 0;
         rightOuterIndex = 0;
+        dualCacheInput = NULL;
     }
 
     virtual void start(unsigned parentExtractSize, const byte *parentExtract, bool paused)
@@ -16617,7 +16618,7 @@ public:
         {   //limited match join (s[1..n])
             dualcache.setown(new CRHDualCache());
             dualcache->init(CRoxieServerActivity::input);
-            input = dualcache->queryOut1();
+            dualCacheInput = dualcache->queryOut1();
             failingOuterAtmost = false;
             matchedLeft = false;
             leftIndex = 0;
@@ -16626,8 +16627,6 @@ public:
             limitedhelper.setown(createRHLimitedCompareHelper());
             limitedhelper->init( helper.getJoinLimit(), dualcache->queryOut2(), collate, helper.queryPrefixCompare() );
         }
-        else
-            input = CRoxieServerActivity::input;
     }
 
     virtual void reset()
@@ -16647,7 +16646,7 @@ public:
             {
                 if (!group.isItem(rightIndex))
                 {
-                    lhs.setown(input->nextInGroup());   //get from dualcache
+                    lhs.setown(dualCacheInput->nextInGroup());
                     if (lhs)
                     {
                         rightIndex = 0;
@@ -16657,7 +16656,6 @@ public:
                     else 
                     {
                         eof = true;
-                        input = (IRoxieInput*)dualcache->input();
                     }
                 }
                 
@@ -16673,104 +16671,107 @@ public:
             }
             return NULL;
         }
-        if (first)
-        {
-            first = false;
-            fillGroup();
-        }
-        while(!eof)
+        else
         {
-            if(failingOuterAtmost)
-                while(group.isItem(leftIndex))
-                {
-                    const void * ret = joinRecords(group.item(leftIndex++), defaultRight);
-                    if(ret)
-                    {
-                        processed++;
-                        return ret;
-                    }
-                }
-            if((joinLimit == 0) || !group.isItem(rightIndex))
+            if (first)
             {
-                if(leftOuterJoin && !matchedLeft && !failingLimit)
-                {
-                    const void * ret = joinRecords(group.item(leftIndex), defaultRight);
-                    if(ret)
-                    {
-                        matchedLeft = true;
-                        processed++;
-                        return ret;
-                    }
-                }
-                leftIndex++;
-                matchedLeft = false;
-                rightIndex = 0;
-                joinLimit = keepLimit;
+                first = false;
+                fillGroup();
             }
-            if(!group.isItem(leftIndex))
+            while(!eof)
             {
-                if(failingLimit || failingOuterAtmost)
+                if(failingOuterAtmost)
+                    while(group.isItem(leftIndex))
+                    {
+                        const void * ret = joinRecords(group.item(leftIndex++), defaultRight);
+                        if(ret)
+                        {
+                            processed++;
+                            return ret;
+                        }
+                    }
+                if((joinLimit == 0) || !group.isItem(rightIndex))
                 {
-                    const void * lhs;
-                    while((lhs = input->nextInGroup()) != NULL)
+                    if(leftOuterJoin && !matchedLeft && !failingLimit)
                     {
-                        const void * ret = joinRecords(lhs, defaultRight, failingLimit);
-                        ReleaseRoxieRow(lhs);
+                        const void * ret = joinRecords(group.item(leftIndex), defaultRight);
                         if(ret)
                         {
+                            matchedLeft = true;
                             processed++;
                             return ret;
                         }
                     }
-                    failingLimit.clear();
+                    leftIndex++;
+                    matchedLeft = false;
+                    rightIndex = 0;
+                    joinLimit = keepLimit;
                 }
-                if(rightOuterJoin && !failingLimit)
-                    while(group.isItem(rightOuterIndex))
-                        if(!matchedRight.item(rightOuterIndex++))
+                if(!group.isItem(leftIndex))
+                {
+                    if(failingLimit || failingOuterAtmost)
+                    {
+                        const void * lhs;
+                        while((lhs = input->nextInGroup()) != NULL)  // dualCache never active here
                         {
-                            const void * ret = joinRecords(defaultLeft, group.item(rightOuterIndex-1));
+                            const void * ret = joinRecords(lhs, defaultRight, failingLimit);
+                            ReleaseRoxieRow(lhs);
                             if(ret)
                             {
                                 processed++;
                                 return ret;
                             }
                         }
-                if(!fillGroup())
-                    return NULL;
-                continue;
-            }
-            const void * lhs = group.item(leftIndex);
-            if(failingLimit)
-            {
-                leftIndex++;
-                const void * ret = joinRecords(lhs, defaultRight, failingLimit);
-                if(ret)
+                        failingLimit.clear();
+                    }
+                    if(rightOuterJoin && !failingLimit)
+                        while(group.isItem(rightOuterIndex))
+                            if(!matchedRight.item(rightOuterIndex++))
+                            {
+                                const void * ret = joinRecords(defaultLeft, group.item(rightOuterIndex-1));
+                                if(ret)
+                                {
+                                    processed++;
+                                    return ret;
+                                }
+                            }
+                    if(!fillGroup())
+                        return NULL;
+                    continue;
+                }
+                const void * lhs = group.item(leftIndex);
+                if(failingLimit)
                 {
-                    processed++;
-                    return ret;
+                    leftIndex++;
+                    const void * ret = joinRecords(lhs, defaultRight, failingLimit);
+                    if(ret)
+                    {
+                        processed++;
+                        return ret;
+                    }
                 }
-            }
-            else
-            {
-                const void * rhs = group.item(rightIndex++);
-                if(helper.match(lhs, rhs))
+                else
                 {
-                    matchedLeft = true;
-                    matchedRight.replace(true, rightIndex-1);
-                    if(!exclude)
+                    const void * rhs = group.item(rightIndex++);
+                    if(helper.match(lhs, rhs))
                     {
-                        const void * ret = joinRecords(lhs, rhs);
-                        if(ret)
+                        matchedLeft = true;
+                        matchedRight.replace(true, rightIndex-1);
+                        if(!exclude)
                         {
-                            processed++;
-                            joinLimit--;
-                            return ret;
+                            const void * ret = joinRecords(lhs, rhs);
+                            if(ret)
+                            {
+                                processed++;
+                                joinLimit--;
+                                return ret;
+                            }
                         }
                     }
                 }
             }
+            return NULL;
         }
-        return NULL;
     }
 };
 
@@ -20335,9 +20336,7 @@ public:
             else
             {
                  // NOTE need to clone this because going to modify below, could special case 1 row only
-                void * cloned = cloneRow(rowAllocator, firstRow, finalSize);
-                //NB: finalSize is updated by the call above, so don't combine these statements
-                rowBuilder.setown(finalSize, cloned);
+                finalSize = cloneRow(rowBuilder, firstRow, meta);
                 ReleaseRoxieRow(firstRow);
             }
             loop
@@ -22072,9 +22071,7 @@ public:
         else
         {
              // NOTE need to clone this because going to modify below, could special case 1 row only
-            void * cloned = cloneRow(rowAllocator, firstRow, finalSize);
-            //NB: finalSize is updated by the call above, so don't combine these statements
-            rowBuilder.setown(finalSize, cloned);
+            finalSize = cloneRow(rowBuilder, firstRow, meta);
             ReleaseRoxieRow(firstRow);
         }
         loop

rtl/include/eclhelper.hpp

@@ -2050,6 +2050,8 @@ enum
     SOAPFnamespace      = 0x0020,
     SOAPFencoding       = 0x0040,
     SOAPFpreserveSpace  = 0x0080,
+    SOAPFlogmin         = 0x0100,
+    SOAPFlogusermsg     = 0x0200,
 };
 
 struct IHThorWebServiceCallActionArg : public IHThorArg
@@ -2086,6 +2088,7 @@ struct IHThorWebServiceCallExtra : public IInterface
     virtual IXmlToRowTransformer * queryInputTransformer() = 0;
     virtual const char * queryInputIteratorPath()       { return NULL; }
     virtual size32_t onFailTransform(ARowBuilder & rowBuilder, const void * left, IException * e) { return 0; }
+    virtual void getLogText(size32_t & lenText, char * & text, const void * left) = 0;  // iff SOAPFlogusermsg set
 };
 typedef IHThorWebServiceCallExtra IHThorSoapCallExtra;
 

rtl/include/eclhelper_base.hpp

@@ -2361,6 +2361,7 @@ class CThorSoapActionArg : public CThorArg, implements IHThorSoapActionArg
     virtual unsigned     getTimeLimit()                 { return 0; }
     virtual const char * queryProxyAddress()            { return NULL; }
     virtual const char * queryAcceptType()              { return NULL; }
+    virtual void getLogText(size32_t & lenText, char * & text, const void * left) { lenText =0; text = NULL; }
 };
 
 class CThorSoapCallArg : public CThorArg, implements IHThorSoapCallArg
@@ -2406,6 +2407,7 @@ class CThorSoapCallArg : public CThorArg, implements IHThorSoapCallArg
     virtual unsigned     getTimeLimit()                 { return 0; }
     virtual const char * queryProxyAddress()            { return NULL; }
     virtual const char * queryAcceptType()              { return NULL; }
+    virtual void getLogText(size32_t & lenText, char * & text, const void * left) { lenText =0; text = NULL; }
 };
 typedef CThorSoapCallArg CThorHttpCallArg;
 

testing/ecl/roxie/soapcall.ecl

@@ -28,7 +28,7 @@ ServiceOutRecord :=
     END;
 
 // simple query->dataset form
-output(SORT(SOAPCALL('http://127.0.0.1:9876','soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord)),record));
+output(SORT(SOAPCALL('http://127.0.0.1:9876','soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord), LOG('simple')),record));
 
 // double query->dataset form
 output(SORT(SOAPCALL('http://127.0.0.1:9876|http://127.0.0.1:9876','soapbase', { string unkname := 'FRED' }, dataset(ServiceOutRecord)),record));
@@ -53,9 +53,9 @@ END;
 
 // Test some failure cases
 
-output(SORT(SOAPCALL(d, 'http://127.0.0.1:9876|http://127.0.0.1:9875','soapbase', { unkname }, DATASET(ServiceOutRecord), onFail(doError(LEFT)),RETRY(0)), record));
-output(SORT(SOAPCALL('http://127.0.0.1:9876','soapbase', { string unkname := 'FAIL' }, dataset(ServiceOutRecord),onFail(doError2),RETRY(0)),record));
-output(SORT(SOAPCALL(d, 'http://127.0.0.1:9876','soapbaseNOSUCHQUERY', { unkname }, DATASET(ServiceOutRecord), onFail(doError(LEFT)),MERGE(25),RETRY(0)), record));
+output(SORT(SOAPCALL(d, 'http://127.0.0.1:9876|http://127.0.0.1:9875','soapbase', { unkname }, DATASET(ServiceOutRecord), onFail(doError(LEFT)),RETRY(0), log('SOAP: ' + unkname)), record));
+output(SORT(SOAPCALL('http://127.0.0.1:9876','soapbase', { string unkname := 'FAIL' }, dataset(ServiceOutRecord),onFail(doError2),RETRY(0), LOG(MIN)),record));
+output(SORT(SOAPCALL(d, 'http://127.0.0.1:9876','soapbaseNOSUCHQUERY', { unkname }, DATASET(ServiceOutRecord), onFail(doError(LEFT)),MERGE(25),RETRY(0), LOG(MIN)), record));
 
 childRecord := record
 unsigned            id;

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -462,6 +462,8 @@ public:
     }
     virtual void stop()
     {
+        if (!gotRHS)
+            getRHS(true);
         rhs.reset(false);
         stopRightInput();
         stopInput(left);
@@ -594,10 +596,7 @@ public:
     {
         ActivityTimer t(totalCycles, timeActivities, NULL);
         if (!gotRHS)
-        {
-            getRHS();
-            gotRHS = true;
-        }
+            getRHS(false);
         if (!abortSoon && !eos)
         {
             if (doRightOuter)
@@ -838,13 +837,15 @@ public:
         info.unknownRowsOutput = true;
         info.canStall = true;
     }
-    void sendToBroadcastSlave()
+    bool sendToBroadcastSlave(bool stopping)
     {
 #ifdef _TRACEBROADCAST
         ActPrintLog("%s: sendToBroadcastSlave sending to slave %d", joinStr.get(), broadcastSlave);
 #endif
+        bool allRequestStop = false;
         try
         {
+            bool sentStop = false;
             MemoryBuffer mb;
             CMemoryRowSerializer mbs(mb);
 
@@ -860,19 +861,30 @@ public:
                 }
                 CMessageBuffer msg;
                 if (!receiveMsg(msg, broadcastSlave, mpTag))
-                    return;
-                if (0 != mb.length())
+                    return false;
+                msg.read(allRequestStop);
+                msg.clear();
+                if (!allRequestStop && 0 != mb.length())
                 {
-                    ThorCompress(mb.toByteArray(), mb.length(), msg.clear());
+                    msg.append(stopping);
+                    ThorCompress(mb.toByteArray(), mb.length(), msg);
 #ifdef _TRACEBROADCAST
                     ActPrintLog("sendToBroadcastSlave Compressing buf from %d to %d",mb.length(),msg.length());
 #endif
 #ifdef _TRACEBROADCAST
                     ActPrintLog("sendToBroadcastSlave sending reply to %d on tag %d",(int)broadcastSlave,(int)mpTag);
+                    if (stopping)
+                        sentStop = true;
 #endif
                 }
+                else
+                {
+                    // prevent stop at this point unless have already sent stop,
+                    // to prevent allRequestStop at broadcaster, before this slave knows about it.
+                    msg.append(sentStop && stopping);
+                }
                 if (!container.queryJob().queryJobComm().reply(msg))
-                    return;
+                    return false;
                 if (0 == mb.length())
                     break;
                 mb.clear();
@@ -886,6 +898,7 @@ public:
             ActPrintLog(e, "CLookupJoinActivity::sendToBroadcastSlave: exception");
             throw;
         }
+        return !allRequestStop;
     }
     void processRows(MemoryBuffer &mb)
     {
@@ -913,20 +926,29 @@ public:
         stopRightInput();
 #endif
     }
-    void getRHS()
+    void getRHS(bool stopping)
     {
+        if (gotRHS)
+            return;
+        gotRHS = true;
         Owned<IException> exception;
         try
         {
-            gatherLocal();
             if (!container.queryLocal() && container.queryJob().querySlaves() > 1)
             {
+                bool allRequestStop = false;
+                gatherLocal();
                 broadcaster.init(container.queryJob().queryMyRank(), container.queryJob().querySlaves(), &container.queryJob().queryJobComm(), mpTag, broadcastSlave);
                 if (container.queryJob().queryMyRank()==broadcastSlave)
                 {
                     unsigned fromNode = 1;
                     Owned<IBitSet> slavesDone = createBitSet();
-
+                    Owned<IBitSet> slavesStopping = createBitSet();
+                    slavesDone->testSet(broadcastSlave-1, true);
+                    slavesStopping->testSet(broadcastSlave-1, true);
+                    // loop, requesting data from all other slaves (in chunks)
+                    // track slaves which have finished sending in slavesDone (signalled via 0 len. packet)
+                    // NB: collates from slaves serially, probably should be in parallel
                     MemoryBuffer tmp;
                     CMessageBuffer msg;
                     bool allDone = false;
@@ -936,19 +958,28 @@ public:
                         ActPrintLog("getRHS Receiving");
 #endif
                         if (fromNode == broadcastSlave)
-                        {
-                            slavesDone->testSet(broadcastSlave-1, true);
                             ++fromNode;
-                        }
-                        {
+
+                        { // request more
                             BooleanOnOff onOff(receiving);
+                            msg.append(allRequestStop);
                             container.queryJob().queryJobComm().sendRecv(msg, fromNode, mpTag);
                         }
 #ifdef _TRACEBROADCAST
                         ActPrintLog("getRHS got %d from %d",msg.length(),(int)fromNode);
 #endif
-                        if (0 == msg.length())
+                        bool slaveStopRequest; // only true if slave stopping without needing RHS
+                        msg.read(slaveStopRequest);
+                        slavesStopping->testSet(fromNode-1, slaveStopRequest);
+                        if (stopping)
+                        {
+                            // can only stop if I'm stopping and all other slaves are
+                            allRequestStop = slavesStopping->scan(0, false) == container.queryJob().querySlaves();
+                            // if true, next request will tell slaves to stop sending
+                        }
+                        if (0 == msg.remaining()) // slave signalled no more data
                         {
+                            msg.clear();
                             bool done = slavesDone->testSet(fromNode-1, true);
                             assertex(false == done);
                             if (slavesDone->scan(0, false) == container.queryJob().querySlaves()) // i.e. got all
@@ -957,73 +988,95 @@ public:
                         }
                         else
                         {
-                            ThorExpand(msg, tmp);
+                            if (allRequestStop)
+                            {
+                                // only here, if all stopping, 1st packet from slave and signalled stopping
+                                msg.clear(); // no longer wanted
+                            }
+                            else
+                            {
+                                ThorExpand(msg, tmp);
 #ifdef _TRACEBROADCAST
-                            ActPrintLog("getRHS expanding.1 %d to %d",msg.length(), tmp.length());
+                                ActPrintLog("getRHS expanding.1 %d to %d",msg.length(), tmp.length());
 #endif
-                            msg.clear();
-                            processRows(tmp);
-                            tmp.clear();
+                                msg.clear();
+                                processRows(tmp);
+                                tmp.clear();
+                            }
                         }
                         if (allDone)
                             break;
                     }
-
-                    // now all (global) rows in this (broadcast node) rhs HT.
-                    CMemoryRowSerializer mbs(tmp.clear());
-                    allDone = false;
-                    unsigned r=0;
-                    while (!abortSoon)
+                    if (!allRequestStop)
                     {
-                        loop
+                        // now all (global) RHS rows on this (broadcast) node
+                        CMemoryRowSerializer mbs(tmp.clear());
+                        allDone = false;
+                        unsigned r=0;
+                        while (!abortSoon)
                         {
-                            if (r == rhs.ordinality())
+                            loop
                             {
-                                allDone = true;
-                                break;
+                                if (r == rhs.ordinality())
+                                {
+                                    allDone = true;
+                                    break;
+                                }
+                                const byte *row = (const byte *)rhs.item(r++);
+                                rightSerializer->serialize(mbs, row);
+                                if (tmp.length() > 0x80000)
+                                    break;
                             }
-                            const byte *row = (const byte *)rhs.item(r++);
-                            rightSerializer->serialize(mbs, row);
-                            if (tmp.length() > 0x80000)
-                                break;
-                        }
-                        if (0 != tmp.length())
-                        {
-                            ThorCompress(tmp, msg);
+                            if (0 != tmp.length())
+                            {
+                                ThorCompress(tmp, msg);
 #ifdef _TRACEBROADCAST
-                            ActPrintLog("getRHS compress.1 %d to %d",tmp.length(), msg.length());
+                                ActPrintLog("getRHS compress.1 %d to %d",tmp.length(), msg.length());
 #endif
-                            tmp.clear();
+                                tmp.clear();
 
-                            broadcaster.broadcast(msg);
-                            msg.clear();
+                                broadcaster.broadcast(msg);
+                                msg.clear();
+                            }
+                            if (allDone)
+                                break;
                         }
-                        if (allDone)
-                            break;
                     }
                 }
                 else
                 {
-                    sendToBroadcastSlave();
-                    rhs.clear();
-                    MemoryBuffer buf;
-                    MemoryBuffer expBuf;
-                    while (broadcaster.receive(buf))
+                    if (!sendToBroadcastSlave(stopping))
+                        allRequestStop = true;
+                    else
                     {
-                        ThorExpand(buf, expBuf);
+                        rhs.clear();
+                        MemoryBuffer buf;
+                        MemoryBuffer expBuf;
+                        while (broadcaster.receive(buf))
+                        {
+                            ThorExpand(buf, expBuf);
 #ifdef _TRACEBROADCAST
-                        ActPrintLog("Expanding received buf from %d to %d",buf.length(),expBuf.length());
+                            ActPrintLog("Expanding received buf from %d to %d",buf.length(),expBuf.length());
 #endif
-                        processRows(expBuf);
-                        expBuf.clear();
-                        broadcaster.broadcast(buf); // will swap buf
-                        buf.clear();
+                            processRows(expBuf);
+                            expBuf.clear();
+                            broadcaster.broadcast(buf); // will swap buf
+                            buf.clear();
+                        }
                     }
                 }
-                broadcaster.endBroadcast();     // send final
-                broadcaster.clear();
+                if (!allRequestStop)
+                {
+                    broadcaster.endBroadcast();     // send final
+                    broadcaster.clear();
+                    prepareRHS();
+                }
+            }
+            else if (!stopping)
+            {   // single node or local
+                gatherLocal();
+                prepareRHS();
             }
-            prepareRHS();
         }
         catch (IOutOfMemException *e) { exception.setown(e); }
         catch (IThorRowArrayException *e) { exception.setown(e); }

thorlcr/thorutil/thcompressutil.cpp

@@ -87,7 +87,9 @@ size32_t ThorExpand(const void * src, size32_t srcSz, MemoryBuffer & dest)
 
 size32_t ThorExpand(MemoryBuffer & src, MemoryBuffer & dest)
 {
-    return ThorExpand((const void *)src.toByteArray(), src.length(), dest);
+    size32_t len = src.remaining();
+    const void *pSrc = src.readDirect(len);
+    return ThorExpand(pSrc, len, dest);
 }